Switch to hash-based dictionary for effective subscribers collection

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

@@ -5,6 +5,9 @@
 import javasabr.mqtt.model.MqttProperties;
 import javasabr.mqtt.model.MqttServerConnectionConfig;
 import javasabr.mqtt.model.QoS;
+import javasabr.mqtt.model.subscriber.tree.ConcurrentSubscriberTree;
+import javasabr.mqtt.model.subscriber.tree.OptimizedSubscriberNode;
+import javasabr.mqtt.model.subscriber.tree.SubscriberNode;
 import javasabr.mqtt.network.MqttConnection;
 import javasabr.mqtt.network.MqttConnectionFactory;
 import javasabr.mqtt.network.handler.NetworkMqttUserReleaseHandler;
@@ -66,6 +69,7 @@
 import org.springframework.core.env.Environment;
 
 @CustomLog
+@SuppressWarnings("unused")
 @Configuration(proxyBeanMethods = false)
 public class MqttBrokerSpringConfig {
 
@@ -98,8 +102,13 @@ AuthenticationService authenticationService(
   }
 
   @Bean
-  SubscriptionService subscriptionService() {
-    return new InMemorySubscriptionService();
+  ConcurrentSubscriberTree subscriberTree() {
+    return new ConcurrentSubscriberTree(new OptimizedSubscriberNode());
+  }
+
+  @Bean
+  SubscriptionService subscriptionService(ConcurrentSubscriberTree subscriberTree) {
+    return new InMemorySubscriptionService(subscriberTree);
   }
 
   @Bean

buildSrc/build.gradle

@@ -5,4 +5,8 @@ plugins {
 repositories {
   mavenCentral()
   gradlePluginPortal()
-}
+}
+
+dependencies {
+  implementation 'me.champeau.jmh:jmh-gradle-plugin:0.7.3'
+}

buildSrc/src/main/groovy/configure-java.gradle

@@ -4,6 +4,7 @@ plugins {
   id("java")
   id("java-test-fixtures")
   id("configure-jacoco")
+  id("me.champeau.jmh")
 }
 
 java {
@@ -16,6 +17,8 @@ dependencies {
   compileOnly libs.jspecify
   compileOnly libs.lombok
   annotationProcessor libs.lombok
+  jmh 'org.openjdk.jmh:jmh-generator-annprocess:1.37'
+  jmhAnnotationProcessor 'org.openjdk.jmh:jmh-generator-annprocess:1.37'
 }
 
 test {
@@ -36,4 +39,4 @@ tasks.withType(JavaCompile).configureEach {
 
 processResources {
   filter(ReplaceTokens, tokens: [])
-}
+}

core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java

@@ -22,12 +22,14 @@
 import javasabr.rlib.collections.array.MutableArray;
 import lombok.AccessLevel;
 import lombok.CustomLog;
+import lombok.RequiredArgsConstructor;
 import lombok.experimental.FieldDefaults;
 
 /**
  * In memory subscription service based on {@link ConcurrentSubscriberTree}
  */
 @CustomLog
+@RequiredArgsConstructor
 @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
 public class InMemorySubscriptionService implements SubscriptionService {
 

model/src/jmh/java/javasabr/mqtt/model/subscriber/tree/SubscriberNodeBenchmark.java

@@ -0,0 +1,179 @@
+package javasabr.mqtt.model.subscriber.tree;
+
+import java.util.concurrent.TimeUnit;
+import javasabr.mqtt.model.MqttUser;
+import javasabr.mqtt.model.subscriber.SingleSubscriber;
+import javasabr.mqtt.model.subscription.Subscription;
+import javasabr.mqtt.model.subscription.TestMqttUser;
+import javasabr.mqtt.model.topic.TopicFilter;
+import javasabr.mqtt.model.topic.TopicName;
+import javasabr.mqtt.model.topic.tree.SubscriberTreeTest;
+import javasabr.rlib.collections.array.ArrayFactory;
+import javasabr.rlib.collections.array.MutableArray;
+import javasabr.rlib.collections.dictionary.DictionaryFactory;
+import javasabr.rlib.collections.dictionary.MutableRefToRefDictionary;
+import org.jspecify.annotations.NonNull;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.GroupThreads;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+@Warmup(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 20, time = 2, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+@Threads(100)
+public class SubscriberNodeBenchmark {
+
+  @Param({"100"})
+  public int subscribeFrequency;
+
+  private SubscriberNode originalImplementation;
+  private OptimizedSubscriberNode optimizedImplementation;
+
+  private TopicName[] topicNames;
+  private int[] lastLevels;
+
+  private TopicFilter[] topicFilters;
+  private Subscription[] subscriptions;
+
+  private static final String[] TEST_TOPIC_NAMES = {
+      "home/kitchen/sensor/temp",
+      "home/livingroom/light/state",
+      "system/events/alert/priority/low",
+      "command/reboot/device/101",
+      "$SYS/broker/clients/total",
+      "data",
+      "metrics/cpu/usage",
+      "user/id_12345/settings",
+      "long/path/with/many/segments/is/unlikely/but/possible",
+      "a/b/c/d/e/f/g",
+      "short"
+  };
+
+  private static final String[] TEST_FILTERS = {
+      "home/#",
+      "home/kitchen/sensor/temp",
+      "home/+/sensor/temp",
+      "home/+/+/temp",
+      "system/events/+/priority/#",
+      "$SYS/#",
+      "$SYS/broker/clients/+",
+      "#",
+      "+",
+      "+/+/+/+",
+      "a/b/c/d/e/f/g/#",
+      "+/kitchen/sensor/temp",
+      "data",
+      "data/#",
+      "command/+",
+      "system/events/#",
+      "user/+/settings"
+  };
+
+  @Setup
+  public void setup() {
+    topicNames = new TopicName[TEST_TOPIC_NAMES.length];
+    lastLevels = new int[TEST_TOPIC_NAMES.length];
+    for (int i = 0; i < TEST_TOPIC_NAMES.length; i++) {
+      topicNames[i] = new TopicName(TEST_TOPIC_NAMES[i]);
+      lastLevels[i] = topicNames[i].levelsCount() - 1;
+    }
+
+    topicFilters = new TopicFilter[TEST_FILTERS.length];
+    subscriptions = new Subscription[TEST_FILTERS.length];
+
+    originalImplementation = new SubscriberNode();
+    optimizedImplementation = new OptimizedSubscriberNode();
+
+    for (int i = 0; i < TEST_FILTERS.length; i++) {
+      String rawTopic = TEST_FILTERS[i];
+      TopicFilter topicFilter = new TopicFilter(rawTopic);
+      TestMqttUser owner = new TestMqttUser("init_user_" + i);
+
+      topicFilters[i] = topicFilter;
+      subscriptions[i] = SubscriberTreeTest.makeSubscription(rawTopic);
+
+      originalImplementation.subscribe(0, owner, subscriptions[i], topicFilter);
+      optimizedImplementation.subscribe(0, owner, subscriptions[i], topicFilter);
+    }
+  }
+
+  @State(Scope.Thread)
+  public static class ThreadState {
+    int index = 0;
+    public MutableArray<@NonNull SingleSubscriber> container;
+    public MutableRefToRefDictionary<MqttUser, SingleSubscriber> mapContainer;
+    private TestMqttUser userForSubscription;
+    @Setup(Level.Trial)
+    public void setupThread() {
+      container = ArrayFactory.mutableArray(SingleSubscriber.class);
+      mapContainer = DictionaryFactory.mutableRefToRefDictionary();
+      userForSubscription = new TestMqttUser("thread_user_" + Thread.currentThread().getId());
+    }
+  }
+
+  @Group("original")
+  @Benchmark
+  public void originalImplementationSubscribe(SubscriberNodeBenchmark benchmark, ThreadState state, Blackhole bh) {
+    int topicFilterIndex = state.index % topicFilters.length;
+    originalImplementation.subscribe(0, state.userForSubscription, subscriptions[topicFilterIndex], topicFilters[topicFilterIndex]);
+    bh.consume(topicFilterIndex);
+    state.index++;
+  }
+  @Group("original")
+  @Benchmark
+  public void originalImplementationUnsubscribe(SubscriberNodeBenchmark benchmark, ThreadState state, Blackhole bh) {
+    int topicFilterIndex = state.index % topicFilters.length;
+    originalImplementation.unsubscribe(0, state.userForSubscription, topicFilters[topicFilterIndex]);
+    bh.consume(topicFilterIndex);
+  }
+  @Group("original")
+  @GroupThreads(5)
+  @Benchmark
+  public void originalImplementationMatchTo(SubscriberNodeBenchmark benchmark, ThreadState state, Blackhole bh) {
+    int topicNameIndex = state.index % topicNames.length;
+    originalImplementation.matchesTo(0, topicNames[topicNameIndex], lastLevels[topicNameIndex], state.container);
+    bh.consume(state.container);
+    state.index++;
+  }
+
+  @Group("optimized")
+  @Benchmark
+  public void optimizedImplementationSubscribe(SubscriberNodeBenchmark benchmark, ThreadState state, Blackhole bh) {
+    int topicFilterIndex = state.index % topicFilters.length;
+    optimizedImplementation.subscribe(0, state.userForSubscription, subscriptions[topicFilterIndex], topicFilters[topicFilterIndex]);
+    bh.consume(topicFilterIndex);
+    state.index++;
+  }
+  @Group("optimized")
+  @Benchmark
+  public void optimizedImplementationUnsubscribe(SubscriberNodeBenchmark benchmark, ThreadState state, Blackhole bh) {
+    int topicFilterIndex = state.index % topicFilters.length;
+    optimizedImplementation.unsubscribe(0, state.userForSubscription, topicFilters[topicFilterIndex]);
+    bh.consume(topicFilterIndex);
+  }
+  @Group("optimized")
+  @GroupThreads(5)
+  @Benchmark
+  public void optimizedImplementationMatchTo(SubscriberNodeBenchmark benchmark, ThreadState state, Blackhole bh) {
+    int topicNameIndex = state.index % topicNames.length;
+    optimizedImplementation.matchesTo(0, topicNames[topicNameIndex], lastLevels[topicNameIndex], state.mapContainer);
+    bh.consume(state.container);
+    state.index++;
+  }
+}

model/src/main/java/javasabr/mqtt/model/subscriber/tree/ConcurrentSubscriberTree.java

@@ -6,19 +6,23 @@
 import javasabr.mqtt.model.topic.TopicFilter;
 import javasabr.mqtt.model.topic.TopicName;
 import javasabr.rlib.collections.array.Array;
-import javasabr.rlib.collections.array.MutableArray;
+import javasabr.rlib.collections.array.ArrayFactory;
+import javasabr.rlib.collections.dictionary.DictionaryFactory;
+import javasabr.rlib.collections.dictionary.MutableRefToRefDictionary;
 import javasabr.rlib.common.ThreadSafe;
 import lombok.AccessLevel;
+import lombok.RequiredArgsConstructor;
 import lombok.experimental.FieldDefaults;
 import org.jspecify.annotations.Nullable;
 
+@RequiredArgsConstructor
 @FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
 public class ConcurrentSubscriberTree implements ThreadSafe {
 
-  SubscriberNode rootNode;
+  SubscriberTreeBase rootNode;
 
   public ConcurrentSubscriberTree() {
-    this.rootNode = new SubscriberNode();
+    this.rootNode = new OptimizedSubscriberNode();
   }
 
   @Nullable
@@ -31,14 +35,8 @@ public boolean unsubscribe(MqttUser user, TopicFilter topicFilter) {
   }
 
   public Array<SingleSubscriber> matches(TopicName topicName) {
-    var resultArray = MutableArray.ofType(SingleSubscriber.class);
-    matchesTo(resultArray, topicName);
-    return resultArray;
-  }
-
-  public MutableArray<SingleSubscriber> matchesTo(MutableArray<SingleSubscriber> container, TopicName topicName) {
-    var resultArray = MutableArray.ofType(SingleSubscriber.class);
-    rootNode.matchesTo(0, topicName, topicName.levelsCount() - 1, container);
-    return resultArray;
+    MutableRefToRefDictionary<MqttUser, SingleSubscriber> resultArray = DictionaryFactory.mutableRefToRefDictionary();
+    rootNode.matchesTo(0, topicName, topicName.levelsCount() - 1, resultArray);
+    return resultArray.values(ArrayFactory.mutableArray(SingleSubscriber.class));
   }
 }