perf(mqtt): wildcard topic trie and a shared strict UTF-8 decoder (R06:M-6, M-7)

Wildcard-authored tags were matched by one MqttTopicFilterComparer call per tag per inbound message on the dispatcher thread. They now live in an allocation-free level trie whose candidates are confirmed by the same comparer (semantics unchanged; pinned differentially against the comparer). TryDecodeUtf8 reuses one static throw-on-invalid UTF8Encoding instead of allocating one per Raw/Scalar delivery. Refs #744 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 15:38:29 -04:00
parent c14a3388f1
commit 6a51025012
3 changed files with 311 additions and 13 deletions
@@ -48,7 +48,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
 ///         accepted by the parser (warned at deploy, see <c>MqttTagDefinitionFactory.Inspect</c>), so
 ///         matching uses MQTTnet's own <see cref="MqttTopicFilterComparer"/> rather than string
 ///         equality — the same comparer a broker applies. Concrete topics stay a single dictionary
-///         lookup; the linear wildcard scan runs only when a wildcard tag actually exists.
+///         lookup; wildcard filters live in a level trie (<c>TopicFilterTrie</c>) whose candidates the comparer
+///         confirms, so a thousand wildcard tags cost a walk over the topic's levels, not a thousand compares.
 ///     </para>
 ///     <para>
 ///         <b>Sparkplug seam.</b> P1 ingests plain topics only. Sparkplug B decodes a
@@ -74,6 +75,13 @@ public sealed class MqttSubscriptionManager
    /// </summary>
    public const int DefaultQos = 1;

+    /// <summary>
+    ///     Throw-on-invalid UTF-8, shared (R06:M-7): it is stateless and thread-safe, and allocating one per
+    ///     Raw/Scalar delivery was per-message garbage on the dispatcher thread. Same choice
+    ///     <c>MqttBrowseSession</c> makes.
+    /// </summary>
+    private static readonly UTF8Encoding Utf8Strict = new(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true);
+
    // OPC UA status codes (values verified against Opc.Ua.StatusCodes, not recalled).
    private const uint StatusGood = 0x00000000u;

@@ -644,13 +652,14 @@ public sealed class MqttSubscriptionManager
            }

            // A published topic name may not itself contain a wildcard, so a wildcard-authored
-            // definition can never also be an exact-index hit: no tag is delivered twice.
-            foreach (var def in authored.Wildcards)
+            // definition can never also be an exact-index hit: no tag is delivered twice. The trie walks
+            // the topic's levels (R06:M-6 — this used to be one comparer call per wildcard tag, per
+            // message, on the shared dispatcher thread) and every candidate it yields is confirmed by
+            // MQTTnet's own comparer inside the visitor, so the semantics are the broker's.
+            if (authored.WildcardTrie.Count > 0)
            {
-                if (MqttTopicFilterComparer.Compare(topic, def.Topic) == MqttTopicFilterCompareResult.IsMatch)
-                {
-                    Deliver(def, payload, retained, now, oversized, ref json);
-                }
+                var visitor = new WildcardDelivery(this, topic, payload, retained, now, oversized, ref json);
+                authored.WildcardTrie.Match(topic, ref visitor);
            }
        }
        finally
@@ -1447,8 +1456,7 @@ public sealed class MqttSubscriptionManager
    {
        try
        {
-            text = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false, throwOnInvalidBytes: true)
-                .GetString(payload);
+            text = Utf8Strict.GetString(payload);
            return true;
        }
        catch (DecoderFallbackException)
@@ -1476,6 +1484,47 @@ public sealed class MqttSubscriptionManager
        return $"mqtt:{_mode}:{id}:[{named}{overflow}]";
    }

+    /// <summary>
+    ///     Delivers each wildcard-trie candidate whose filter the MQTTnet comparer confirms. A <c>ref struct</c>
+    ///     so the payload span and the fan-out's shared JSON parse ride along without a closure or a list.
+    /// </summary>
+    private ref struct WildcardDelivery : ITopicFilterVisitor<MqttTagDefinition>
+    {
+        private readonly MqttSubscriptionManager _manager;
+        private readonly string _topic;
+        private readonly ReadOnlySpan<byte> _payload;
+        private readonly bool _retained;
+        private readonly DateTime _now;
+        private readonly bool _oversized;
+        private ref PayloadJson _json;
+
+        public WildcardDelivery(
+            MqttSubscriptionManager manager,
+            string topic,
+            ReadOnlySpan<byte> payload,
+            bool retained,
+            DateTime now,
+            bool oversized,
+            ref PayloadJson json)
+        {
+            _manager = manager;
+            _topic = topic;
+            _payload = payload;
+            _retained = retained;
+            _now = now;
+            _oversized = oversized;
+            _json = ref json;
+        }
+
+        public void Visit(MqttTagDefinition def)
+        {
+            if (MqttTopicFilterComparer.Compare(_topic, def.Topic) == MqttTopicFilterCompareResult.IsMatch)
+            {
+                _manager.Deliver(def, _payload, _retained, _now, _oversized, ref _json);
+            }
+        }
+    }
+
    /// <summary>Opaque subscription identity; the diagnostic id names the mode and its topics.</summary>
    /// <param name="DiagnosticId">The human-readable identity surfaced for diagnostics.</param>
    private sealed record MqttSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;
