diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs index 1504328a..0c8b1a9a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttSubscriptionManager.cs @@ -48,7 +48,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// accepted by the parser (warned at deploy, see MqttTagDefinitionFactory.Inspect), so /// matching uses MQTTnet's own 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 (TopicFilterTrie) whose candidates the comparer +/// confirms, so a thousand wildcard tags cost a walk over the topic's levels, not a thousand compares. /// /// /// Sparkplug seam. P1 ingests plain topics only. Sparkplug B decodes a @@ -74,6 +75,13 @@ public sealed class MqttSubscriptionManager /// public const int DefaultQos = 1; + /// + /// 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 + /// MqttBrowseSession makes. + /// + 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}]"; } + /// + /// Delivers each wildcard-trie candidate whose filter the MQTTnet comparer confirms. A ref struct + /// so the payload span and the fan-out's shared JSON parse ride along without a closure or a list. + /// + private ref struct WildcardDelivery : ITopicFilterVisitor + { + private readonly MqttSubscriptionManager _manager; + private readonly string _topic; + private readonly ReadOnlySpan _payload; + private readonly bool _retained; + private readonly DateTime _now; + private readonly bool _oversized; + private ref PayloadJson _json; + + public WildcardDelivery( + MqttSubscriptionManager manager, + string topic, + ReadOnlySpan 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); + } + } + } + /// Opaque subscription identity; the diagnostic id names the mode and its topics. /// The human-readable identity surfaced for diagnostics. private sealed record MqttSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle; @@ -1487,7 +1536,8 @@ public sealed class MqttSubscriptionManager /// /// Two topic indexes, and the difference is load-bearing. Delivery matches an /// incoming published topic, which never carries a wildcard, so it uses - /// (concrete only) plus a comparer scan of . + /// (concrete only) plus a walk whose candidates the + /// comparer confirms. /// Every path that instead reconstructs state from a subscribed filter string — /// reconnect re-subscribe and per-topic degradation — must use , which is /// keyed by the tag's authored topic including wildcard patterns, because that is what @@ -1501,12 +1551,12 @@ public sealed class MqttSubscriptionManager /// included. Keyed the same way _subscribedTopics and SUBACK outcomes are. /// /// Concrete topic → its definitions; the delivery fast path. - /// Definitions whose authored topic carries an MQTT wildcard. + /// Definitions whose authored topic carries an MQTT wildcard, by filter level. private sealed record AuthoredTable( FrozenDictionary ByRawPath, FrozenDictionary ByFilter, FrozenDictionary ByExactTopic, - MqttTagDefinition[] Wildcards) + TopicFilterTrie WildcardTrie) { public static readonly AuthoredTable Empty = Build(new Dictionary(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.Build(wildcards.Select(d => (d.Topic, d)))); } } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/TopicFilterTrie.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/TopicFilterTrie.cs new file mode 100644 index 00000000..976523f7 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/TopicFilterTrie.cs @@ -0,0 +1,152 @@ +using System.Collections.Frozen; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// Receives each candidate a match produces. +/// The value type stored against each filter. +internal interface ITopicFilterVisitor +{ + /// Called once per value whose filter may match the topic. + /// The candidate. + void Visit(T value); +} + +/// +/// An immutable trie over MQTT topic FILTERS (+ / # 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). +/// +/// +/// +/// A candidate generator, not the authority. The trie implements the plain MQTT 3.1.1 §4.7 +/// level rules — + is exactly one level (an empty one included), # is the parent level +/// and everything under it — and deliberately nothing subtler (the $-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 MqttTopicFilterComparer, 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. +/// +/// +/// Allocation-free per match. Levels are walked as spans and looked up through the +/// dictionary's span alternate lookup; candidates are pushed to a caller-supplied visitor (a +/// ref struct in production) rather than collected into a list. +/// +/// +/// The value type stored against each filter. +internal sealed class TopicFilterTrie +{ + private readonly Node _root; + + private TopicFilterTrie(Node root, int count) + { + _root = root; + Count = count; + } + + /// An empty trie. + public static TopicFilterTrie Empty { get; } = Build([]); + + /// How many values were inserted. + public int Count { get; } + + /// Builds a trie from (filter, value) pairs. + /// The filters and their values. + /// The immutable trie. + public static TopicFilterTrie 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(root.Freeze(), count); + } + + /// Pushes every value whose filter may match to . + /// The visitor; a ref struct is allowed. + /// A concrete published topic (never a filter). + /// Receives each candidate — a superset of the true matches; confirm each one. + public void Match(string topic, ref TVisitor visitor) + where TVisitor : ITopicFilterVisitor, allows ref struct + { + if (Count == 0 || topic is null) return; + Walk(_root, topic.AsSpan(), exhausted: false, ref visitor); + } + + private static void Walk(Node node, ReadOnlySpan rest, bool exhausted, ref TVisitor visitor) + where TVisitor : ITopicFilterVisitor, 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.Empty : rest[(slash + 1)..]; + var nextExhausted = slash < 0; + + if (node.Children.Count > 0 + && node.Children.GetAlternateLookup>().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 children, Node? plus, T[] exact, T[] hash) + { + public FrozenDictionary Children { get; } = children; + + public Node? Plus { get; } = plus; + + public T[] Exact { get; } = exact; + + public T[] Hash { get; } = hash; + } + + private sealed class Builder + { + public Dictionary Children { get; } = new(StringComparer.Ordinal); + + public Builder? Plus { get; set; } + + public List Exact { get; } = []; + + public List Hash { get; } = []; + + public Node Freeze() => new( + Children.ToFrozenDictionary(kv => kv.Key, kv => kv.Value.Freeze(), StringComparer.Ordinal), + Plus?.Freeze(), + [.. Exact], + [.. Hash]); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs new file mode 100644 index 00000000..f21677e1 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/TopicFilterTrieTests.cs @@ -0,0 +1,96 @@ +using MQTTnet; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// R06:M-6. The wildcard trie replaced a linear MqttTopicFilterComparer 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, $-topics and wildcard-only filters. +/// +public sealed class TopicFilterTrieTests +{ + private static readonly string[] Levels = ["a", "b", "c", "", "$SYS", "x y"]; + + private sealed class Collect : ITopicFilterVisitor + { + public HashSet 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.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.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.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(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)])); +}