AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleanup and edge case handling for peer locking bits. The specific problem

Browse Source 
we were seeing in Spiral Knights was something like peer A created a handler
for B's acquisition of a lock, but then C would hijack the acquisition and
A would be left waiting indefinitely for B's acquisition.  Now the handler
can detect the hijacking and notify its listeners.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6671 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Andrzej Kapolka
2011-06-29 20:38:03 +00:00
parent 155c281b18
commit de6d2aac57
2 changed files with 156 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/threerings/presents/peer/server/PeerManager.java
+139 -66
View File
@@ -573,7 +573,7 @@ public abstract class PeerManager
public <T extends DObject> void proxyRemoteObject (
final DObjectAddress remote, final ResultListener<Integer> listener)
{
if (remote.nodeName.equals(_nodeName)) {
if (remote.nodeName.equals(_nodeName)) {
_omgr.postRunnable(new Runnable() {
public void run () {
listener.requestCompleted(remote.oid);
@@ -1282,21 +1282,7 @@ public abstract class PeerManager
*/
protected void connectedToPeer (PeerNode peer)
{
// check for lock conflicts
String nodeName = peer.getNodeName();
for (NodeObject.Lock lock : peer.nodeobj.locks) {
PeerManager.LockHandler handler = _locks.get(lock);
if (handler != null) {
log.warning("Client hijacked lock in process of resolution",
"handler", handler, "node", nodeName);
handler.clientHijackedLock(nodeName);
} else if (_nodeobj.locks.contains(lock)) {
log.warning("Client hijacked lock owned by this node",
"lock", lock, "node", nodeName);
droppedLock(lock);
}
}
// nothing by default
}
/**
@@ -1304,12 +1290,104 @@ public abstract class PeerManager
*/
protected void disconnectedFromPeer (PeerNode peer)
{
// clear any locks held by that peer
String nodeName = peer.getNodeName();
for (LockHandler handler : _locks.values().toArray(new LockHandler[_locks.size()])) {
if (handler.getNodeName().equals(nodeName)) {
handler.clientDidLogoff();
}
// nothing by default
}
/**
* Called when a peer announces its intention to acquire a lock.
*/
protected void peerAcquiringLock (PeerNode peer, NodeObject.Lock lock)
{
// refuse to ratify if we believe someone else owns the lock
String owner = queryLock(lock);
if (owner != null) {
log.warning("Refusing to ratify lock acquisition.", "lock", lock,
"node", peer.getNodeName(), "owner", owner);
return;
}
// check for an existing handler
LockHandler handler = _locks.get(lock);
if (handler == null) {
createLockHandler(peer, lock, true);
return;
}
// if the existing node has priority, we're done
if (hasPriority(handler.getNodeName(), peer.getNodeName())) {
return;
}
// the new node has priority, so cancel the existing handler and take over
// its listeners
ResultListenerList<String> olisteners = handler.listeners;
handler.cancel();
handler = createLockHandler(peer, lock, true);
handler.listeners = olisteners;
}
/**
* Called when a peer announces its intention to release a lock.
*/
protected void peerReleasingLock (PeerNode peer, NodeObject.Lock lock)
{
// refuse to ratify if we don't believe they own the lock
String owner = queryLock(lock);
if (!peer.getNodeName().equals(owner)) {
log.warning("Refusing to ratify lock release.", "lock", lock,
"node", peer.getNodeName(), "owner", owner);
return;
}
// check for an existing handler
LockHandler handler = _locks.get(lock);
if (handler == null) {
createLockHandler(peer, lock, false);
} else {
log.warning("Received request to release resolving lock",
"node", peer.getNodeName(), "handler", handler);
}
}
/**
* Called when a peer adds a lock.
*/
protected void peerAddedLock (String nodeName, NodeObject.Lock lock)
{
// check for hijacking
if (_nodeobj.locks.contains(lock)) {
log.warning("Client hijacked lock owned by this node", "lock", lock, "node", nodeName);
droppedLock(lock);
}
// notify the handler, if any
LockHandler handler = _locks.get(lock);
if (handler != null) {
handler.peerAddedLock(nodeName);
}
}
/**
* Called when a peer updates a lock.
*/
protected void peerUpdatedLock (String nodeName, NodeObject.Lock lock)
{
// notify the handler, if any
LockHandler handler = _locks.get(lock);
if (handler != null) {
handler.peerUpdatedLock(nodeName);
}
}
/**
* Called when a peer removes a lock.
*/
protected void peerRemovedLock (String nodeName, NodeObject.Lock lock)
{
// notify the handler, if any
LockHandler handler = _locks.get(lock);
if (handler != null) {
handler.peerRemovedLock(nodeName);
}
}
@@ -1380,11 +1458,19 @@ public abstract class PeerManager
return DEFAULT_LOCK_TIMEOUT;
}
/**
* 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;
}
/**
* Handles a lock in a state of resolution.
*/
protected class LockHandler
implements SetListener<NodeObject.Lock>
{
/** Listeners waiting for resolution. */
public ResultListenerList<String> listeners = new ResultListenerList<String>();
@@ -1431,9 +1517,6 @@ public abstract class PeerManager
// ratify the action
peer.nodeobj.peerService.ratifyLockAction(lock, acquire);
// listen for the act to take place
peer.nodeobj.addListener(this);
}
/**
@@ -1480,23 +1563,43 @@ public abstract class PeerManager
}
/**
* Called when the connection to the controlling node has been broken.
* Called to notify us that a node has added the lock.
*/
public void clientDidLogoff ()
public void peerAddedLock (String nodeName)
{
_locks.remove(_lock);
listeners.requestCompleted(null);
if (!(_acquire && getNodeName().equals(nodeName))) {
log.warning("Node hijacked lock in process of resolution.",
"node", nodeName, "handler", this);
_stats.locksHijacked++;
}
cancel();
wasActivated(nodeName);
}
/**
* Called when a client hijacks the lock by having it in its node object when it connects.
* Called to notify us that a node has updated the lock.
*/
public void clientHijackedLock (String nodeName)
public void peerUpdatedLock (String nodeName)
{
// updating is a signal of reacquisition
if (!(!_acquire && getNodeName().equals(nodeName))) {
log.warning("Unexpected lock update.", "node", nodeName, "handler", this);
_stats.locksHijacked++;
}
cancel();
_locks.remove(_lock);
_stats.locksHijacked++;
listeners.requestCompleted(nodeName);
wasActivated(nodeName);
}
/**
* Called to notify us that a node has removed the lock.
*/
public void peerRemovedLock (String nodeName)
{
if (getNodeName().equals(nodeName)) {
wasActivated(null);
} else {
log.warning("Unexpected lock removal.", "node", nodeName, "handler", this);
}
}
/**
@@ -1504,40 +1607,11 @@ public abstract class PeerManager
*/
public void cancel ()
{
if (_peer != null) {
_peer.nodeobj.removeListener(this);
} else {
if (_peer == null) {
_timeout.cancel();
}
}
// documentation inherited from interface SetListener
public void entryAdded (EntryAddedEvent<NodeObject.Lock> event)
{
if (_acquire && event.getName().equals(NodeObject.LOCKS) &&
event.getEntry().equals(_lock)) {
wasActivated(_peer.getNodeName());
}
}
// documentation inherited from interface SetListener
public void entryRemoved (EntryRemovedEvent<NodeObject.Lock> event)
{
if (!_acquire && event.getName().equals(NodeObject.LOCKS) &&
event.getOldEntry().equals(_lock)) {
wasActivated(null);
}
}
// documentation inherited from interface SetListener
public void entryUpdated (EntryUpdatedEvent<NodeObject.Lock> event)
{
if (!_acquire && event.getName().equals(NodeObject.LOCKS) &&
event.getEntry().equals(_lock)) {
wasActivated(_peer.getNodeName());
}
}
@Override
public String toString ()
{
@@ -1573,7 +1647,6 @@ public abstract class PeerManager
*/
protected void wasActivated (String owner)
{
_peer.nodeobj.removeListener(this);
_locks.remove(_lock);
listeners.requestCompleted(owner);
}
src/main/java/com/threerings/presents/peer/server/PeerNode.java
+17 -43
View File
@@ -27,8 +27,6 @@ import java.util.Date;
import com.google.inject.Inject;
import com.samskivert.util.ResultListenerList;
import com.threerings.presents.client.Client;
import com.threerings.presents.client.ClientObserver;
import com.threerings.presents.client.Communicator;
@@ -220,6 +218,9 @@ public class PeerNode
for (ClientInfo clinfo : nodeobj.clients) {
_peermgr.clientLoggedOff(nodeName, clinfo);
}
for (NodeObject.Lock lock : nodeobj.locks) {
_peermgr.peerRemovedLock(nodeName, lock);
}
nodeobj.removeListener(_listener);
@@ -247,6 +248,9 @@ public class PeerNode
for (ClientInfo clinfo : nodeobj.clients) {
_peermgr.clientLoggedOn(nodeName, clinfo);
}
for (NodeObject.Lock lock : nodeobj.locks) {
_peermgr.peerAddedLock(nodeName, lock);
}
}
// documentation inherited from interface Subscriber
@@ -255,15 +259,6 @@ public class PeerNode
log.warning("Failed to subscribe to peer's node object", "peer", _record, "cause", cause);
}
/**
* 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;
}
protected Communicator createCommunicator (Client client)
{
return new ServerCommunicator(client, _conmgr, _omgr);
@@ -287,39 +282,10 @@ public class PeerNode
public void attributeChanged (AttributeChangedEvent event) {
String name = event.getName();
if (name.equals(NodeObject.ACQUIRING_LOCK)) {
NodeObject.Lock lock = nodeobj.acquiringLock;
PeerManager.LockHandler handler = _peermgr.getLockHandler(lock);
if (handler == null) {
if (_peermgr.getNodeObject().locks.contains(lock)) {
log.warning("Peer trying to acquire lock owned by this node", "lock", lock,
"node", _record.nodeName);
return;
}
_peermgr.createLockHandler(PeerNode.this, lock, true);
return;
}
// if the other node has priority, we're done
if (hasPriority(handler.getNodeName(), _record.nodeName)) {
return;
}
// this node has priority, so cancel the existing handler and take over
// its listeners
ResultListenerList<String> olisteners = handler.listeners;
handler.cancel();
handler = _peermgr.createLockHandler(PeerNode.this, lock, true);
handler.listeners = olisteners;
_peermgr.peerAcquiringLock(PeerNode.this, (NodeObject.Lock)event.getValue());
} else if (name.equals(NodeObject.RELEASING_LOCK)) {
NodeObject.Lock lock = nodeobj.releasingLock;
PeerManager.LockHandler handler = _peermgr.getLockHandler(lock);
if (handler == null) {
_peermgr.createLockHandler(PeerNode.this, lock, false);
} else {
log.warning("Received request to release resolving lock",
"node", _record.nodeName, "handler", handler);
}
_peermgr.peerReleasingLock(PeerNode.this, (NodeObject.Lock)event.getValue());
} else if (name.equals(NodeObject.CACHE_DATA)) {
_peermgr.changedCacheData(nodeobj.cacheData.cache, nodeobj.cacheData.data);
@@ -331,12 +297,17 @@ public class PeerNode
String name = event.getName();
if (NodeObject.CLIENTS.equals(name)) {
_peermgr.clientLoggedOn(getNodeName(), (ClientInfo)event.getEntry());
} else if (NodeObject.LOCKS.equals(name)) {
_peermgr.peerAddedLock(getNodeName(), (NodeObject.Lock)event.getEntry());
}
}
// documentation inherited from interface SetListener
public void entryUpdated (EntryUpdatedEvent<DSet.Entry> event) {
// nada
if (NodeObject.LOCKS.equals(event.getName())) {
_peermgr.peerUpdatedLock(getNodeName(), (NodeObject.Lock)event.getEntry());
}
}
// documentation inherited from interface SetListener
@@ -344,6 +315,9 @@ public class PeerNode
String name = event.getName();
if (NodeObject.CLIENTS.equals(name)) {
_peermgr.clientLoggedOff(getNodeName(), (ClientInfo)event.getOldEntry());
} else if (NodeObject.LOCKS.equals(name)) {
_peermgr.peerRemovedLock(getNodeName(), (NodeObject.Lock)event.getOldEntry());
}
}
} // END: class NodeObjectListener