@@ -1487,7 +1536,8 @@ public sealed class MqttSubscriptionManager
    /// <remarks>
    ///     <b>Two topic indexes, and the difference is load-bearing.</b> Delivery matches an
    ///     <i>incoming published topic</i>, which never carries a wildcard, so it uses
-    ///     <see cref="ByExactTopic"/> (concrete only) plus a comparer scan of <see cref="Wildcards"/>.
+    ///     <see cref="ByExactTopic"/> (concrete only) plus a <see cref="WildcardTrie"/> walk whose candidates the
+    ///     comparer confirms.
    ///     Every path that instead reconstructs state from a <i>subscribed filter string</i> —
    ///     reconnect re-subscribe and per-topic degradation — must use <see cref="ByFilter"/>, which is
    ///     keyed by the tag's authored topic <b>including wildcard patterns</b>, because that is what
@@ -1501,12 +1551,12 @@ public sealed class MqttSubscriptionManager
    ///     included. Keyed the same way <c>_subscribedTopics</c> and SUBACK outcomes are.
    /// </param>
    /// <param name="ByExactTopic">Concrete topic → its definitions; the delivery fast path.</param>
-    /// <param name="Wildcards">Definitions whose authored topic carries an MQTT wildcard.</param>
+    /// <param name="WildcardTrie">Definitions whose authored topic carries an MQTT wildcard, by filter level.</param>
    private sealed record AuthoredTable(
        FrozenDictionary<string, MqttTagDefinition> ByRawPath,
        FrozenDictionary<string, MqttTagDefinition[]> ByFilter,
        FrozenDictionary<string, MqttTagDefinition[]> ByExactTopic,
-        MqttTagDefinition[] Wildcards)
+        TopicFilterTrie<MqttTagDefinition> WildcardTrie)
    {
        public static readonly AuthoredTable Empty =
            Build(new Dictionary<string, MqttTagDefinition>(StringComparer.Ordinal));
@@ -1556,7 +1606,7 @@ public sealed class MqttSubscriptionManager
                byRawPath.ToFrozenDictionary(StringComparer.Ordinal),
                byFilter.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal),
                concrete.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.ToArray(), StringComparer.Ordinal),
-                [.. wildcards]);
+                TopicFilterTrie<MqttTagDefinition>.Build(wildcards.Select(d => (d.Topic, d))));
        }
    }
 }
