Converted subscription from HashSet to ArrayList (#758)

Converted subscription from HashSet to ArrayList Copying a Set is much slower than copying an ArrayList. Since the slow part in the code is the copy step, using an ArrayList is much faster overall. Since the code using the subscriptions only iterates over them, the change from Set to List makes no real difference. Fixes #633
moquette-io · Jul 31, 2023 · 8e816a3 · 8e816a3
1 parent c7017bf
commit 8e816a3
Show file tree

Hide file tree

Showing 13 changed files with 73 additions and 49 deletions.
diff --git a/ChangeLog.txt b/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).

diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/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);

diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CNode.java b/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;
     }

diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java b/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

diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java b/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

diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java b/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);
 

diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/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);

diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/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);

diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/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);

diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/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();

diff --git a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSpeedTest.java b/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;

diff --git a/...rc/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java b/...rc/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);