Optimize message bus topic-matching logic by 100× (#2634)

- Add is_matching regex based randomized testing
- Add bulk benchmarking logic
- Add iterative backtracking implementation
- Use iterative matching for 1000x improvement

Author: twitu

Cargo.lock

@@ -77,6 +77,8 @@ sysinfo = "0.35.1"
 proptest = { workspace = true }
 tempfile = { workspace = true }
 criterion = {workspace = true}
+rand.workspace = true
+regex = "1.11.1"
 
 [build-dependencies]
 cbindgen = { workspace = true, optional = true }
@@ -85,3 +87,8 @@ cbindgen = { workspace = true, optional = true }
 name = "cache_orders"
 path = "benches/cache/orders.rs"
 harness = false
+
+[[bench]]
+name = "matching"
+path = "benches/matching.rs"
+harness = false

crates/common/Cargo.toml

@@ -0,0 +1,106 @@
+// -------------------------------------------------------------------------------------------------
+//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
+//  https://nautechsystems.io
+//
+//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+//  You may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+// -------------------------------------------------------------------------------------------------
+
+use criterion::{Criterion, black_box, criterion_group, criterion_main};
+use nautilus_common::msgbus::{is_matching, is_matching_backtracking};
+use rand::{Rng, SeedableRng, rngs::StdRng};
+use regex::Regex;
+use ustr::Ustr;
+
+fn create_topics(n: usize, rng: &mut StdRng) -> Vec<Ustr> {
+    let cat = ["data", "info", "order"];
+    let model = ["quotes", "trades", "orderbooks", "depths"];
+    let venue = ["BINANCE", "BYBIT", "OKX", "FTX", "KRAKEN"];
+    let instrument = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "XRPUSDT", "DOGEUSDT"];
+
+    let mut topics = Vec::new();
+    for _ in 0..n {
+        let cat = cat[rng.random_range(0..cat.len())];
+        let model = model[rng.random_range(0..model.len())];
+        let venue = venue[rng.random_range(0..venue.len())];
+        let instrument = instrument[rng.random_range(0..instrument.len())];
+        topics.push(Ustr::from(&format!(
+            "{}.{}.{}.{}",
+            cat, model, venue, instrument
+        )));
+    }
+    topics
+}
+
+fn bench_matching(c: &mut Criterion) {
+    let pattern = "data.*.BINANCE.ETH???";
+    let pattern_ustr = Ustr::from(pattern);
+
+    {
+        let mut rng = StdRng::seed_from_u64(42);
+        let mut custom_group = c.benchmark_group("Custom matching");
+
+        for ele in [1, 10, 100, 1000] {
+            let topics = create_topics(ele, &mut rng);
+
+            custom_group.bench_function(format!("{} topics", ele), |b| {
+                b.iter(|| {
+                    for topic in topics.iter() {
+                        black_box(is_matching(&pattern_ustr, topic));
+                    }
+                });
+            });
+        }
+
+        custom_group.finish();
+    }
+
+    {
+        let mut rng = StdRng::seed_from_u64(42);
+        let mut regex_group = c.benchmark_group("Regex matching");
+
+        for ele in [1, 10, 100, 1000] {
+            let topics = create_topics(ele, &mut rng);
+
+            regex_group.bench_function(format!("{} topics", ele), |b| {
+                b.iter(|| {
+                    let regex = Regex::new(pattern).unwrap();
+                    for topic in topics.iter() {
+                        black_box(regex.is_match(topic));
+                    }
+                });
+            });
+        }
+
+        regex_group.finish();
+    }
+
+    {
+        let mut rng = StdRng::seed_from_u64(42);
+        let mut iter_group = c.benchmark_group("Iterative backtracking matching");
+
+        for ele in [1, 10, 100, 1000] {
+            let topics = create_topics(ele, &mut rng);
+
+            iter_group.bench_function(format!("{} topics", ele), |b| {
+                b.iter(|| {
+                    for topic in topics.iter() {
+                        black_box(is_matching_backtracking(&pattern_ustr, topic));
+                    }
+                });
+            });
+        }
+
+        iter_group.finish();
+    }
+}
+
+criterion_group!(benches, bench_matching);
+criterion_main!(benches);

