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

Merge pull request #365 from mwilliford/master

Browse Source 
Fix for CTrie TNode exception and other improvements to CTrie related
This commit is contained in:
Andrea Selva
2017-10-25 09:05:41 +02:00
committed by GitHub
parent 3293478dd7 1bc6bdbbce
commit 95a8b313e5
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);
}
}