anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix for CTrie TNode exception and other improvements to CTrie related

Browse Source 
subscribe and unsubscribe issues

-  Added unit tests to cover a few more cases which I my system ran into
   under concurrent load testing.
-  Remove and detect TNode, implement more of
   http://bravenewgeek.com/breaking-and-entering-lose-the-lock-while-embracing-concurrency/
   related to TNode state
-  NOTE:  The graph is left with nodes which have no subscriptions, the
   TNode cleanup only cleanss the last node for now.  We should
   autoprune further up by placing more TNodes and pruning them off.
-  Adding concurrency Sets and Lists to CNode, because those ARE edited
   and read in parallel in our use-case.  Before a CNode is CAS
   replaced, more than 1 thread may be attempting to modify and read the
   subscriptions and children CNode datastructures, and I hit this
   during testing.
-  Fixed an issue where we unsubscribe all sub-topics if an unsubscribe
   is sent on a parent node.
-  Fixed an issue with unsubscribing multiple times when no subscrition
   exists, we should handle this gracefully.
This commit is contained in:
Marcus Williford
2017-10-24 09:45:12 -07:00
parent 3293478dd7
commit 1bc6bdbbce
4 changed files with 233 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java
+9 -7
View File
@@ -24,9 +24,7 @@ class CNode {
Token token;
private List<INode> children;
Set<Subscription> subscriptions = new HashSet<>();
//private int subtreeSubscriptions;
Set<Subscription> subscriptions;
CNode() {
this.children = new ArrayList<>();
@@ -35,9 +33,9 @@ class CNode {
//Copy constructor
private CNode(Token token, List<INode> children, Set<Subscription> subscriptions) {
this.token = token;
this.children = children;
this.subscriptions = subscriptions;
this.token = token; // keep reference, root comparison in directory logic relies on it for now.
this.subscriptions = new HashSet<>(subscriptions);
this.children = new ArrayList<>(children);
}
boolean anyChildrenMatch(Token token) {
@@ -75,6 +73,9 @@ class CNode {
public void add(INode newINode) {
this.children.add(newINode);
}
public void remove(INode node) {
this.children.remove(node);
}
CNode addSubscription(String clientId, Topic topic) {
this.subscriptions.add(new Subscription(clientId, topic));
@@ -83,6 +84,7 @@ class CNode {
/**
* @return true iff the subscriptions contained in this node are owned by clientId
* AND at least one subscription is actually present for that clientId
* */
boolean containsOnly(String clientId) {
for (Subscription sub : this.subscriptions) {
@@ -90,7 +92,7 @@ class CNode {
return false;
}
}
return true;
return !this.subscriptions.isEmpty();
}
//TODO this is equivalent to negate(containsOnly(clientId))
broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java
+48 -7
View File
@@ -31,6 +31,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
private static final Logger LOG = LoggerFactory.getLogger(CTrieSubscriptionDirectory.class);
private static final Token ROOT = new Token("root");
private static final INode NO_PARENT = null;
INode root;
private volatile SessionsRepository sessionsRepository;
@@ -49,10 +50,13 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
@Override
public void visit(CNode node, int deep) {
String indentTabs = indentTabs(deep);
s += indentTabs + (node.token == null ? "" : node.token.toString()) + prettySubscriptions(node) + "\n";
s += indentTabs + (node.token == null ? "''" : node.token.toString()) + prettySubscriptions(node) + "\n";
}
private String prettySubscriptions(CNode node) {
if (node instanceof TNode) {
return "TNode";
}
if (node.subscriptions.isEmpty()) {
return StringUtil.EMPTY_STRING;
}
@@ -73,6 +77,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
private String indentTabs(int deep) {
StringBuilder s = new StringBuilder();
if (deep > 0) {
s.append(" ");
for (int i = 0; i < deep - 1; i++) {
s.append("| ");
}
@@ -196,6 +201,23 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
return subscriptions;
}
/**
*
* Cleans Disposes of TNode in separate Atomic CAS operation per
* http://bravenewgeek.com/breaking-and-entering-lose-the-lock-while-embracing-concurrency/
*
* We roughly follow this theory above, but we allow CNode with no Subscriptions to linger (for now).
*
*
* @param inode
* @param iParent
* @return
*/
public Action cleanTomb(INode inode, INode iParent) {
CNode updatedCnode = iParent.mainNode().copy();
updatedCnode.remove(inode);
return iParent.compareAndSet(iParent.mainNode(), updatedCnode) ? Action.OK : Action.REPEAT;
}
public void add(Subscription newSubscription) {
Action res;
@@ -259,25 +281,44 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
}
}
/**
*
* Removes subscription from CTrie, adds TNode when the last client unsubscribes, then calls for cleanTomb in a
* seperate atomic CAS operation.
*
*
* @param topic
* @param clientID
*/
public void removeSubscription(Topic topic, String clientID) {
Action res;
do {
res = remove(clientID, topic, this.root);
res = remove(clientID, topic, this.root, NO_PARENT);
} while (res == Action.REPEAT);
}
private Action remove(String clientId, Topic topic, INode inode) {
private Action remove(String clientId, Topic topic, INode inode, INode iParent) {
Token token = topic.headToken();
if (!topic.isEmpty() && (inode.mainNode().anyChildrenMatch(token))) {
Topic remainingTopic = topic.exceptHeadToken();
INode nextInode = inode.mainNode().childOf(token);
return remove(clientId, remainingTopic, nextInode);
return remove(clientId, remainingTopic, nextInode, inode);
} else {
final CNode cnode = inode.mainNode();
if (cnode.containsOnly(clientId)) {
if (cnode instanceof TNode) {
// this inode is a tomb, has no clients and should be cleaned up
// Because we implemented cleanTomb below, this should be rare, but possible
// Consider calling cleanTomb here too
return Action.OK;
}
if (cnode.containsOnly(clientId) && topic.isEmpty() && cnode.allChildren().isEmpty()) {
// last client to leave this node, AND there are no downstream children, remove via TNode tomb
if (inode == this.root) {
return inode.compareAndSet(cnode, inode.mainNode().copy()) ? Action.OK : Action.REPEAT;
}
TNode tnode = new TNode();
return inode.compareAndSet(cnode, tnode) ? Action.OK : Action.REPEAT;
} else if (cnode.contains(clientId)) {
return inode.compareAndSet(cnode, tnode) ? cleanTomb(inode, iParent) : Action.REPEAT;
} else if (cnode.contains(clientId) && topic.isEmpty()) {
CNode updatedCnode = cnode.copy();
updatedCnode.removeSubscriptionsFor(clientId);
return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT;
broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java
+7
View File
@@ -22,6 +22,9 @@ class INode {
INode(CNode mainNode) {
this.mainNode.set(mainNode);
if (mainNode instanceof TNode) { // this should never happen
throw new IllegalStateException("TNode should not be set on mainNnode");
}
}
boolean compareAndSet(CNode old, CNode newNode) {
@@ -35,4 +38,8 @@ class INode {
CNode mainNode() {
return this.mainNode.get();
}
boolean isTombed() {
return this.mainNode() instanceof TNode;
}
}
broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java
+169
View File
@@ -132,6 +132,95 @@ public class CTrieSubscriptionDirectoryTest {
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
}
@Test
public void givenTreeWithSomeNodeUnsubscribeAndResubscribeCleanTomb() {
sut.add(clientSubOnTopic("TempSensor1", "test"));
sut.removeSubscription(asTopic("test"), "TempSensor1");
sut.add(clientSubOnTopic("TempSensor1", "test"));
assertTrue(sut.root.mainNode().allChildren().size() == 1); // looking to see if TNode is cleaned up
}
@Test
public void givenTreeWithSomeNodeWhenRemoveMultipleTimes() {
sut.add(clientSubOnTopic("TempSensor1", "test"));
// make sure no TNode exceptions
sut.removeSubscription(asTopic("test"), "TempSensor1");
sut.removeSubscription(asTopic("test"), "TempSensor1");
sut.removeSubscription(asTopic("test"), "TempSensor1");
sut.removeSubscription(asTopic("test"), "TempSensor1");
//Verify
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
}
@Test
public void givenTreeWithSomeDeepNodeWhenRemoveMultipleTimes() {
sut.add(clientSubOnTopic("TempSensor1", "/test/me/1/2/3"));
// make sure no TNode exceptions
sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1");
sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1");
sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1");
//Verify
final Optional<CNode> matchedNode = sut.lookup(asTopic("/temp"));
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
}
@Test
public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionThenNodeIsUpdated() {
sut.add(clientSubOnTopic("TempSensor1", "/temp/1"));
sut.add(clientSubOnTopic("TempSensor1", "/temp/2"));
//Exercise
sut.removeSubscription(asTopic("/temp/1"), "TempSensor1");
sut.removeSubscription(asTopic("/temp/1"), "TempSensor1");
final Set<Subscription> matchingSubs = sut.recursiveMatch(asTopic("/temp/2"), sut.root);
//Verify
final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp/2"));
assertThat(matchingSubs).contains(expectedMatchingsub);
}
@Test
public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionSmallerThenNodeIsNotUpdated() {
sut.add(clientSubOnTopic("TempSensor1", "/temp/1"));
sut.add(clientSubOnTopic("TempSensor1", "/temp/2"));
//Exercise
sut.removeSubscription(asTopic("/temp"), "TempSensor1");
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1"), sut.root);
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("/temp/2"), sut.root);
//Verify
// not clear to me, but I believe /temp unsubscribe should not unsub you from downstream /temp/1 or /temp/2
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("/temp/1"));
assertThat(matchingSubs1).contains(expectedMatchingsub1);
final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("/temp/2"));
assertThat(matchingSubs2).contains(expectedMatchingsub2);
}
@Test
public void givenTreeWithDeepNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() {
sut.add(clientSubOnTopic("TempSensor1", "/bah/bin/bash"));
sut.removeSubscription(asTopic("/bah/bin/bash"),"TempSensor1");
//Verify
final Optional<CNode> matchedNode = sut.lookup(asTopic("/bah/bin/bash"));
assertFalse("Node on path /temp can't be present", matchedNode.isPresent());
}
@Test
public void testMatchSubscriptionNoWildcards() {
sut.add(clientSubOnTopic("TempSensor1", "/temp"));
@@ -143,4 +232,84 @@ public class CTrieSubscriptionDirectoryTest {
final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp"));
assertThat(matchingSubs).contains(expectedMatchingsub);
}
@Test
public void testRemovalInnerTopicOffRootSameClient() {
sut.add(clientSubOnTopic("TempSensor1", "temp"));
sut.add(clientSubOnTopic("TempSensor1", "temp/1"));
//Exercise
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root);
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
//Verify
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"));
final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("temp/1"));
assertThat(matchingSubs1).contains(expectedMatchingsub1);
assertThat(matchingSubs2).contains(expectedMatchingsub2);
sut.removeSubscription(asTopic("temp"),"TempSensor1");
//Exercise
final Set<Subscription> matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root);
final Set<Subscription> matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1);
assertThat(matchingSubs4).contains(expectedMatchingsub2);
}
@Test
public void testRemovalInnerTopicOffRootDiffClient() {
sut.add(clientSubOnTopic("TempSensor1", "temp"));
sut.add(clientSubOnTopic("TempSensor2", "temp/1"));
//Exercise
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root);
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
//Verify
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"));
final Subscription expectedMatchingsub2 = new Subscription("TempSensor2", asTopic("temp/1"));
assertThat(matchingSubs1).contains(expectedMatchingsub1);
assertThat(matchingSubs2).contains(expectedMatchingsub2);
sut.removeSubscription(asTopic("temp"),"TempSensor1");
//Exercise
final Set<Subscription> matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root);
final Set<Subscription> matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1);
assertThat(matchingSubs4).contains(expectedMatchingsub2);
}
@Test
public void testRemovalOuterTopicOffRootDiffClient() {
sut.add(clientSubOnTopic("TempSensor1", "temp"));
sut.add(clientSubOnTopic("TempSensor2", "temp/1"));
//Exercise
final Set<Subscription> matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root);
final Set<Subscription> matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
//Verify
final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"));
final Subscription expectedMatchingsub2 = new Subscription("TempSensor2", asTopic("temp/1"));
assertThat(matchingSubs1).contains(expectedMatchingsub1);
assertThat(matchingSubs2).contains(expectedMatchingsub2);
sut.removeSubscription(asTopic("temp/1"),"TempSensor2");
//Exercise
final Set<Subscription> matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root);
final Set<Subscription> matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root);
assertThat(matchingSubs3).contains(expectedMatchingsub1);
assertThat(matchingSubs4).doesNotContain(expectedMatchingsub2);
}
}