crates/common/benches/matching.rs

@@ -182,7 +182,7 @@ pub fn subscribe<T: AsRef<str>>(topic: T, handler: ShareableMessageHandler, prio
     // Find existing patterns which match this topic
     let mut matches = Vec::new();
     for (pattern, subs) in msgbus_ref_mut.patterns.iter_mut() {
-        if is_matching(&Ustr::from(topic.as_ref()), pattern) {
+        if is_matching_backtracking(&Ustr::from(topic.as_ref()), pattern) {
             subs.push(sub.clone());
             subs.sort();
             // subs.sort_by(|a, b| a.priority.cmp(&b.priority).then_with(|| a.cmp(b)));
@@ -465,7 +465,7 @@ impl MessageBus {
 
         // Collect matching subscriptions from direct subscriptions
         matching_subs.extend(self.subscriptions.iter().filter_map(|(sub, _)| {
-            if is_matching(&sub.topic, pattern) {
+            if is_matching_backtracking(&sub.topic, pattern) {
                 Some(sub.clone())
             } else {
                 None
@@ -511,7 +511,7 @@ impl MessageBus {
         pattern: &'a Ustr,
     ) -> impl Iterator<Item = &'a ShareableMessageHandler> {
         self.subscriptions.iter().filter_map(move |(sub, _)| {
-            if is_matching(&sub.topic, pattern) {
+            if is_matching_backtracking(&sub.topic, pattern) {
                 Some(&sub.handler)
             } else {
                 None
@@ -593,6 +593,61 @@ pub fn is_matching(topic: &Ustr, pattern: &Ustr) -> bool {
     table[n][m]
 }
 
+/// Match a topic and a string pattern using iterative backtracking algorithm
+/// pattern can contains -
+/// '*' - match 0 or more characters after this
+/// '?' - match any character once
+/// 'a-z' - match the specific character
+#[must_use]
+pub fn is_matching_backtracking(topic: &Ustr, pattern: &Ustr) -> bool {
+    let topic_bytes = topic.as_bytes();
+    let pattern_bytes = pattern.as_bytes();
+
+    // Stack to store states for backtracking (topic_idx, pattern_idx)
+    let mut stack = vec![(0, 0)];
+
+    while let Some((mut i, mut j)) = stack.pop() {
+        loop {
+            // Found a match if we've consumed both strings
+            if i == topic.len() && j == pattern.len() {
+                return true;
+            }
+
+            // If we've reached the end of the pattern, break to try other paths
+            if j == pattern.len() {
+                break;
+            }
+
+            // Handle '*' wildcard
+            if pattern_bytes[j] == b'*' {
+                // Try skipping '*' entirely first
+                stack.push((i, j + 1));
+
+                // Continue with matching current character and keeping '*'
+                if i < topic.len() {
+                    i += 1;
+                    continue;
+                }
+                break;
+            }
+            // Handle '?' or exact character match
+            else if i < topic.len()
+                && (pattern_bytes[j] == b'?' || topic_bytes[i] == pattern_bytes[j])
+            {
+                // Continue matching linearly without stack operations
+                i += 1;
+                j += 1;
+                continue;
+            }
+
+            // No match found in current path
+            break;
+        }
+    }
+
+    false
+}
+
 impl Default for MessageBus {
     /// Creates a new default [`MessageBus`] instance.
     fn default() -> Self {

crates/common/src/msgbus/mod.rs

@@ -15,11 +15,13 @@
 
 use nautilus_core::UUID4;
 use nautilus_model::identifiers::TraderId;
+use rand::{Rng, SeedableRng, rngs::StdRng};
+use regex::Regex;
 use rstest::rstest;
 use ustr::Ustr;
 
 use crate::msgbus::{
-    self, MessageBus, get_message_bus, is_matching,
+    self, MessageBus, get_message_bus, is_matching, is_matching_backtracking,
     stubs::{
         check_handler_was_called, get_call_check_shareable_handler, get_stub_shareable_handler,
     },
@@ -217,4 +219,100 @@ fn test_is_matching(#[case] topic: &str, #[case] pattern: &str, #[case] expected
         is_matching(&Ustr::from(topic), &Ustr::from(pattern)),
         expected
     );
+    assert_eq!(
+        is_matching_backtracking(&Ustr::from(topic), &Ustr::from(pattern)),
+        expected
+    );
+}
+
+fn convert_pattern_to_regex(pattern: &str) -> String {
+    let mut regex = String::new();
+    regex.push('^');
+
+    for c in pattern.chars() {
+        match c {
+            '.' => regex.push_str("\\."),
+            '*' => regex.push_str(".*"),
+            '?' => regex.push_str("."),
+            _ => regex.push(c),
+        }
+    }
+
+    regex.push('$');
+    regex
+}
+
+#[rstest]
+#[case("a??.quo*es.?I?AN*ET?US*T", "^a..\\.quo.*es\\..I.AN.*ET.US.*T$")]
+#[case("da?*.?u*?s??*NC**ETH?", "^da..*\\..u.*.s...*NC.*.*ETH.$")]
+fn test_convert_pattern_to_regex(#[case] pat: &str, #[case] regex: &str) {
+    assert_eq!(convert_pattern_to_regex(pat), regex);
+}
+
+fn generate_pattern_from_topic(topic: &str, rng: &mut StdRng) -> String {
+    let mut pattern = String::new();
+
+    for c in topic.chars() {
+        let val: f64 = rng.random();
+        // 10% chance of wildcard
+        if val < 0.1 {
+            pattern.push('*');
+        }
+        // 20% chance of question mark
+        else if val < 0.3 {
+            pattern.push('?');
+        }
+        // 20% chance of skipping
+        else if val < 0.5 {
+            continue;
+        }
+        // 50% chance of keeping the character
+        else {
+            pattern.push(c);
+        };
+    }
+
+    pattern
+}
+
+#[rstest]
+fn test_matching_with_regex() {
+    let topic = "data.quotes.BINANCE.ETHUSDT";
+    let mut rng = StdRng::seed_from_u64(42);
+
+    for i in 0..1000 {
+        let pattern = generate_pattern_from_topic(topic, &mut rng);
+        let regex_pattern = convert_pattern_to_regex(&pattern);
+        let regex = Regex::new(&regex_pattern).unwrap();
+        assert_eq!(
+            is_matching(&Ustr::from(topic), &Ustr::from(&pattern)),
+            regex.is_match(topic),
+            "Failed to match on iteration: {}, pattern: \"{}\", topic: {}, regex: \"{}\"",
+            i,
+            pattern,
+            topic,
+            regex_pattern
+        );
+    }
+}
+
+#[rstest]
+fn test_matching_backtracking() {
+    let topic = "data.quotes.BINANCE.ETHUSDT";
+    let mut rng = StdRng::seed_from_u64(42);
+
+    for i in 0..1000 {
+        let pattern = generate_pattern_from_topic(topic, &mut rng);
+        let regex_pattern = convert_pattern_to_regex(&pattern);
+        let regex = Regex::new(&regex_pattern).unwrap();
+        assert_eq!(
+            is_matching_backtracking(&Ustr::from(topic), &Ustr::from(&pattern)),
+            regex.is_match(topic),
+            "Failed to match on iteration: {}, pattern: \"{}\", topic: {}, regex: \"{}\"",
+            i,
+            pattern,
+            topic,
+            regex_pattern
+        );
+    }
 }