Fixed #633 adding subscriptions

ChangeLog.txt

@@ -1,4 +1,5 @@
 Version 0.17-SNAPSHOT:
+   [fix] resolved issue #633 of bad perfomance when adding many subscriptions to few topics, resolved in #758.
    [fix] resolved issue #629 that originated from subscription trees wide and flat, resolved in #630
    [dependency] updated Netty to 4.1.93 and tcnative to 2.0.61 (#755)
    [feature] add saved session expiry configurable through the `persistent_client_expiration` setting (#739).

broker/src/main/java/io/moquette/broker/PostOffice.java

@@ -455,7 +455,7 @@ public int countBatches() {
 
     private RoutingResults publish2Subscribers(ByteBuf payload, Topic topic, MqttQoS publishingQos,
                                                Set<String> filterTargetClients) {
-        Set<Subscription> topicMatchingSubscriptions = subscriptions.matchQosSharpening(topic);
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matchQosSharpening(topic);
         if (topicMatchingSubscriptions.isEmpty()) {
             // no matching subscriptions, clean exit
             LOG.trace("No matching subscriptions for topic: {}", topic);

broker/src/main/java/io/moquette/broker/subscriptions/CNode.java

@@ -21,18 +21,18 @@ class CNode implements Comparable<CNode> {
 
     private final Token token;
     private final List<INode> children;
-    Set<Subscription> subscriptions;
+    List<Subscription> subscriptions;
 
     CNode(Token token) {
         this.children = new ArrayList<>();
-        this.subscriptions = new HashSet<>();
+        this.subscriptions = new ArrayList<>();
         this.token = token;
     }
 
     //Copy constructor
-    private CNode(Token token, List<INode> children, Set<Subscription> subscriptions) {
+    private CNode(Token token, List<INode> children, List<Subscription> subscriptions) {
         this.token = token; // keep reference, root comparison in directory logic relies on it for now.
-        this.subscriptions = new HashSet<>(subscriptions);
+        this.subscriptions = new ArrayList<>(subscriptions);
         this.children = new ArrayList<>(children);
     }
 
@@ -86,16 +86,15 @@ public void remove(INode node) {
 
     CNode addSubscription(Subscription newSubscription) {
         // if already contains one with same topic and same client, keep that with higher QoS
-        if (subscriptions.contains(newSubscription)) {
-            final Subscription existing = subscriptions.stream()
-                .filter(s -> s.equals(newSubscription))
-                .findFirst().get();
+        int idx = Collections.binarySearch(subscriptions, newSubscription);
+        if (idx >= 0) {
+            // Subscription already exists
+            final Subscription existing = subscriptions.get(idx);
             if (existing.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
-                subscriptions.remove(existing);
-                subscriptions.add(new Subscription(newSubscription));
+                subscriptions.set(idx, newSubscription);
             }
         } else {
-            this.subscriptions.add(new Subscription(newSubscription));
+            this.subscriptions.add(-1 - idx, new Subscription(newSubscription));
         }
         return this;
     }

broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java

@@ -1,7 +1,9 @@
 package io.moquette.broker.subscriptions;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 
@@ -64,24 +66,24 @@ private NavigationAction evaluate(Topic topic, CNode cnode) {
         return NavigationAction.GODEEP;
     }
 
-    public Set<Subscription> recursiveMatch(Topic topic) {
+    public List<Subscription> recursiveMatch(Topic topic) {
         return recursiveMatch(topic, this.root);
     }
 
-    private Set<Subscription> recursiveMatch(Topic topic, INode inode) {
+    private List<Subscription> recursiveMatch(Topic topic, INode inode) {
         CNode cnode = inode.mainNode();
         if (cnode instanceof TNode) {
-            return Collections.emptySet();
+            return Collections.emptyList();
         }
         NavigationAction action = evaluate(topic, cnode);
         if (action == NavigationAction.MATCH) {
             return cnode.subscriptions;
         }
         if (action == NavigationAction.STOP) {
-            return Collections.emptySet();
+            return Collections.emptyList();
         }
         Topic remainingTopic = (ROOT.equals(cnode.getToken())) ? topic : topic.exceptHeadToken();
-        Set<Subscription> subscriptions = new HashSet<>();
+        List<Subscription> subscriptions = new ArrayList<>();
 
         // We should only consider the maximum three children children of
         // type #, + or exact match

broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java

@@ -76,13 +76,13 @@ Optional<CNode> lookup(Topic topic) {
      * @return the list of matching subscriptions, or empty if not matching.
      */
     @Override
-    public Set<Subscription> matchWithoutQosSharpening(Topic topic) {
+    public List<Subscription> matchWithoutQosSharpening(Topic topic) {
         return ctrie.recursiveMatch(topic);
     }
 
     @Override
-    public Set<Subscription> matchQosSharpening(Topic topic) {
-        final Set<Subscription> subscriptions = matchWithoutQosSharpening(topic);
+    public List<Subscription> matchQosSharpening(Topic topic) {
+        final List<Subscription> subscriptions = matchWithoutQosSharpening(topic);
 
         Map<String, Subscription> subsGroupedByClient = new HashMap<>();
         for (Subscription sub : subscriptions) {
@@ -92,7 +92,7 @@ public Set<Subscription> matchQosSharpening(Topic topic) {
                 subsGroupedByClient.put(sub.clientId, sub);
             }
         }
-        return new HashSet<>(subsGroupedByClient.values());
+        return new ArrayList<>(subsGroupedByClient.values());
     }
 
     @Override

broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java

@@ -26,9 +26,9 @@ public interface ISubscriptionsDirectory {
 
     Set<String> listAllSessionIds();
 
-    Set<Subscription> matchWithoutQosSharpening(Topic topic);
+    List<Subscription> matchWithoutQosSharpening(Topic topic);
 
-    Set<Subscription> matchQosSharpening(Topic topic);
+    List<Subscription> matchQosSharpening(Topic topic);
 
     void add(Subscription newSubscription);
 

broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java

@@ -42,6 +42,7 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonMap;
+import java.util.List;
 import static org.junit.jupiter.api.Assertions.*;
 
 public class PostOfficeInternalPublishTest {
@@ -337,7 +338,7 @@ protected void subscribe(MQTTConnection connection, String topic, MqttQoS desire
         final String clientId = connection.getClientId();
         Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos);
 
-        final Set<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
+        final List<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
         assertEquals(1, matchedSubscriptions.size());
         final Subscription onlyMatchedSubscription = matchedSubscriptions.iterator().next();
         assertEquals(expectedSubscription, onlyMatchedSubscription);

broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java

@@ -196,7 +196,7 @@ protected void subscribe(MQTTConnection connection, String topic, MqttQoS desire
         final String clientId = connection.getClientId();
         Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos);
 
-        final Set<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
+        final List<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
         assertEquals(1, matchedSubscriptions.size());
         final Subscription onlyMatchedSubscription = matchedSubscriptions.iterator().next();
         assertEquals(expectedSubscription, onlyMatchedSubscription);

broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java

@@ -144,7 +144,7 @@ protected void subscribe(EmbeddedChannel channel, String topic, MqttQoS desiredQ
         final String clientId = NettyUtils.clientID(channel);
         Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos);
 
-        final Set<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
+        final List<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
         assertEquals(1, matchedSubscriptions.size());
         final Subscription onlyMatchedSubscription = matchedSubscriptions.iterator().next();
         assertEquals(expectedSubscription, onlyMatchedSubscription);
@@ -164,7 +164,7 @@ protected void subscribe(MQTTConnection connection, String topic, MqttQoS desire
         final String clientId = connection.getClientId();
         Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos);
 
-        final Set<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
+        final List<Subscription> matchedSubscriptions = subscriptions.matchWithoutQosSharpening(new Topic(topic));
         assertEquals(1, matchedSubscriptions.size());
         final Subscription onlyMatchedSubscription = matchedSubscriptions.iterator().next();
         assertEquals(expectedSubscription, onlyMatchedSubscription);

broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java

@@ -41,6 +41,7 @@
 import static io.moquette.broker.PostOfficePublishTest.PUBLISHER_ID;
 import static io.netty.handler.codec.mqtt.MqttQoS.*;
 import static java.util.Collections.*;
+import java.util.List;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 
@@ -123,7 +124,7 @@ protected void subscribe(MQTTConnection connection, String topic, MqttQoS desire
         final String clientId = connection.getClientId();
         Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos);
 
-        final Set<Subscription> matchedSubscriptions = subscriptions.matchQosSharpening(new Topic(topic));
+        final List<Subscription> matchedSubscriptions = subscriptions.matchQosSharpening(new Topic(topic));
         assertEquals(1, matchedSubscriptions.size());
         //assertTrue(matchedSubscriptions.size() >=1);
         final Subscription onlyMatchedSubscription = matchedSubscriptions.iterator().next();

broker/src/test/java/io/moquette/broker/subscriptions/CTrieSpeedTest.java

@@ -16,8 +16,10 @@
 package io.moquette.broker.subscriptions;
 
 import static io.moquette.broker.subscriptions.Topic.asTopic;
+import io.netty.handler.codec.mqtt.MqttQoS;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Random;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.slf4j.Logger;
@@ -35,6 +37,13 @@ static Subscription clientSubOnTopic(String clientID, String topicName) {
         return new Subscription(clientID, asTopic(topicName), null);
     }
 
+    @Test
+    @Timeout(value = MAX_DURATION_S)
+    public void testManyClientsFewTopics() {
+        List<Subscription> subscriptionList = prepareSubscriptionsManyClientsFewTopic();
+        createSubscriptions(subscriptionList);
+    }
+
     @Test
     @Timeout(value = MAX_DURATION_S)
     public void testFlat() {
@@ -73,6 +82,15 @@ public void createSubscriptions(List<Subscription> results) {
         LOGGER.info("Added " + count + " subscriptions in " + duration + " ms (" + Math.round(1000.0 * count / duration) + "/s)");
     }
 
+    public List<Subscription> prepareSubscriptionsManyClientsFewTopic() {
+        List<Subscription> subscriptionList = new ArrayList<>(TOTAL_SUBSCRIPTIONS);
+        for (int i = 0; i < TOTAL_SUBSCRIPTIONS; i++) {
+            Topic topic = asTopic("topic/test/" + new Random().nextInt(1 + i % 10) + "/test");
+            subscriptionList.add(new Subscription("TestClient-" + i, topic, MqttQoS.AT_LEAST_ONCE));
+        }
+        return subscriptionList;
+    }
+
     public List<Subscription> prepareSubscriptionsFlat() {
         List<Subscription> results = new ArrayList<>(TOTAL_SUBSCRIPTIONS);
         int count = 0;

broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java

@@ -27,6 +27,7 @@
 
 import static io.moquette.broker.subscriptions.CTrieTest.clientSubOnTopic;
 import static io.moquette.broker.subscriptions.Topic.asTopic;
+import java.util.List;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -203,7 +204,7 @@ public void testOverlappingSubscriptions() {
         sut.add(specificSub);
 
         //Exercise
-        final Set<Subscription> matchingForSpecific = sut.matchQosSharpening(asTopic("a/b"));
+        final List<Subscription> matchingForSpecific = sut.matchQosSharpening(asTopic("a/b"));
 
         // Verify
         assertThat(matchingForSpecific.size()).isEqualTo(1);
@@ -234,7 +235,7 @@ public void removeSubscription_sameClients_subscribedSameTopic() {
         sut.removeSubscription(asTopic("/topic"), slashSub.clientId);
 
         // Verify
-        final Set<Subscription> matchingSubscriptions = sut.matchWithoutQosSharpening(asTopic("/topic"));
+        final List<Subscription> matchingSubscriptions = sut.matchWithoutQosSharpening(asTopic("/topic"));
         assertThat(matchingSubscriptions).isEmpty();
     }
 
@@ -252,7 +253,7 @@ public void duplicatedSubscriptionsWithDifferentQos() {
         this.sut.add(client1SubQoS2);
 
         // Verify
-        Set<Subscription> subscriptions = this.sut.matchQosSharpening(asTopic("client/test/b"));
+        List<Subscription> subscriptions = this.sut.matchQosSharpening(asTopic("client/test/b"));
         assertThat(subscriptions).contains(client1SubQoS2);
         assertThat(subscriptions).contains(client2Sub);