From 1bc6bdbbce91dfce1dc69feec0959ddbdf495da1 Mon Sep 17 00:00:00 2001 From: Marcus Williford Date: Fri, 20 Oct 2017 11:54:36 -0700 Subject: [PATCH] Fix for CTrie TNode exception and other improvements to CTrie related 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. --- .../spi/impl/subscriptions/CNode.java | 16 +- .../CTrieSubscriptionDirectory.java | 55 +++++- .../spi/impl/subscriptions/INode.java | 7 + .../CTrieSubscriptionDirectoryTest.java | 169 ++++++++++++++++++ 4 files changed, 233 insertions(+), 14 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java index 74a29df0..4c722dcc 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java @@ -24,9 +24,7 @@ class CNode { Token token; private List children; - Set subscriptions = new HashSet<>(); - - //private int subtreeSubscriptions; + Set subscriptions; CNode() { this.children = new ArrayList<>(); @@ -35,9 +33,9 @@ class CNode { //Copy constructor private CNode(Token token, List children, Set 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)) diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java index 56968c61..47e3b77f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java @@ -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; diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java index 0dbeec59..2385d77c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java @@ -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; + } } diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java index 9fb34588..9ce84dc4 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java @@ -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 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 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 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 matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1"), sut.root); + final Set 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 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 matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set 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 matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set 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 matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set 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 matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set 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 matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set 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 matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + assertThat(matchingSubs3).contains(expectedMatchingsub1); + assertThat(matchingSubs4).doesNotContain(expectedMatchingsub2); + } } + +