From 85872eec1909282da3275f2d3b71c8850761cb7c Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 10 Oct 2015 19:16:30 +0200 Subject: [PATCH] Changed SubscriptionsStore overall tree (ones based on clientID) operations to copy all the tree --- .../moquette/spi/impl/ProtocolProcessor.java | 3 +- .../spi/impl/subscriptions/Subscription.java | 10 +- .../subscriptions/SubscriptionsStore.java | 114 ++++++++++-------- .../spi/impl/subscriptions/TreeNode.java | 46 ++++--- 4 files changed, 107 insertions(+), 66 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 1fe642bf..ea24545a 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -18,7 +18,6 @@ package org.eclipse.moquette.spi.impl; import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMatchingCondition; @@ -427,6 +426,8 @@ public class ProtocolProcessor { } LOG.debug("clientIDs are {}", m_clientIDs); if (m_clientIDs.get(clientId) == null) { + //TODO while we were publishing to the target client, that client disconnected, + // could happen is not an error HANDLE IT throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs)); } ServerChannel session = m_clientIDs.get(clientId).getSession(); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java index bce6c855..5c652f1a 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java @@ -39,7 +39,15 @@ public class Subscription implements Serializable { this.topicFilter = topicFilter; this.cleanSession = cleanSession; } - + + public Subscription(Subscription orig) { + this.requestedQos = orig.requestedQos; + this.clientId = orig.clientId; + this.topicFilter = orig.topicFilter; + this.cleanSession = orig.cleanSession; + this.active = orig.active; + } + /** * Factory method for empty subscriptions */ diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index fc8bfafd..fb78242c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -64,6 +64,7 @@ public class SubscriptionsStore { String s = ""; + @Override public void visit(TreeNode node, int deep) { String subScriptionsStr = ""; String indentTabs = indentTabs(deep); @@ -82,7 +83,8 @@ public class SubscriptionsStore { } return s; } - + + @Override public String getResult() { return s; } @@ -90,12 +92,14 @@ public class SubscriptionsStore { private class SubscriptionTreeCollector implements IVisitor> { - private List m_allSubscriptions = new ArrayList(); + private List m_allSubscriptions = new ArrayList<>(); + @Override public void visit(TreeNode node, int deep) { m_allSubscriptions.addAll(node.subscriptions()); } - + + @Override public List getResult() { return m_allSubscriptions; } @@ -136,32 +140,9 @@ public class SubscriptionsStore { couple.createdNode.addSubscription(newSubscription); //createdNode could be null? //spin lock repeating till we can, swap root, if can't swap just re-do the operation } while(!subscriptions.compareAndSet(oldRoot, couple.root)); + LOG.debug("root ref {}, original root was {}", couple.root, oldRoot); } - private TreeNode findMatchingNode(String topic) { - List tokens = new ArrayList<>(); - try { - tokens = parseTopic(topic); - } catch (ParseException ex) { - //TODO handle the parse exception - LOG.error(null, ex); -// return; - } - - TreeNode current = subscriptions.get(); - for (Token token : tokens) { - TreeNode matchingChildren; - - //check if a children with the same token already exists - if ((matchingChildren = current.childWithToken(token)) != null) { - current = matchingChildren; - } else { - //can't find any match - return null; - } - } - return current; - } protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) { List tokens = new ArrayList<>(); @@ -203,20 +184,27 @@ public class SubscriptionsStore { public void removeSubscription(String topic, String clientID) { - TreeNode matchNode = findMatchingNode(topic); - - //search for the subscription to remove - Subscription toBeRemoved = null; - for (Subscription sub : matchNode.subscriptions()) { - if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) { - toBeRemoved = sub; - break; + TreeNode oldRoot; + NodeCouple couple; + do { + oldRoot = subscriptions.get(); + couple = recreatePath(topic, oldRoot); + + //do the job + //search for the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : couple.createdNode.subscriptions()) { + if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) { + toBeRemoved = sub; + break; + } } - } - - if (toBeRemoved != null) { - matchNode.subscriptions().remove(toBeRemoved); - } + + if (toBeRemoved != null) { + couple.createdNode.subscriptions().remove(toBeRemoved); + } + //spin lock repeating till we can, swap root, if can't swap just re-do the operation + } while(!subscriptions.compareAndSet(oldRoot, couple.root)); } /** @@ -228,35 +216,61 @@ public class SubscriptionsStore { List allSubscriptions = subsCollector.getResult(); for (Subscription subscription : allSubscriptions) { - //TODO modify the data structure removeSubscription(subscription.getTopicFilter(), subscription.getClientId()); } } /** - * Visit the topics tree to remove matching subscriptions with clientID + * Visit the topics tree to remove matching subscriptions with clientID. + * It's a mutating structure operation so create a new subscription tree (partial or total). */ - //TODO modify the data structure public void removeForClient(String clientID) { - subscriptions.get().removeClientSubscriptions(clientID); + TreeNode oldRoot; + TreeNode newRoot; + do { + oldRoot = subscriptions.get(); + newRoot = oldRoot.removeClientSubscriptions(clientID); + //spin lock repeating till we can, swap root, if can't swap just re-do the operation + } while(!subscriptions.compareAndSet(oldRoot, newRoot)); //persist the update m_sessionsStore.wipeSubscriptions(clientID); } - //TODO modify the data structure + /** + * Visit the topics tree to deactivate matching subscriptions with clientID. + * It's a mutating structure operation so create a new subscription tree (partial or total). + */ public void deactivate(String clientID) { - subscriptions.get().deactivate(clientID); + LOG.debug("Disactivating subscriptions for clientID <{}>", clientID); + TreeNode oldRoot; + TreeNode newRoot; + do { + oldRoot = subscriptions.get(); + newRoot = oldRoot.deactivate(clientID); + //spin lock repeating till we can, swap root, if can't swap just re-do the operation + } while(!subscriptions.compareAndSet(oldRoot, newRoot)); + //persist the update - Set subs = subscriptions.get().findAllByClientID(clientID); + Set subs = newRoot.findAllByClientID(clientID); m_sessionsStore.updateSubscriptions(clientID, subs); } - //TODO modify the data structure + /** + * Visit the topics tree to activate matching subscriptions with clientID. + * It's a mutating structure operation so create a new subscription tree (partial or total). + */ public void activate(String clientID) { LOG.debug("Activating subscriptions for clientID <{}>", clientID); - subscriptions.get().activate(clientID); + TreeNode oldRoot; + TreeNode newRoot; + do { + oldRoot = subscriptions.get(); + newRoot = oldRoot.activate(clientID); + //spin lock repeating till we can, swap root, if can't swap just re-do the operation + } while(!subscriptions.compareAndSet(oldRoot, newRoot)); + //persist the update - Set subs = subscriptions.get().findAllByClientID(clientID); + Set subs = newRoot.findAllByClientID(clientID); m_sessionsStore.updateSubscriptions(clientID, subs); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java index 46ea1683..2ca8da24 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java @@ -73,7 +73,10 @@ class TreeNode { final TreeNode copy = new TreeNode(this); copy.m_parent = m_parent; copy.m_children = new ArrayList<>(m_children); - copy.m_subscriptions = new ArrayList<>(m_subscriptions); + copy.m_subscriptions = new ArrayList<>(m_subscriptions.size()); + for (Subscription sub : m_subscriptions) { + copy.m_subscriptions.add(new Subscription(sub)); + } copy.m_token = m_token; return copy; } @@ -144,55 +147,70 @@ class TreeNode { return res; } - void removeClientSubscriptions(String clientID) { + /** + * Create a copied subtree rooted on this node but purged of clientID's subscriptions. + * */ + TreeNode removeClientSubscriptions(String clientID) { //collect what to delete and then delete to avoid ConcurrentModification + TreeNode newSubRoot = this.copy(); List subsToRemove = new ArrayList<>(); - for (Subscription s : m_subscriptions) { + for (Subscription s : newSubRoot.m_subscriptions) { if (s.clientId.equals(clientID)) { subsToRemove.add(s); } } for (Subscription s : subsToRemove) { - m_subscriptions.remove(s); + newSubRoot.m_subscriptions.remove(s); } //go deep - for (TreeNode child : m_children) { - child.removeClientSubscriptions(clientID); + List newChildren = new ArrayList<>(newSubRoot.m_children.size()); + for (TreeNode child : newSubRoot.m_children) { + newChildren.add(child.removeClientSubscriptions(clientID)); } + newSubRoot.m_children = newChildren; + return newSubRoot; } /** * Deactivate all topic subscriptions for the given clientID. * */ - void deactivate(String clientID) { - for (Subscription s : m_subscriptions) { + TreeNode deactivate(String clientID) { + TreeNode newSubRoot = this.copy(); + for (Subscription s : newSubRoot.m_subscriptions) { if (s.clientId.equals(clientID)) { s.setActive(false); } } //go deep - for (TreeNode child : m_children) { - child.deactivate(clientID); + List newChildren = new ArrayList<>(newSubRoot.m_children.size()); + for (TreeNode child : newSubRoot.m_children) { + newChildren.add(child.deactivate(clientID)); } + newSubRoot.m_children = newChildren; + return newSubRoot; } /** * Activate all topic subscriptions for the given clientID. * */ - public void activate(String clientID) { - for (Subscription s : m_subscriptions) { + TreeNode activate(String clientID) { + TreeNode newSubRoot = this.copy(); + for (Subscription s : newSubRoot.m_subscriptions) { if (s.clientId.equals(clientID)) { s.setActive(true); } } //go deep - for (TreeNode child : m_children) { - child.activate(clientID); + List newChildren = new ArrayList<>(newSubRoot.m_children.size()); + for (TreeNode child : newSubRoot.m_children) { + newChildren.add(child.activate(clientID)); } + newSubRoot.m_children = newChildren; + return newSubRoot; } /**