@@ -0,0 +1,152 @@
+using System.Collections.Frozen;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt;
+
+/// <summary>Receives each candidate a <see cref="TopicFilterTrie{T}"/> match produces.</summary>
+/// <typeparam name="T">The value type stored against each filter.</typeparam>
+internal interface ITopicFilterVisitor<in T>
+{
+    /// <summary>Called once per value whose filter may match the topic.</summary>
+    /// <param name="value">The candidate.</param>
+    void Visit(T value);
+}
+
+/// <summary>
+///     An immutable trie over MQTT topic FILTERS (<c>+</c> / <c>#</c> wildcards), so matching one published
+///     topic costs a walk over its levels instead of a comparer call per wildcard-authored tag (R06:M-6:
+///     the linear scan ran on MQTTnet's shared dispatcher thread for every inbound message).
+/// </summary>
+/// <remarks>
+///     <para>
+///         <b>A candidate generator, not the authority.</b> The trie implements the plain MQTT 3.1.1 §4.7
+///         level rules — <c>+</c> is exactly one level (an empty one included), <c>#</c> is the parent level
+///         and everything under it — and deliberately nothing subtler (the <c>$</c>-topic exclusion, shared
+///         subscriptions). It can therefore only ever return a SUPERSET of the true matches, and every
+///         caller confirms each candidate with MQTTnet's own <c>MqttTopicFilterComparer</c>, the comparer a
+///         broker applies. Semantics stay byte-identical to the scan it replaces; only the work changes.
+///         A differential test pins both halves against that comparer.
+///     </para>
+///     <para>
+///         <b>Allocation-free per match.</b> Levels are walked as spans and looked up through the
+///         dictionary's span alternate lookup; candidates are pushed to a caller-supplied visitor (a
+///         <c>ref struct</c> in production) rather than collected into a list.
+///     </para>
+/// </remarks>
+/// <typeparam name="T">The value type stored against each filter.</typeparam>
+internal sealed class TopicFilterTrie<T>
+{
+    private readonly Node _root;
+
+    private TopicFilterTrie(Node root, int count)
+    {
+        _root = root;
+        Count = count;
+    }
+
+    /// <summary>An empty trie.</summary>
+    public static TopicFilterTrie<T> Empty { get; } = Build([]);
+
+    /// <summary>How many values were inserted.</summary>
+    public int Count { get; }
+
+    /// <summary>Builds a trie from <c>(filter, value)</c> pairs.</summary>
+    /// <param name="entries">The filters and their values.</param>
+    /// <returns>The immutable trie.</returns>
+    public static TopicFilterTrie<T> Build(IEnumerable<(string Filter, T Value)> entries)
+    {
+        var root = new Builder();
+        var count = 0;
+        foreach (var (filter, value) in entries)
+        {
+            var node = root;
+            var levels = filter.Split('/');
+            var terminated = false;
+            for (var i = 0; i < levels.Length; i++)
+            {
+                var level = levels[i];
+                if (level == "#")
+                {
+                    // '#' must be the last level; anything after it is a malformed filter the comparer
+                    // will refuse anyway, so treating '#' as terminal here can only over-generate.
+                    node.Hash.Add(value);
+                    terminated = true;
+                    break;
+                }
+
+                node = level == "+"
+                    ? node.Plus ??= new Builder()
+                    : node.Children.TryGetValue(level, out var child) ? child : node.Children[level] = new Builder();
+            }
+
+            if (!terminated) node.Exact.Add(value);
+            count++;
+        }
+
+        return new TopicFilterTrie<T>(root.Freeze(), count);
+    }
+
+    /// <summary>Pushes every value whose filter may match <paramref name="topic"/> to <paramref name="visitor"/>.</summary>
+    /// <typeparam name="TVisitor">The visitor; a <c>ref struct</c> is allowed.</typeparam>
+    /// <param name="topic">A concrete published topic (never a filter).</param>
+    /// <param name="visitor">Receives each candidate — a superset of the true matches; confirm each one.</param>
+    public void Match<TVisitor>(string topic, ref TVisitor visitor)
+        where TVisitor : ITopicFilterVisitor<T>, allows ref struct
+    {
+        if (Count == 0 || topic is null) return;
+        Walk(_root, topic.AsSpan(), exhausted: false, ref visitor);
+    }
+
+    private static void Walk<TVisitor>(Node node, ReadOnlySpan<char> rest, bool exhausted, ref TVisitor visitor)
+        where TVisitor : ITopicFilterVisitor<T>, allows ref struct
+    {
+        // '#' here covers this level's parent-match AND every deeper level.
+        foreach (var value in node.Hash) visitor.Visit(value);
+
+        if (exhausted)
+        {
+            foreach (var value in node.Exact) visitor.Visit(value);
+            return;
+        }
+
+        var slash = rest.IndexOf('/');
+        var level = slash < 0 ? rest : rest[..slash];
+        var next = slash < 0 ? ReadOnlySpan<char>.Empty : rest[(slash + 1)..];
+        var nextExhausted = slash < 0;
+
+        if (node.Children.Count > 0
+            && node.Children.GetAlternateLookup<ReadOnlySpan<char>>().TryGetValue(level, out var child))
+        {
+            Walk(child, next, nextExhausted, ref visitor);
+        }
+
+        if (node.Plus is { } plus) Walk(plus, next, nextExhausted, ref visitor);
+    }
+
+    private sealed class Node(FrozenDictionary<string, Node> children, Node? plus, T[] exact, T[] hash)
+    {
+        public FrozenDictionary<string, Node> Children { get; } = children;
+
+        public Node? Plus { get; } = plus;
+
+        public T[] Exact { get; } = exact;
+
+        public T[] Hash { get; } = hash;
+    }
+
+    private sealed class Builder
+    {
+        public Dictionary<string, Builder> Children { get; } = new(StringComparer.Ordinal);
+
+        public Builder? Plus { get; set; }
+
+        public List<T> Exact { get; } = [];
+
+        public List<T> Hash { get; } = [];
+
+        public Node Freeze() => new(
+            Children.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.Freeze(), StringComparer.Ordinal),
+            Plus?.Freeze(),
+            [.. Exact],
+            [.. Hash]);
+    }
+}
@@ -0,0 +1,96 @@
+using MQTTnet;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
+
+/// <summary>
+///     <b>R06:M-6.</b> The wildcard trie replaced a linear <c>MqttTopicFilterComparer</c> scan on the
+///     dispatcher thread. Its contract is "a superset of the true matches, confirmed by the comparer", so it
+///     is pinned differentially against that comparer — the oracle a broker uses — over a seeded random
+///     space that deliberately includes empty levels, <c>$</c>-topics and wildcard-only filters.
+/// </summary>
+public sealed class TopicFilterTrieTests
+{
+    private static readonly string[] Levels = ["a", "b", "c", "", "$SYS", "x y"];
+
+    private sealed class Collect : ITopicFilterVisitor<int>
+    {
+        public HashSet<int> Seen { get; } = [];
+
+        public void Visit(int value) => Seen.Add(value).ShouldBeTrue("a filter must be visited at most once per match");
+    }
+
+    [Fact]
+    public void Match_IsASupersetOfTheComparer_AndConfirmingYieldsExactlyTheComparersAnswer()
+    {
+        var random = new Random(744);
+        for (var round = 0; round < 200; round++)
+        {
+            var filters = Enumerable.Range(0, random.Next(1, 25)).Select(_ => RandomFilter(random)).ToList();
+            var trie = TopicFilterTrie<int>.Build(filters.Select((f, i) => (f, i)));
+
+            for (var t = 0; t < 50; t++)
+            {
+                var topic = RandomTopic(random);
+                var expected = filters
+                    .Select((f, i) => (f, i))
+                    .Where(p => MqttTopicFilterComparer.Compare(topic, p.f) == MqttTopicFilterCompareResult.IsMatch)
+                    .Select(p => p.i)
+                    .ToHashSet();
+
+                var collect = new Collect();
+                trie.Match(topic, ref collect);
+
+                collect.Seen.IsSupersetOf(expected).ShouldBeTrue($"topic '{topic}' missed a match; filters: {string.Join(" | ", filters)}");
+                collect.Seen
+                    .Where(i => MqttTopicFilterComparer.Compare(topic, filters[i]) == MqttTopicFilterCompareResult.IsMatch)
+                    .ToHashSet()
+                    .SetEquals(expected)
+                    .ShouldBeTrue();
+            }
+        }
+    }
+
+    [Theory]
+    [InlineData("a/#", "a", true)]          // '#' includes the parent level
+    [InlineData("a/#", "a/b/c", true)]
+    [InlineData("a/+", "a/", true)]         // '+' matches an empty level
+    [InlineData("a/+", "a", false)]
+    [InlineData("+/+", "a/b", true)]
+    [InlineData("#", "anything/at/all", true)]
+    [InlineData("a/b", "a/b", true)]
+    [InlineData("a/b", "a/c", false)]
+    public void Match_FindsTheCanonicalCases(string filter, string topic, bool expected)
+    {
+        var trie = TopicFilterTrie<int>.Build([(filter, 1)]);
+        var collect = new Collect();
+        trie.Match(topic, ref collect);
+        collect.Seen.Contains(1).ShouldBe(expected);
+    }
+
+    [Fact]
+    public void Empty_MatchesNothing()
+    {
+        var collect = new Collect();
+        TopicFilterTrie<int>.Empty.Match("a/b", ref collect);
+        collect.Seen.ShouldBeEmpty();
+    }
+
+    private static string RandomFilter(Random random)
+    {
+        var count = random.Next(1, 5);
+        var parts = new List<string>(count);
+        for (var i = 0; i < count; i++)
+        {
+            var roll = random.Next(10);
+            if (roll < 2 && i == count - 1) { parts.Add("#"); break; }
+            parts.Add(roll < 5 ? "+" : Levels[random.Next(Levels.Length)]);
+        }
+
+        return string.Join('/', parts);
+    }
+
+    private static string RandomTopic(Random random) =>
+        string.Join('/', Enumerable.Range(0, random.Next(1, 6)).Select(_ => Levels[random.Next(Levels.Length)]));
+}