From 98d962b24bbacc5abb63fd331476e3843dd1aa82 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:42:29 -0500 Subject: [PATCH 1/5] =?UTF-8?q?perf(opcua):=20#855=20=E2=80=94=20shared=20?= =?UTF-8?q?per-report=20projection=20for=20condition=20events=20under=20Lo?= =?UTF-8?q?ck?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Condition events delivered to many event items (folder and Server-object) no longer pay the SDK's per-item QueueEvent(IFilterTarget) work under the node-manager Lock: the role-permission verdict is shared per (session, identity), the #826 grant check per identity per hop, and the where clause + select-clause projection per (value-equal EventFilter, locales); each item gets its own EventFieldList over the shared read-only fields. The SDK's QueueEvent body is reproduced under the item's own lock (three private SDK members, guard-tested); anything outside that shape falls back to the SDK. The per-report duplicate check replaces the SDK's O(queue) scan. ProjectedEventDeliveryWireTests compares both paths over the wire. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- ...26-11-11-folder-event-per-item-delivery.md | 26 ++ .../OtOpcUaNodeManager.cs | 85 +++++- .../ProjectedEventDelivery.cs | 275 ++++++++++++++++++ .../ProjectedEventDeliveryWireTests.cs | 165 +++++++++++ 4 files changed, 543 insertions(+), 8 deletions(-) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs diff --git a/docs/plans/2026-11-11-folder-event-per-item-delivery.md b/docs/plans/2026-11-11-folder-event-per-item-delivery.md index 72ea22512..86a944b70 100644 --- a/docs/plans/2026-11-11-folder-event-per-item-delivery.md +++ b/docs/plans/2026-11-11-folder-event-per-item-delivery.md @@ -100,3 +100,29 @@ The per-event verdict for folder items moves from the SDK's consumer task (cache ConditionRefresh2, resubscribe), the gate race, newer-than-cut replay, ACL omission not withheld, ACL revocation during deferral, the in-flight suspended event, and two stress tests (production overload with no stall; stall + refresh + reports concurrently). + +## Follow-up: cost under `Lock` (projected delivery) + +Measured with `FolderEventDeliveryPerfTests` (opt-in, `OTOPCUA_EVENT_PERF=1`): the per-item work moved under `Lock` by +this change cost ~5–6 µs/item publishing and ~10–30 µs/item at a full queue (N = 50 stalled items: p50 0.4–1.6 ms). +Profiling split it into the SDK's role-permission check (~3–9 µs/item) and `MonitoredItem.QueueEvent(IFilterTarget)`, +whose duplicate scan over the last ≤ 1000 queued events grows with the queue (~3 µs at QueueSize 10, ~10–16 at 1000, +~16–31 at 10 000) on top of the per-item where-clause + select-clause projection. + +`ProjectedEventDelivery` / `EventDeliveryCache` (per report, hung on the condition snapshot, dropped when the walk ends): + +- the SDK role verdict once per (session, effective identity); the #826 grant check once per identity per hop; +- the where clause and the select-clause projection once per (value-equal `EventFilter`, preferred locales) — filters + are grouped once per filter object (`Utils.IsEqual`, ≤ 4096 shared groups) — and the field values shared read-only by + every item queued from it (each item still gets its own `EventFieldList`: the SDK rewrites a queued list's + `ClientHandle` on modify); +- the SDK's `QueueEvent(instance, false)` body reproduced under the item's own lock (event item; Disabled ⇒ nothing; + duplicate ⇒ nothing — asked of the report's "already queued" set instead of the queue; the filter in use; + `SetQueueOverflowIfFull`; then the public `QueueEvent(EventFieldList)`), reading three private members (`m_lock`, + `m_eventQueueHandler`, `FilterToUse`) — pinned by a guard test; anything else (members missing, a subclassed item, a + non-`EventFilter` filter, a `SupportsFilteredRetain` condition, a non-snapshot event, refresh replays) uses the SDK's + own `QueueEvent(IFilterTarget)`. + +`ProjectedEventDeliveryWireTests` runs one scenario through each path (`UseProjectedEventDelivery`) and requires +identical streams on folder and Server-object items (value-equal filters, a severity where clause, an OfType where +clause with an absent field, a Disabled item); it fails if the where clause or a field is projected differently. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 0bed75dad..88594c839 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1520,7 +1520,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 ValueTask delivery; try { - delivery = DeliverConditionEventAsync(condition, condition, e); + delivery = DeliverReportAsync(condition, e); } catch (Exception ex) { @@ -1611,7 +1611,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 try { - current = DeliverConditionEventAsync(condition, condition, next).AsTask(); + current = DeliverReportAsync(condition, next).AsTask(); } catch (Exception ex) { @@ -1689,6 +1689,58 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 await DeliverConditionEventAsync(condition, parent, e).ConfigureAwait(false); } + /// + /// One report's whole walk, then its per-report delivery cache is dropped (#855 perf): the cache only serves the + /// walk, and the snapshot it hangs on stays referenced from every item queue the event reached. + /// + private async ValueTask DeliverReportAsync(AlarmConditionState condition, IFilterTarget e) + { + try + { + await DeliverConditionEventAsync(condition, condition, e).ConfigureAwait(false); + } + finally + { + if (e is ConditionEventSnapshot snapshot) snapshot.ReleaseDeliveryCache(); + } + } + + /// + /// Queues a condition event to one item: through the report's shared projection (#855 perf) for a condition + /// snapshot, else through the SDK's own QueueEvent. + /// + private void QueueToItem(IEventMonitoredItem item, IFilterTarget e) + { + if (e is ConditionEventSnapshot snapshot) snapshot.DeliveryCache.Queue(Server, item, e, UseProjectedEventDelivery); + else item.QueueEvent(e); + } + + /// + /// Whether condition events are queued through the report's shared projection (#855 perf, default) or through the + /// SDK's own QueueEvent(IFilterTarget) per item. A test switch: the two must deliver identical events. + /// + internal bool UseProjectedEventDelivery { get; set; } = true; + + /// The SDK's role-permission verdict for one item, shared per (session, identity) within a report (#855 perf). + private bool RoleAllows(IEventMonitoredItem item, IFilterTarget e) => + e is ConditionEventSnapshot snapshot + ? snapshot.DeliveryCache.RoleAllows(item, i => !ServiceResult.IsBad(ValidateEventRolePermissions(i, e))) + : !ServiceResult.IsBad(ValidateEventRolePermissions(item, e)); + + /// + /// per effective identity within one delivery (#855 perf): the decision reads the + /// item's identity only, and a delivery's scopes are fixed. + /// + private bool IsConditionVisibleTo( + IEventMonitoredItem item, List scopes, Dictionary perIdentity) + { + if (item.EffectiveIdentity is not { } identity) return IsConditionVisibleTo(item, scopes); + if (perIdentity.TryGetValue(identity, out var cached)) return cached; + var visible = IsConditionVisibleTo(item, scopes); + perIdentity[identity] = visible; + return visible; + } + /// Marks an event whose remaining delivery will run off Lock (#855) — before the walk suspends. private static void MarkDeferred(IFilterTarget e) { @@ -1756,14 +1808,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 scopes = ConditionVisibilityScopes(live); } + var visibleByIdentity = scopes is null ? null : new Dictionary(System.Collections.Generic.ReferenceEqualityComparer.Instance); foreach (var item in monitored.EventMonitoredItems.Values) { if (item is null) continue; try { if (IsAuditEvent(e) && !IsAuditDeliverableTo(item)) continue; - if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; + if (!RoleAllows(item, e)) continue; + if (scopes is not null && !IsConditionVisibleTo(item, scopes, visibleByIdentity!)) continue; QueueConditionEventToItem(item, e, deferred, DeferredFolderDeliveryProbe); } catch (Exception ex) @@ -1877,6 +1930,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 get => _deferred; set => _deferred = value; } + + private EventDeliveryCache? _deliveryCache; + + /// + /// The report's shared delivery state (#855 perf) — projections, role verdicts, the items it was queued to. Only + /// the report's own walk uses it (sequentially, even when part of the walk runs off Lock); dropped when the + /// walk ends. + /// + public EventDeliveryCache DeliveryCache => + Volatile.Read(ref _deliveryCache) + ?? Interlocked.CompareExchange(ref _deliveryCache, new EventDeliveryCache(), null) + ?? _deliveryCache!; + + /// Drops the delivery cache once the walk is over (the snapshot stays referenced from the item queues). + public void ReleaseDeliveryCache() => Volatile.Write(ref _deliveryCache, null); } /// @@ -5841,14 +5909,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var reportingSession = (context as ISessionSystemContext)?.SessionId; var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; + var visibleByIdentity = scopes is null ? null : new Dictionary(System.Collections.Generic.ReferenceEqualityComparer.Instance); foreach (var item in _allEventsItems.Keys) { try { if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; - if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; - if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + if (scopes is not null && !IsConditionVisibleTo(item, scopes, visibleByIdentity!)) continue; + if (!RoleAllows(item, e)) continue; if (reportingSession is { IsNull: false } reportingId && item.Session?.Id is { } own && !own.Equals(reportingId)) @@ -5884,7 +5953,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 if (deferred is not null) state = EventItemStateOf(item); else if (!_eventItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot) { - item.QueueEvent(e); + QueueToItem(item, e); return; } @@ -5907,7 +5976,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 state.NewerSinceCut.Add(ce.ConditionId); if (deferred is not null) probe?.Invoke(); - item.QueueEvent(e); + QueueToItem(item, e); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs new file mode 100644 index 000000000..c15b59896 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -0,0 +1,275 @@ +using System.Linq.Expressions; +using System.Reflection; +using System.Runtime.CompilerServices; +using Opc.Ua; +using Opc.Ua.Server; +using SdkMonitoredItem = Opc.Ua.Server.MonitoredItem; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; + +/// +/// Per-report state for delivering ONE condition event to many event items (Gitea #855 perf): the event's projected +/// fields computed once per distinct (EventFilter, preferred locales) instead of once per item, the SDK's +/// role-permission verdict once per (session, identity), and the set of items it was already queued to. +/// +/// +/// +/// Why. Since #855 condition events are queued to folder items by the reporter under the node manager's +/// Lock. The SDK's MonitoredItem.QueueEvent(IFilterTarget) re-does, for every item, the where-clause +/// evaluation and the select-clause projection, and first scans the last ≤ 1000 queued events for a duplicate — +/// at a full queue that scan alone dominated (measured ~2.8 µs/item at QueueSize 10, ~10–16 µs at 1000, ~16–31 µs +/// at 10 000). The projection and verdicts are the same for every item that shares a filter and an identity. +/// +/// +/// What is reproduced, exactly. performs the SDK's QueueEvent(instance, false) +/// body (decompiled, SDK 2.0.1) under the item's own lock: event-item check; no queue handler (Disabled) ⇒ nothing; +/// duplicate ⇒ nothing; the filter in use (FilterToUse, the revised one); CanSendFilteredAlarm, which +/// without SupportsFilteredRetain on the condition is the where clause alone; SetQueueOverflowIfFull; +/// then GetEventFields (rejected select clauses ⇒ null, LocalizedText translated to the session's +/// preferred locales, Handle = the event, the item's ClientHandle) and the public +/// QueueEvent(EventFieldList). Each item gets its OWN (the SDK mutates a queued +/// list's ClientHandle on a modify); the field values are shared (immutable). +/// +/// +/// Duplicate check. The SDK compares the event INSTANCE against the last 1000 queued handles. A condition +/// snapshot reaches an item only through this per-report delivery, so "already queued in this report" is the same +/// question, asked of a set instead of the queue (it stays true after the event leaves the queue, where the SDK's +/// would not — a second delivery of one report to one item never happens on purpose). +/// +/// +/// Fallback. Anything outside that shape — the private members not found (an SDK upgrade; pinned by a guard +/// test), an item type that is not exactly the SDK's MonitoredItem, a filter in use that is not an +/// EventFilter, a condition with SupportsFilteredRetain, or an event that is not a condition snapshot — +/// goes through the SDK's own QueueEvent(IFilterTarget), unchanged. +/// +/// +internal sealed class EventDeliveryCache +{ + private readonly object _sync = new(); + private readonly HashSet _queued = new(System.Collections.Generic.ReferenceEqualityComparer.Instance); + private readonly Dictionary<(object? Session, object? Identity), bool> _roleVerdicts = new(); + private readonly Dictionary<(FilterGroup Group, string Locales), Projection> _projections = new(); + + /// The SDK role-permission verdict for an item, computed once per (session, effective identity). + /// The event item. + /// The SDK check (ValidateEventRolePermissions) for that item. + /// when the item's identity may receive the event. + /// + /// The SDK check builds an OperationContext from the item and asks ReceiveEvents on the event's type + /// and source for that context's identity and session — nothing else of the item takes part, so two items of one + /// session and identity always get the same answer for one event. + /// + public bool RoleAllows(IEventMonitoredItem item, Func compute) + { + var key = ((object?)item.Session, (object?)item.EffectiveIdentity); + lock (_sync) + { + if (_roleVerdicts.TryGetValue(key, out var cached)) return cached; + } + + var verdict = compute(item); + lock (_sync) _roleVerdicts[key] = verdict; + return verdict; + } + + /// + /// Queues the event to the item through the projected fast path when it has exactly the shape the SDK's + /// QueueEvent(IFilterTarget) would handle, else through that method itself. + /// + /// The server (namespaces, type tree, resource manager). + /// The event item. + /// The condition event. + /// False routes it through the SDK's own QueueEvent (the node manager's test switch). + public void Queue(IServerInternal server, IEventMonitoredItem item, IFilterTarget e, bool projected) + { + if (!projected + || e is not InstanceStateSnapshot { Handle: ConditionState condition } + || condition.SupportsFilteredRetain?.Value == true + || item.GetType() != typeof(SdkMonitoredItem) + || !ProjectedEventDelivery.TryQueue(server, (SdkMonitoredItem)item, e, this)) + { + item.QueueEvent(e); + } + } + + /// Marks the item as queued for this report; false when it already was (the SDK's duplicate check). + internal bool TryMarkQueued(IEventMonitoredItem item) + { + lock (_sync) return _queued.Add(item); + } + + /// The projection of the event for a filter group and locale list, computed on first use. + internal Projection ProjectionFor(FilterGroup group, string locales, Func compute) + { + lock (_sync) + { + if (_projections.TryGetValue((group, locales), out var cached)) return cached; + } + + var projection = compute(); + lock (_sync) _projections[(group, locales)] = projection; + return projection; + } + + /// The event as one filter sees it: whether the where clause passes, and the selected field values. + /// The fields are shared by every item queued from this projection: is a read-only view, and the array behind it is never written after it is built. + internal sealed record Projection(bool Passes, ArrayOf Fields); +} + +/// A set of value-equal s (one canonical representative), so they can share a projection. +internal sealed class FilterGroup(EventFilter representative) +{ + public EventFilter Representative { get; } = representative; +} + +/// The SDK internals the projected fast path reads, resolved once by reflection — see . +internal static class ProjectedEventDelivery +{ + private static readonly Func? s_itemLock = FieldGetter("m_lock"); + private static readonly Func? s_eventQueue = FieldGetter("m_eventQueueHandler"); + private static readonly Func? s_filterToUse = PropertyGetter("FilterToUse"); + + private static long s_projectedQueued; + + /// Events queued through the projected path since start (test observation: the fast path is really taken). + internal static long ProjectedQueued => Interlocked.Read(ref s_projectedQueued); + + /// Whether every SDK member the fast path needs was found (pinned by a guard test). + internal static bool Available => s_itemLock is not null && s_eventQueue is not null && s_filterToUse is not null; + + private static readonly ConditionalWeakTable s_groups = new(); + private static readonly object s_registryLock = new(); + private static readonly Dictionary> s_registry = new(StringComparer.Ordinal); + private static int s_registered; + + /// Distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing). + internal const int MaxSharedFilterGroups = 4096; + + /// + /// The SDK's QueueEvent(instance, bypassFilter: false) for one condition snapshot, with the projection taken + /// from . Returns false (nothing done) when the item is not in a shape this reproduces. + /// + internal static bool TryQueue(IServerInternal server, SdkMonitoredItem item, IFilterTarget e, EventDeliveryCache cache) + { + if (!Available || (item.MonitoredItemType & 2) == 0) return false; + + // Everything below happens under the item's own lock, as in the SDK: a concurrent ModifyMonitoredItems (filter, + // client handle, queue) or SetMonitoringMode (queue handler) cannot interleave between the decision and the queueing. + var itemLock = s_itemLock!(item); + if (itemLock is null) return false; + using (itemLock.EnterScope()) + { + var handler = s_eventQueue!(item); + if (s_filterToUse!(item) is not EventFilter filter) return false; + if (handler is null) return true; // Disabled: the SDK queues nothing (no handler) + if (!cache.TryMarkQueued(item)) return true; // the SDK's duplicate check (see EventDeliveryCache) + + var locales = item.Session?.PreferredLocales; // string[]?, exactly what the SDK hands its FilterContext and TranslateValue + var projection = cache.ProjectionFor(GroupOf(filter), LocalesKey(locales), () => Project(server, filter, locales, e)); + if (!projection.Passes || handler.SetQueueOverflowIfFull()) return true; + + Interlocked.Increment(ref s_projectedQueued); + item.QueueEvent(new EventFieldList + { + ClientHandle = item.ClientHandle, + Handle = e, + EventFields = projection.Fields, // shared, read-only (see Projection) + }); + return true; + } + } + + /// The SDK's where-clause evaluation and GetEventFields projection, for one filter and locale list. + private static EventDeliveryCache.Projection Project( + IServerInternal server, EventFilter filter, string[]? locales, IFilterTarget e) + { + var context = new FilterContext( + server.NamespaceUris, server.TypeTree, locales!, server.Telemetry); // as the SDK: Session?.PreferredLocales, null included + var passes = filter.WhereClause.Evaluate(context, e); + var clauses = filter.SelectClauses; + var fields = new Variant[clauses.Count]; + for (var i = 0; i < fields.Length; i++) + { + var clause = clauses[i]; + if (clause is null || (!clause.Validated && ServiceResult.IsBad(clause.Validate(context, 0)))) + { + fields[i] = Variant.Null; // a rejected select clause + continue; + } + + var value = e.GetAttributeValue(context, clause.TypeDefinitionId, clause.BrowsePath, clause.AttributeId, clause.ParsedIndexRange); + if (value.IsNull) + { + fields[i] = Variant.Null; + continue; + } + + if (value.TypeInfo.BuiltInType == BuiltInType.LocalizedText) + value = server.ResourceManager.TranslateValue(locales!, value); + fields[i] = value; + } + + return new EventDeliveryCache.Projection(passes, (ArrayOf)fields); + } + + /// The canonical group of a filter object: value-equal filters share one (decided once per filter object). + private static FilterGroup GroupOf(EventFilter filter) => + s_groups.GetValue(filter, static f => + { + var key = SelectKey(f); + lock (s_registryLock) + { + if (s_registry.TryGetValue(key, out var bucket)) + { + foreach (var group in bucket) + { + if (Utils.IsEqual(group.Representative, f)) return group; + } + } + + var created = new FilterGroup(f); + if (s_registered >= MaxSharedFilterGroups) return created; + s_registered++; + (bucket ??= s_registry[key] = []).Add(created); + return created; + } + }); + + /// A cheap bucket key over the select clauses (equality within a bucket is decided by Utils.IsEqual). + private static string SelectKey(EventFilter filter) + { + var parts = new List(); + foreach (var clause in filter.SelectClauses) + { + if (clause is null) + { + parts.Add("-"); + continue; + } + + var path = new List(); + foreach (var name in clause.BrowsePath) path.Add(name.ToString()); + parts.Add(string.Join("/", path) + "@" + clause.AttributeId); + } + + return string.Join("|", parts); + } + + private static string LocalesKey(string[]? locales) => + locales is null ? "\u0002" : string.Join("\u0001", locales); + + private static Func? FieldGetter(string name) where T : class + { + var field = typeof(SdkMonitoredItem).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic); + if (field is null || !typeof(T).IsAssignableFrom(field.FieldType)) return null; + var item = Expression.Parameter(typeof(SdkMonitoredItem)); + return Expression.Lambda>(Expression.Convert(Expression.Field(item, field), typeof(T)), item).Compile(); + } + + private static Func? PropertyGetter(string name) where T : class + { + var property = typeof(SdkMonitoredItem).GetProperty(name, BindingFlags.Instance | BindingFlags.NonPublic); + if (property?.GetMethod is null || !typeof(T).IsAssignableFrom(property.PropertyType)) return null; + return property.GetMethod.CreateDelegate>(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs new file mode 100644 index 000000000..904aad35b --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs @@ -0,0 +1,165 @@ +using Opc.Ua; +using Opc.Ua.Client; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Authorization; +using MonitoredItem = Opc.Ua.Client.MonitoredItem; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// #855 perf: condition events are queued to many items through one shared projection per (filter, locales) instead +/// of the SDK's per-item QueueEvent(IFilterTarget). The two paths must deliver the SAME events with the SAME +/// fields: this runs one scenario through each (the node manager's switch) and compares every item's stream — +/// folder and Server-object items, value-equal filters on distinct objects, a where clause that rejects some events, +/// a where clause on the type, a narrower select list, and a Disabled item. +/// +public sealed class ProjectedEventDeliveryWireTests +{ + private const string EqA = ServerEventRig.EqA; + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Projected_delivery_queues_exactly_what_the_sdk_queues(bool enforced) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Projected855", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var eqA = ServerEventRig.UnsNodeId(session, EqA); + var a1 = RecordedEvents.AddEventItem(subscription, eqA); + var a2 = RecordedEvents.AddEventItem(subscription, eqA); // a value-equal filter on a distinct object + var severe = AddItem(subscription, eqA, SevereOnly()); + var ofType = AddItem(subscription, eqA, OfTypeAlarm()); + var disabled = RecordedEvents.AddEventItem(subscription, eqA); + var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server); + await subscription.ApplyChangesAsync(ct); + foreach (var item in subscription.MonitoredItems) item.Status.Error.ShouldBeNull(); + await subscription.SetMonitoringModeAsync(MonitoringMode.Disabled, [disabled], ct); + + var before = ProjectedEventDelivery.ProjectedQueued; + await PhaseAsync(rig, events, a1, onServer, "p", projectedPath: true); + ProjectedEventDelivery.ProjectedQueued.ShouldBeGreaterThan(before, "the projected path must actually be taken"); + await PhaseAsync(rig, events, a1, onServer, "s", projectedPath: false); + + foreach (var item in new[] { a1, a2, severe, ofType, disabled, onServer }) + { + var p = Normalised(events.Of(item), "p"); + var s = Normalised(events.Of(item), "s"); + p.ShouldBe(s, $"{item.DisplayName}: the projected path must deliver exactly the SDK's events and fields"); + } + + Normalised(events.Of(a1), "p").Count.ShouldBe(Severities.Length); + Normalised(events.Of(severe), "p").Count.ShouldBe(Severities.Count(v => v > 800), "the where clause rejects some"); + Normalised(events.Of(ofType), "p").Count.ShouldBe(Severities.Length); + events.Of(disabled).ShouldBeEmpty("a Disabled item queues nothing on either path"); + } + + /// + /// The projected path rests on PRIVATE members of the pinned SDK's MonitoredItem (its lock, its event queue + /// handler, the filter in use). If an SDK upgrade drops one, delivery silently falls back to the SDK's per-item + /// QueueEvent (correct, slower) — this fails first, so the upgrade is looked at. + /// + [Fact] + public void The_sdk_members_the_projected_path_reads_are_still_resolvable() => + ProjectedEventDelivery.Available.ShouldBeTrue( + "MonitoredItem.m_lock / m_eventQueueHandler / FilterToUse changed — re-check ProjectedEventDelivery.TryQueue against the new SDK's QueueEvent(IFilterTarget)"); + + private static readonly ushort[] Severities = [700, 900, 300, 850, 801, 800, 999]; + + private static async Task PhaseAsync( + ServerEventRig rig, RecordedEvents events, MonitoredItem probe, MonitoredItem onServer, string prefix, bool projectedPath) + { + rig.Nm.UseProjectedEventDelivery = projectedPath; + try + { + for (var i = 0; i < Severities.Length; i++) + { + rig.Project(native: true, new AlarmConditionSnapshot( + i % 2 == 0, i % 3 == 0, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"{prefix}{i}")); + } + + var last = $"{prefix}{Severities.Length - 1}"; + await RecordedEvents.WaitUntilAsync(() => events.Has(probe, last) && events.Has(onServer, last)); + await Task.Delay(TimeSpan.FromMilliseconds(400)); + } + finally + { + rig.Nm.UseProjectedEventDelivery = true; + } + } + + /// One phase's events of an item, with the per-event identity (EventId, the phase prefix) removed. + private static List Normalised(IReadOnlyList> stream, string prefix) => + stream + .Where(f => RecordedEvents.Message(f)?.StartsWith(prefix, StringComparison.Ordinal) == true) + .Select(f => string.Join(" | ", f.ToArray()!.Select((v, i) => i switch + { + RecordedEvents.EventIdField => v.IsNull ? "null" : "id", + RecordedEvents.MessageField => RecordedEvents.Message(f)![prefix.Length..], + _ => v.ToString(), + }))) + .ToList(); + + private static MonitoredItem AddItem(Subscription subscription, NodeId source, EventFilter filter) + { + var item = new MonitoredItem(subscription.DefaultItem) + { + StartNodeId = source, + AttributeId = Attributes.EventNotifier, + Filter = filter, + QueueSize = 1000, + SamplingInterval = 0, + }; + subscription.AddItem(item); + return item; + } + + /// EventId, Message, Severity — only events with Severity > 800. + private static EventFilter SevereOnly() + { + var filter = new EventFilter(); + filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.EventId)); + filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Message)); + filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Severity)); + var where = new ContentFilter(); + where.Push(FilterOperator.GreaterThan, + Variant.FromStructure(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.Severity)) { AttributeId = Attributes.Value }), + Variant.FromStructure(new LiteralOperand(new Variant((ushort)800)))); + filter.WhereClause = where; + return filter; + } + + /// EventId, Message, Retain, a field the event does not have — OfType AlarmConditionType. + private static EventFilter OfTypeAlarm() + { + var filter = new EventFilter(); + filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.EventId)); + filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Message)); + filter.AddSelectClause(ObjectTypeIds.ConditionType, QualifiedName.From(BrowseNames.Retain)); + filter.AddSelectClause(ObjectTypeIds.LimitAlarmType, QualifiedName.From(BrowseNames.HighLimit)); + var where = new ContentFilter(); + where.Push(FilterOperator.OfType, Variant.FromStructure(new LiteralOperand(new Variant(ObjectTypeIds.AlarmConditionType)))); + filter.WhereClause = where; + return filter; + } + + private static NodeAcl KeyEquipmentGrant(string equipment) => new() + { + NodeAclId = $"acl-key-{equipment}", + ClusterId = ServerEventRig.Cluster, + Principal = ServerEventRig.KeyPrincipal, + ScopeKind = NodeAclScopeKind.Equipment, + ScopeId = $"eq-{equipment}", + PermissionFlags = NodePermissions.ReadOnly, + }; +} From ca0e73fbca22c29fcd58992601c7997b8ba59bb7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:59:22 -0500 Subject: [PATCH 2/5] =?UTF-8?q?fix(opcua):=20#855=20perf=20review=201=20?= =?UTF-8?q?=E2=80=94=20groups=20evaluate=20their=20own=20filter=20clone;?= =?UTF-8?q?=20lazy=20fields;=20tighter=20equivalence=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A filter group owns a deep clone of its first filter, validated once and evaluated under the group's lock: no item-owned filter is mutated or raced (Validate mutates a select clause). - The projection is computed in the SDK's order: where clause first, fields only for an item that will queue. - Unambiguous locale key; counter for filters given a private group past the registry cap; DeliveryCache getter returns its own allocation. - UseProjectedEventDelivery=false also asks the role and grant checks per item, so the equivalence test compares against the pre-optimisation path. - Equivalence test: same select list under three where clauses, a filter modified mid-run, a session with other locales, an ungranted anonymous Server-object item, per-rig counter; plus a small overflowing queue with DiscardOldest true and false. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- ...26-11-11-folder-event-per-item-delivery.md | 18 +- .../OtOpcUaNodeManager.cs | 35 +++- .../ProjectedEventDelivery.cs | 185 ++++++++++++----- .../ProjectedEventDeliveryWireTests.cs | 195 +++++++++++++----- .../ServerEventRig.cs | 4 +- 5 files changed, 317 insertions(+), 120 deletions(-) diff --git a/docs/plans/2026-11-11-folder-event-per-item-delivery.md b/docs/plans/2026-11-11-folder-event-per-item-delivery.md index 86a944b70..5d3fbf678 100644 --- a/docs/plans/2026-11-11-folder-event-per-item-delivery.md +++ b/docs/plans/2026-11-11-folder-event-per-item-delivery.md @@ -112,9 +112,10 @@ whose duplicate scan over the last ≤ 1000 queued events grows with the queue ( `ProjectedEventDelivery` / `EventDeliveryCache` (per report, hung on the condition snapshot, dropped when the walk ends): - the SDK role verdict once per (session, effective identity); the #826 grant check once per identity per hop; -- the where clause and the select-clause projection once per (value-equal `EventFilter`, preferred locales) — filters - are grouped once per filter object (`Utils.IsEqual`, ≤ 4096 shared groups) — and the field values shared read-only by - every item queued from it (each item still gets its own `EventFieldList`: the SDK rewrites a queued list's +- the where clause and the select-clause projection once per (value-equal `EventFilter`, preferred locales), in the + SDK's order (fields only for an item that will queue) — filters are grouped once per filter object (`Utils.IsEqual`, + ≤ 4096 shared groups, then private groups, counted); a group evaluates its OWN deep clone, validated once, under its own + lock, never an item's filter object — and the field values shared read-only by every item queued from it (each item still gets its own `EventFieldList`: the SDK rewrites a queued list's `ClientHandle` on modify); - the SDK's `QueueEvent(instance, false)` body reproduced under the item's own lock (event item; Disabled ⇒ nothing; duplicate ⇒ nothing — asked of the report's "already queued" set instead of the queue; the filter in use; @@ -123,6 +124,13 @@ whose duplicate scan over the last ≤ 1000 queued events grows with the queue ( non-`EventFilter` filter, a `SupportsFilteredRetain` condition, a non-snapshot event, refresh replays) uses the SDK's own `QueueEvent(IFilterTarget)`. -`ProjectedEventDeliveryWireTests` runs one scenario through each path (`UseProjectedEventDelivery`) and requires +`ProjectedEventDeliveryWireTests` runs one scenario through each path (`UseProjectedEventDelivery`, which also turns the +verdict sharing off) and requires identical streams on folder and Server-object items (value-equal filters, a severity where clause, an OfType where -clause with an absent field, a Disabled item); it fails if the where clause or a field is projected differently. +clause with an absent field, the same select list under three where clauses, a filter modified mid-run, a second session +with other locales, an ungranted anonymous Server-object item, a Disabled item) plus a small overflowing queue with +DiscardOldest true and false; it fails if the where clause, a field or the grouping is wrong. + +Residual: at a full queue the SDK's own `EventMonitoredItemQueue` still does `List.RemoveAt(0)` per item (memmove of the +whole queue) — ~1–10 µs/item at QueueSize 1000–10 000; replacing the queue (`IMonitoredItemQueueFactory`) is a separate +change. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 88594c839..a4284036a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1711,19 +1711,31 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// private void QueueToItem(IEventMonitoredItem item, IFilterTarget e) { - if (e is ConditionEventSnapshot snapshot) snapshot.DeliveryCache.Queue(Server, item, e, UseProjectedEventDelivery); - else item.QueueEvent(e); + if (e is ConditionEventSnapshot snapshot) + { + if (snapshot.DeliveryCache.Queue(Server, item, e, UseProjectedEventDelivery)) Interlocked.Increment(ref _projectedDeliveries); + } + else + { + item.QueueEvent(e); + } } + private long _projectedDeliveries; + + /// Items this node manager served through the projected path (test observation: it is really taken). + internal long ProjectedDeliveries => Interlocked.Read(ref _projectedDeliveries); + /// /// Whether condition events are queued through the report's shared projection (#855 perf, default) or through the - /// SDK's own QueueEvent(IFilterTarget) per item. A test switch: the two must deliver identical events. + /// SDK's own QueueEvent(IFilterTarget) per item, with the role and grant checks asked per item (the + /// pre-optimisation path). A test switch: the two must deliver identical events. /// internal bool UseProjectedEventDelivery { get; set; } = true; /// The SDK's role-permission verdict for one item, shared per (session, identity) within a report (#855 perf). private bool RoleAllows(IEventMonitoredItem item, IFilterTarget e) => - e is ConditionEventSnapshot snapshot + UseProjectedEventDelivery && e is ConditionEventSnapshot snapshot ? snapshot.DeliveryCache.RoleAllows(item, i => !ServiceResult.IsBad(ValidateEventRolePermissions(i, e))) : !ServiceResult.IsBad(ValidateEventRolePermissions(item, e)); @@ -1734,7 +1746,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 private bool IsConditionVisibleTo( IEventMonitoredItem item, List scopes, Dictionary perIdentity) { - if (item.EffectiveIdentity is not { } identity) return IsConditionVisibleTo(item, scopes); + if (!UseProjectedEventDelivery || item.EffectiveIdentity is not { } identity) return IsConditionVisibleTo(item, scopes); if (perIdentity.TryGetValue(identity, out var cached)) return cached; var visible = IsConditionVisibleTo(item, scopes); perIdentity[identity] = visible; @@ -1938,10 +1950,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// the report's own walk uses it (sequentially, even when part of the walk runs off Lock); dropped when the /// walk ends. /// - public EventDeliveryCache DeliveryCache => - Volatile.Read(ref _deliveryCache) - ?? Interlocked.CompareExchange(ref _deliveryCache, new EventDeliveryCache(), null) - ?? _deliveryCache!; + public EventDeliveryCache DeliveryCache + { + get + { + if (Volatile.Read(ref _deliveryCache) is { } existing) return existing; + var created = new EventDeliveryCache(); + return Interlocked.CompareExchange(ref _deliveryCache, created, null) ?? created; + } + } /// Drops the delivery cache once the walk is over (the snapshot stays referenced from the item queues). public void ReleaseDeliveryCache() => Volatile.Write(ref _deliveryCache, null); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs index c15b59896..877cc57f1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -57,7 +57,8 @@ internal sealed class EventDeliveryCache /// /// The SDK check builds an OperationContext from the item and asks ReceiveEvents on the event's type /// and source for that context's identity and session — nothing else of the item takes part, so two items of one - /// session and identity always get the same answer for one event. + /// session and identity always get the same answer for one event. Held for the report's walk: on a walk that + /// suspends (none of ours does) a RolePermissions change between two hops lands one event later. /// public bool RoleAllows(IEventMonitoredItem item, Func compute) { @@ -80,16 +81,20 @@ internal sealed class EventDeliveryCache /// The event item. /// The condition event. /// False routes it through the SDK's own QueueEvent (the node manager's test switch). - public void Queue(IServerInternal server, IEventMonitoredItem item, IFilterTarget e, bool projected) + /// True when the projected path handled the item (queued, or decided not to queue exactly as the SDK would). + public bool Queue(IServerInternal server, IEventMonitoredItem item, IFilterTarget e, bool projected) { - if (!projected - || e is not InstanceStateSnapshot { Handle: ConditionState condition } - || condition.SupportsFilteredRetain?.Value == true - || item.GetType() != typeof(SdkMonitoredItem) - || !ProjectedEventDelivery.TryQueue(server, (SdkMonitoredItem)item, e, this)) + if (projected + && e is InstanceStateSnapshot { Handle: ConditionState condition } + && condition.SupportsFilteredRetain?.Value != true + && item.GetType() == typeof(SdkMonitoredItem) + && ProjectedEventDelivery.TryQueue(server, (SdkMonitoredItem)item, e, this)) { - item.QueueEvent(e); + return true; } + + item.QueueEvent(e); + return false; } /// Marks the item as queued for this report; false when it already was (the SDK's duplicate check). @@ -98,28 +103,67 @@ internal sealed class EventDeliveryCache lock (_sync) return _queued.Add(item); } - /// The projection of the event for a filter group and locale list, computed on first use. - internal Projection ProjectionFor(FilterGroup group, string locales, Func compute) + /// The projection of the event for a filter group and locale list, created on first use. + internal Projection ProjectionFor(FilterGroup group, string locales, Func create) { lock (_sync) { if (_projections.TryGetValue((group, locales), out var cached)) return cached; + var projection = create(); + _projections[(group, locales)] = projection; + return projection; } - - var projection = compute(); - lock (_sync) _projections[(group, locales)] = projection; - return projection; } - /// The event as one filter sees it: whether the where clause passes, and the selected field values. - /// The fields are shared by every item queued from this projection: is a read-only view, and the array behind it is never written after it is built. - internal sealed record Projection(bool Passes, ArrayOf Fields); + /// + /// The event as one filter group sees it, each part computed once and only when needed, in the SDK's order: the + /// where clause first, the select-clause fields only for an item that will actually queue the event. + /// + /// + /// The fields are shared by every item queued from this projection: is a read-only view, + /// and the array behind it is never written after it is built. Only the report's own walk uses a projection, one + /// step at a time; the lazies are thread-safe regardless. + /// + internal sealed class Projection(Func passes, Func> fields) + { + private readonly Lazy _passes = new(passes, LazyThreadSafetyMode.ExecutionAndPublication); + private readonly Lazy> _fields = new(fields, LazyThreadSafetyMode.ExecutionAndPublication); + + public bool Passes => _passes.Value; + + public ArrayOf Fields => _fields.Value; + } } -/// A set of value-equal s (one canonical representative), so they can share a projection. -internal sealed class FilterGroup(EventFilter representative) +/// +/// A set of value-equal s sharing one projection. The group owns a deep CLONE of the first +/// member (validated once, here), so evaluating it never touches — or races — a filter object an item owns (#855 perf +/// review: Validate mutates a select clause). +/// +internal sealed class FilterGroup { - public EventFilter Representative { get; } = representative; + public FilterGroup(EventFilter filter, IServerInternal server) + { + Representative = (EventFilter)CoreUtils.Clone(filter)!; + // The SDK's GetRejectedSelectClauses: validation depends on the namespaces and type tree, never on the locales. + var context = new FilterContext(server.NamespaceUris, server.TypeTree, server.Telemetry); + var clauses = Representative.SelectClauses; + Rejected = new bool[clauses.Count]; + for (var i = 0; i < Rejected.Length; i++) + { + var clause = clauses[i]; + Rejected[i] = clause is null || (!clause.Validated && ServiceResult.IsBad(clause.Validate(context, 0))); + } + } + + /// The group's own copy of the filter; evaluated only under . + public EventFilter Representative { get; } + + /// Per select clause: rejected (always projected as null), as the SDK decides once per filter. + public bool[] Rejected { get; } + + /// Serialises evaluation of (two reports may project it at once). + public object Gate { get; } = new(); } /// The SDK internals the projected fast path reads, resolved once by reflection — see . @@ -165,7 +209,9 @@ internal static class ProjectedEventDelivery if (!cache.TryMarkQueued(item)) return true; // the SDK's duplicate check (see EventDeliveryCache) var locales = item.Session?.PreferredLocales; // string[]?, exactly what the SDK hands its FilterContext and TranslateValue - var projection = cache.ProjectionFor(GroupOf(filter), LocalesKey(locales), () => Project(server, filter, locales, e)); + var group = GroupOf(filter, server); + var projection = cache.ProjectionFor(group, LocalesKey(locales), () => new EventDeliveryCache.Projection( + () => Passes(server, group, locales, e), () => Project(server, group, locales, e))); if (!projection.Passes || handler.SetQueueOverflowIfFull()) return true; Interlocked.Increment(ref s_projectedQueued); @@ -179,42 +225,55 @@ internal static class ProjectedEventDelivery } } - /// The SDK's where-clause evaluation and GetEventFields projection, for one filter and locale list. - private static EventDeliveryCache.Projection Project( - IServerInternal server, EventFilter filter, string[]? locales, IFilterTarget e) + /// The SDK's CanSendFilteredAlarm without filtered retain: the where clause, for one group and locale list. + private static bool Passes(IServerInternal server, FilterGroup group, string[]? locales, IFilterTarget e) { - var context = new FilterContext( - server.NamespaceUris, server.TypeTree, locales!, server.Telemetry); // as the SDK: Session?.PreferredLocales, null included - var passes = filter.WhereClause.Evaluate(context, e); - var clauses = filter.SelectClauses; - var fields = new Variant[clauses.Count]; - for (var i = 0; i < fields.Length; i++) + lock (group.Gate) { - var clause = clauses[i]; - if (clause is null || (!clause.Validated && ServiceResult.IsBad(clause.Validate(context, 0)))) - { - fields[i] = Variant.Null; // a rejected select clause - continue; - } - - var value = e.GetAttributeValue(context, clause.TypeDefinitionId, clause.BrowsePath, clause.AttributeId, clause.ParsedIndexRange); - if (value.IsNull) - { - fields[i] = Variant.Null; - continue; - } - - if (value.TypeInfo.BuiltInType == BuiltInType.LocalizedText) - value = server.ResourceManager.TranslateValue(locales!, value); - fields[i] = value; + return group.Representative.WhereClause.Evaluate(ContextFor(server, locales), e); } - - return new EventDeliveryCache.Projection(passes, (ArrayOf)fields); } + /// The SDK's GetEventFields values, for one group and locale list. + private static ArrayOf Project(IServerInternal server, FilterGroup group, string[]? locales, IFilterTarget e) + { + lock (group.Gate) + { + var context = ContextFor(server, locales); + var clauses = group.Representative.SelectClauses; + var fields = new Variant[clauses.Count]; + for (var i = 0; i < fields.Length; i++) + { + if (group.Rejected[i]) + { + fields[i] = Variant.Null; + continue; + } + + var clause = clauses[i]; + var value = e.GetAttributeValue(context, clause.TypeDefinitionId, clause.BrowsePath, clause.AttributeId, clause.ParsedIndexRange); + if (value.IsNull) + { + fields[i] = Variant.Null; + continue; + } + + if (value.TypeInfo.BuiltInType == BuiltInType.LocalizedText) + value = server.ResourceManager.TranslateValue(locales!, value); + fields[i] = value; + } + + return fields; + } + } + + /// The SDK's per-item FilterContext: the server's namespaces and type tree, the session's locales (null included). + private static FilterContext ContextFor(IServerInternal server, string[]? locales) => + new(server.NamespaceUris, server.TypeTree, locales!, server.Telemetry); + /// The canonical group of a filter object: value-equal filters share one (decided once per filter object). - private static FilterGroup GroupOf(EventFilter filter) => - s_groups.GetValue(filter, static f => + private static FilterGroup GroupOf(EventFilter filter, IServerInternal server) => + s_groups.GetValue(filter, f => { var key = SelectKey(f); lock (s_registryLock) @@ -227,14 +286,24 @@ internal static class ProjectedEventDelivery } } - var created = new FilterGroup(f); - if (s_registered >= MaxSharedFilterGroups) return created; + var created = new FilterGroup(f, server); + if (s_registered >= MaxSharedFilterGroups) + { + Interlocked.Increment(ref s_unsharedGroups); + return created; + } + s_registered++; (bucket ??= s_registry[key] = []).Add(created); return created; } }); + private static long s_unsharedGroups; + + /// Filters given a private group because the shared registry is full (observability; sharing stops, delivery is unchanged). + internal static long UnsharedFilterGroups => Interlocked.Read(ref s_unsharedGroups); + /// A cheap bucket key over the select clauses (equality within a bucket is decided by Utils.IsEqual). private static string SelectKey(EventFilter filter) { @@ -255,8 +324,14 @@ internal static class ProjectedEventDelivery return string.Join("|", parts); } - private static string LocalesKey(string[]? locales) => - locales is null ? "\u0002" : string.Join("\u0001", locales); + /// An unambiguous key for a locale list (length-prefixed entries; null distinct from empty). + private static string LocalesKey(string[]? locales) + { + if (locales is null) return "null"; + var key = new System.Text.StringBuilder().Append(locales.Length).Append(':'); + foreach (var locale in locales) key.Append(locale?.Length ?? -1).Append(':').Append(locale).Append(';'); + return key.ToString(); + } private static Func? FieldGetter(string name) where T : class { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs index 904aad35b..8cc860272 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs @@ -11,16 +11,24 @@ using MonitoredItem = Opc.Ua.Client.MonitoredItem; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// -/// #855 perf: condition events are queued to many items through one shared projection per (filter, locales) instead -/// of the SDK's per-item QueueEvent(IFilterTarget). The two paths must deliver the SAME events with the SAME -/// fields: this runs one scenario through each (the node manager's switch) and compares every item's stream — -/// folder and Server-object items, value-equal filters on distinct objects, a where clause that rejects some events, -/// a where clause on the type, a narrower select list, and a Disabled item. +/// #855 perf: condition events are queued to many items through one shared projection per (filter, locales), with the +/// role / grant verdicts shared per identity, instead of the SDK's per-item QueueEvent(IFilterTarget) and +/// per-item checks. The two paths must deliver the SAME events with the SAME fields: these run one scenario through +/// each (the node manager's UseProjectedEventDelivery switch, which also turns the verdict sharing off) and +/// compare every item's stream. /// public sealed class ProjectedEventDeliveryWireTests { private const string EqA = ServerEventRig.EqA; + private static readonly ushort[] Severities = [700, 900, 300, 850, 801, 800, 999, 950]; + + /// + /// Folder and Server-object items; value-equal filters on distinct objects; three filters with the SAME select + /// list and different where clauses (they share a registry bucket, never a projection); an OfType where clause + /// with a field the event lacks; a Disabled item; a second session with other preferred locales; an anonymous + /// session (no grant under the ACL) on the Server object; and a filter modified half-way through each phase. + /// [Theory] [InlineData(false)] [InlineData(true)] @@ -38,30 +46,138 @@ public sealed class ProjectedEventDeliveryWireTests var eqA = ServerEventRig.UnsNodeId(session, EqA); var a1 = RecordedEvents.AddEventItem(subscription, eqA); var a2 = RecordedEvents.AddEventItem(subscription, eqA); // a value-equal filter on a distinct object - var severe = AddItem(subscription, eqA, SevereOnly()); - var ofType = AddItem(subscription, eqA, OfTypeAlarm()); + var severe = AddItem(subscription, eqA, Narrow(severityAbove: 800)); + var severer = AddItem(subscription, eqA, Narrow(severityAbove: 900)); + var unfiltered = AddItem(subscription, eqA, Narrow(severityAbove: null)); + var ofType = AddItem(subscription, eqA, OfTypeAlarm(alternative: false)); var disabled = RecordedEvents.AddEventItem(subscription, eqA); var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server); await subscription.ApplyChangesAsync(ct); foreach (var item in subscription.MonitoredItems) item.Status.Error.ShouldBeNull(); await subscription.SetMonitoringModeAsync(MonitoringMode.Disabled, [disabled], ct); - var before = ProjectedEventDelivery.ProjectedQueued; - await PhaseAsync(rig, events, a1, onServer, "p", projectedPath: true); - ProjectedEventDelivery.ProjectedQueued.ShouldBeGreaterThan(before, "the projected path must actually be taken"); - await PhaseAsync(rig, events, a1, onServer, "s", projectedPath: false); + using var german = await rig.ConnectAsync(anonymous: false, ct, preferredLocales: ["de-DE", "fr"]); + var germanEvents = new RecordedEvents(); + var germanSubscription = await RecordedEvents.SubscribeAsync(german, germanEvents, ct); + var g1 = RecordedEvents.AddEventItem(germanSubscription, ServerEventRig.UnsNodeId(german, EqA)); + await germanSubscription.ApplyChangesAsync(ct); + g1.Status.Error.ShouldBeNull(); - foreach (var item in new[] { a1, a2, severe, ofType, disabled, onServer }) + using var anonymous = await rig.ConnectAsync(anonymous: true, ct); + var anonEvents = new RecordedEvents(); + var anonSubscription = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct); + var anonServer = RecordedEvents.AddEventItem(anonSubscription, ObjectIds.Server); + await anonSubscription.ApplyChangesAsync(ct); + anonServer.Status.Error.ShouldBeNull(); + + async Task PhaseAsync(string prefix, bool projectedPath) { - var p = Normalised(events.Of(item), "p"); - var s = Normalised(events.Of(item), "s"); - p.ShouldBe(s, $"{item.DisplayName}: the projected path must deliver exactly the SDK's events and fields"); + rig.Nm.UseProjectedEventDelivery = projectedPath; + try + { + for (var i = 0; i < Severities.Length; i++) + { + if (i == Severities.Length / 2) + { + ofType.Filter = OfTypeAlarm(alternative: true); // ModifyMonitoredItems mid-phase + await subscription.ApplyChangesAsync(ct); + ofType.Status.Error.ShouldBeNull(); + } + + rig.Project(native: true, new AlarmConditionSnapshot( + i % 2 == 0, i % 3 == 0, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"{prefix}{i}")); + } + + var last = $"{prefix}{Severities.Length - 1}"; + await RecordedEvents.WaitUntilAsync(() => + events.Has(a1, last) && events.Has(onServer, last) && germanEvents.Has(g1, last)); + await Task.Delay(TimeSpan.FromMilliseconds(400), ct); + } + finally + { + rig.Nm.UseProjectedEventDelivery = true; + ofType.Filter = OfTypeAlarm(alternative: false); + await subscription.ApplyChangesAsync(ct); + } + } + + var before = rig.Nm.ProjectedDeliveries; + await PhaseAsync("p", projectedPath: true); + rig.Nm.ProjectedDeliveries.ShouldBeGreaterThan(before, "the projected path must actually be taken by this rig"); + var afterProjected = rig.Nm.ProjectedDeliveries; + await PhaseAsync("s", projectedPath: false); + rig.Nm.ProjectedDeliveries.ShouldBe(afterProjected, "the SDK phase must not use the projected path"); + + var streams = new (RecordedEvents Events, MonitoredItem Item)[] + { + (events, a1), (events, a2), (events, severe), (events, severer), (events, unfiltered), (events, ofType), + (events, disabled), (events, onServer), (germanEvents, g1), (anonEvents, anonServer), + }; + foreach (var (recorded, item) in streams) + { + Normalised(recorded.Of(item), "p").ShouldBe( + Normalised(recorded.Of(item), "s"), + $"{item.DisplayName}: the projected path must deliver exactly the SDK's events and fields"); } Normalised(events.Of(a1), "p").Count.ShouldBe(Severities.Length); - Normalised(events.Of(severe), "p").Count.ShouldBe(Severities.Count(v => v > 800), "the where clause rejects some"); - Normalised(events.Of(ofType), "p").Count.ShouldBe(Severities.Length); + Normalised(events.Of(severe), "p").Count.ShouldBe(Severities.Count(v => v > 800)); + Normalised(events.Of(severer), "p").Count.ShouldBe(Severities.Count(v => v > 900), "same select list, other where clause"); + Normalised(events.Of(unfiltered), "p").Count.ShouldBe(Severities.Length); events.Of(disabled).ShouldBeEmpty("a Disabled item queues nothing on either path"); + if (enforced) anonEvents.Of(anonServer).Count(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)).ShouldBe(0); + else Normalised(anonEvents.Of(anonServer), "p").Count.ShouldBe(Severities.Length); + } + + /// + /// A small queue overflowing while the client does not publish — DiscardOldest true and false — delivers the same + /// survivors and the same EventQueueOverflow event, in the same position, on both paths (each on a fresh server). + /// + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task An_overflowing_small_queue_is_identical_on_both_paths(bool discardOldest) + { + var projected = await OverflowScenarioAsync(discardOldest, projectedPath: true); + var sdk = await OverflowScenarioAsync(discardOldest, projectedPath: false); + projected.ShouldBe(sdk); + projected.ShouldContain(f => f.StartsWith("overflow", StringComparison.Ordinal), "the queue really overflowed"); + } + + private static async Task> OverflowScenarioAsync(bool discardOldest, bool projectedPath) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync($"Projected855Overflow{projectedPath}", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + rig.Nm.UseProjectedEventDelivery = projectedPath; + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var item = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + item.QueueSize = 3; + item.DiscardOldest = discardOldest; + await subscription.ApplyChangesAsync(ct); + item.Status.Error.ShouldBeNull(); + await subscription.SetPublishingModeAsync(false, ct); + + for (var i = 0; i < Severities.Length; i++) + { + rig.Project(native: true, new AlarmConditionSnapshot( + true, false, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"o{i}")); + } + + await subscription.SetPublishingModeAsync(true, ct); + await RecordedEvents.WaitUntilAsync(() => + events.Of(item).Any(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.EventQueueOverflowEventType)); + await Task.Delay(TimeSpan.FromMilliseconds(400), ct); + return events.Of(item) + .Select(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.EventQueueOverflowEventType + ? "overflow" + : $"{RecordedEvents.Message(f)} sev={f[RecordedEvents.SeverityField]}") + .ToList(); } /// @@ -74,30 +190,6 @@ public sealed class ProjectedEventDeliveryWireTests ProjectedEventDelivery.Available.ShouldBeTrue( "MonitoredItem.m_lock / m_eventQueueHandler / FilterToUse changed — re-check ProjectedEventDelivery.TryQueue against the new SDK's QueueEvent(IFilterTarget)"); - private static readonly ushort[] Severities = [700, 900, 300, 850, 801, 800, 999]; - - private static async Task PhaseAsync( - ServerEventRig rig, RecordedEvents events, MonitoredItem probe, MonitoredItem onServer, string prefix, bool projectedPath) - { - rig.Nm.UseProjectedEventDelivery = projectedPath; - try - { - for (var i = 0; i < Severities.Length; i++) - { - rig.Project(native: true, new AlarmConditionSnapshot( - i % 2 == 0, i % 3 == 0, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"{prefix}{i}")); - } - - var last = $"{prefix}{Severities.Length - 1}"; - await RecordedEvents.WaitUntilAsync(() => events.Has(probe, last) && events.Has(onServer, last)); - await Task.Delay(TimeSpan.FromMilliseconds(400)); - } - finally - { - rig.Nm.UseProjectedEventDelivery = true; - } - } - /// One phase's events of an item, with the per-event identity (EventId, the phase prefix) removed. private static List Normalised(IReadOnlyList> stream, string prefix) => stream @@ -124,29 +216,34 @@ public sealed class ProjectedEventDeliveryWireTests return item; } - /// EventId, Message, Severity — only events with Severity > 800. - private static EventFilter SevereOnly() + /// EventId, Message, Severity — with the optional where clause Severity > . + private static EventFilter Narrow(int? severityAbove) { var filter = new EventFilter(); filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.EventId)); filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Message)); filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Severity)); var where = new ContentFilter(); - where.Push(FilterOperator.GreaterThan, - Variant.FromStructure(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.Severity)) { AttributeId = Attributes.Value }), - Variant.FromStructure(new LiteralOperand(new Variant((ushort)800)))); + if (severityAbove is { } floor) + { + where.Push(FilterOperator.GreaterThan, + Variant.FromStructure(new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.Severity)) { AttributeId = Attributes.Value }), + Variant.FromStructure(new LiteralOperand(new Variant((ushort)floor)))); + } + filter.WhereClause = where; return filter; } - /// EventId, Message, Retain, a field the event does not have — OfType AlarmConditionType. - private static EventFilter OfTypeAlarm() + /// OfType AlarmConditionType: EventId, Message, Retain, a field the event lacks (or, alternatively, Severity). + private static EventFilter OfTypeAlarm(bool alternative) { var filter = new EventFilter(); filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.EventId)); filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Message)); filter.AddSelectClause(ObjectTypeIds.ConditionType, QualifiedName.From(BrowseNames.Retain)); - filter.AddSelectClause(ObjectTypeIds.LimitAlarmType, QualifiedName.From(BrowseNames.HighLimit)); + if (alternative) filter.AddSelectClause(new NodeId(ObjectTypes.BaseEventType), QualifiedName.From(BrowseNames.Severity)); + else filter.AddSelectClause(ObjectTypeIds.LimitAlarmType, QualifiedName.From(BrowseNames.HighLimit)); var where = new ContentFilter(); where.Push(FilterOperator.OfType, Variant.FromStructure(new LiteralOperand(new Variant(ObjectTypeIds.AlarmConditionType)))); filter.WhereClause = where; diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs index 44ae8279e..da0ad7520 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ServerEventRig.cs @@ -298,7 +298,7 @@ internal sealed class ServerEventRig : IAsyncDisposable .Invoke(server, [on]); } - public async Task ConnectAsync(bool anonymous, CancellationToken ct, bool encrypted = false) + public async Task ConnectAsync(bool anonymous, CancellationToken ct, bool encrypted = false, string[]? preferredLocales = null) { var pki = TestClientSecurity.AllocatePkiRoot(); var security = TestClientSecurity.Build(pki); @@ -337,7 +337,7 @@ internal sealed class ServerEventRig : IAsyncDisposable sessionName: "ServerEventRig", sessionTimeout: 60_000, identity: identity, - preferredLocales: default, + preferredLocales: preferredLocales is { } locales ? (ArrayOf)locales : default, ct: ct); } From b77bb2ded179aebbec52aed60d75776058d0ec4a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 14:10:02 -0500 Subject: [PATCH 3/5] =?UTF-8?q?fix(opcua):=20#855=20perf=20review=202=20?= =?UTF-8?q?=E2=80=94=20where=20clause=20in=20the=20group=20bucket=20key;?= =?UTF-8?q?=20pin=20dedupe,=20modify=20and=20keys?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Filter-group buckets include the where clause's (value-based) hash, so per-source filters sharing a select list are not all scanned under Lock. - Locale keys are cached per session array. - Tests: a notifier reached twice in one walk (Line above two wired equipment) gets each event once on both paths (fails without the per-report set); queued events re-projected by a modify are identical on both paths; locale keys unambiguous and where clauses hash by value. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../ProjectedEventDelivery.cs | 20 ++- .../ProjectedEventDeliveryWireTests.cs | 118 ++++++++++++++++++ 2 files changed, 133 insertions(+), 5 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs index 877cc57f1..5ef544255 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -33,8 +33,11 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// /// Duplicate check. The SDK compares the event INSTANCE against the last 1000 queued handles. A condition /// snapshot reaches an item only through this per-report delivery, so "already queued in this report" is the same -/// question, asked of a set instead of the queue (it stays true after the event leaves the queue, where the SDK's -/// would not — a second delivery of one report to one item never happens on purpose). +/// question, asked of a set instead of the queue. It matters: a walk can reach one notifier twice (a Line folder above +/// two equipment folders the alarm is wired into), and the SDK's scan suppressed that second copy too. The set +/// stays true after the event has left the queue, where the SDK's scan would not — a second visit within one walk +/// happens long before anything is published; at a full DiscardOldest=false queue the SDK also counted that second +/// visit as one more discarded event, this counts it once. /// /// /// Fallback. Anything outside that shape — the private members not found (an SDK upgrade; pinned by a guard @@ -275,7 +278,9 @@ internal static class ProjectedEventDelivery private static FilterGroup GroupOf(EventFilter filter, IServerInternal server) => s_groups.GetValue(filter, f => { - var key = SelectKey(f); + // The where clause takes part in the bucket (its hash is value-based), so filters that share a select list but + // differ in their where clauses (per-source filters) do not all land in one bucket scanned under Lock. + var key = SelectKey(f) + "#" + f.WhereClause?.GetHashCode().ToString(System.Globalization.CultureInfo.InvariantCulture); lock (s_registryLock) { if (s_registry.TryGetValue(key, out var bucket)) @@ -325,9 +330,14 @@ internal static class ProjectedEventDelivery } /// An unambiguous key for a locale list (length-prefixed entries; null distinct from empty). - private static string LocalesKey(string[]? locales) + private static readonly ConditionalWeakTable s_localeKeys = new(); + + /// An unambiguous key for a locale list, cached per array (a session's array is fixed for its lifetime). + internal static string LocalesKey(string[]? locales) => + locales is null ? "null" : s_localeKeys.GetValue(locales, static l => BuildLocalesKey(l)); + + private static string BuildLocalesKey(string[] locales) { - if (locales is null) return "null"; var key = new System.Text.StringBuilder().Append(locales.Length).Append(':'); foreach (var locale in locales) key.Append(locale?.Length ?? -1).Append(':').Append(locale).Append(';'); return key.ToString(); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs index 8cc860272..617539a6f 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs @@ -5,6 +5,7 @@ using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Authorization; using MonitoredItem = Opc.Ua.Client.MonitoredItem; @@ -180,6 +181,123 @@ public sealed class ProjectedEventDeliveryWireTests .ToList(); } + /// + /// A walk that reaches one notifier twice — a Line folder above two equipment folders the native alarm is wired into + /// (alarm → E1 → Line, alarm → E2 → Line) — must queue each event to the Line's item once, as the SDK's duplicate + /// scan did. Both paths, compared; no EventId may repeat. + /// + [Fact] + public async Task A_notifier_reached_twice_in_one_walk_gets_each_event_once_on_both_paths() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Projected855Twice", encrypted: false, ct); + rig.KeyRoles = []; + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + var sink = rig.Sink; + sink.EnsureFolder(ServerEventRig.RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw); + sink.MaterialiseAlarmCondition(ServerEventRig.RawAlarmPath, ServerEventRig.RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + sink.EnsureFolder("L1", parentNodeId: null, displayName: "L1", realm: AddressSpaceRealm.Uns); + sink.EnsureFolder("E1", parentNodeId: "L1", displayName: "E1", realm: AddressSpaceRealm.Uns); + sink.EnsureFolder("E2", parentNodeId: "L1", displayName: "E2", realm: AddressSpaceRealm.Uns); + // A condition directly under L1 makes L1 an event notifier (only folders a condition hangs under, or is wired into, are). + sink.MaterialiseAlarmCondition("line-alarm", "L1", "line", "AlarmCondition", 500, AddressSpaceRealm.Uns, isNative: false); + sink.WireAlarmNotifiers(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw, ["E1", "E2"], AddressSpaceRealm.Uns); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onLine = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, "L1")); + var onE1 = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, "E1")); + await subscription.ApplyChangesAsync(ct); + onLine.Status.Error.ShouldBeNull(); + + foreach (var (prefix, projected) in new[] { ("p", true), ("s", false) }) + { + rig.Nm.UseProjectedEventDelivery = projected; + for (var i = 0; i < 5; i++) rig.Project(native: true, State($"{prefix}{i}")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onLine, $"{prefix}4") && events.Has(onE1, $"{prefix}4")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + rig.Nm.UseProjectedEventDelivery = true; + } + + var ids = events.Of(onLine) + .Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .Select(f => Convert.ToHexString(f[RecordedEvents.EventIdField].GetByteString().ToArray()!)) + .ToList(); + ids.Count.ShouldBe(10); + ids.Distinct().Count().ShouldBe(ids.Count, "each event once per item, though the walk reaches L1 twice"); + Normalised(events.Of(onLine), "p").ShouldBe(Normalised(events.Of(onLine), "s")); + } + + /// + /// Events still QUEUED when the item is modified (publishing off, then a new select list) are re-projected by the + /// SDK from each queued event's Handle; the projected path's per-item must take part in + /// that exactly like the SDK's own (each on a fresh server). + /// + [Fact] + public async Task Queued_events_re_projected_by_a_modify_are_identical_on_both_paths() + { + var projected = await ModifyScenarioAsync(projectedPath: true); + var sdk = await ModifyScenarioAsync(projectedPath: false); + projected.ShouldBe(sdk); + projected.Count.ShouldBe(Severities.Length); + } + + private static async Task> ModifyScenarioAsync(bool projectedPath) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync($"Projected855Modify{projectedPath}", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + rig.Nm.UseProjectedEventDelivery = projectedPath; + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var item = AddItem(subscription, ServerEventRig.UnsNodeId(session, EqA), OfTypeAlarm(alternative: false)); + await subscription.ApplyChangesAsync(ct); + item.Status.Error.ShouldBeNull(); + await subscription.SetPublishingModeAsync(false, ct); + + for (var i = 0; i < Severities.Length; i++) + { + rig.Project(native: true, new AlarmConditionSnapshot( + true, false, false, true, AlarmShelvingKind.Unshelved, Severities[i], $"m{i}")); + } + + item.Filter = OfTypeAlarm(alternative: true); // a new select list while every event is still queued + await subscription.ApplyChangesAsync(ct); + item.Status.Error.ShouldBeNull(); + await subscription.SetPublishingModeAsync(true, ct); + await RecordedEvents.WaitUntilAsync(() => events.Has(item, $"m{Severities.Length - 1}")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + return Normalised(events.Of(item), "m"); + } + + /// + /// The two keys the grouping rests on: the locale key is unambiguous (null, empty, one empty locale, and a locale + /// containing the separator are all distinct), and a where clause hashes by value (two value-equal filters land in + /// one bucket and share a projection). + /// + [Fact] + public void Locale_keys_are_unambiguous_and_where_clauses_hash_by_value() + { + string?[] keys = + [ + ProjectedEventDelivery.LocalesKey(null), ProjectedEventDelivery.LocalesKey([]), + ProjectedEventDelivery.LocalesKey([""]), ProjectedEventDelivery.LocalesKey(["a\u0001b"]), + ProjectedEventDelivery.LocalesKey(["a", "b"]), ProjectedEventDelivery.LocalesKey(["a;b"]), + ProjectedEventDelivery.LocalesKey(["a", ";b"]), + ]; + keys.Distinct().Count().ShouldBe(keys.Length); + Narrow(800).WhereClause.GetHashCode().ShouldBe(Narrow(800).WhereClause.GetHashCode()); + Narrow(800).WhereClause.GetHashCode().ShouldNotBe(Narrow(900).WhereClause.GetHashCode()); + } + + private static AlarmConditionSnapshot State(string message) => + new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message); + /// /// The projected path rests on PRIVATE members of the pinned SDK's MonitoredItem (its lock, its event queue /// handler, the filter in use). If an SDK upgrade drops one, delivery silently falls back to the SDK's per-item From 733dfe94dabd322efdd62ed1a4f2aa766de607c2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 14:20:19 -0500 Subject: [PATCH 4/5] =?UTF-8?q?fix(opcua):=20#855=20perf=20review=203=20?= =?UTF-8?q?=E2=80=94=20weak=20filter-group=20registry;=20where=20clause=20?= =?UTF-8?q?validated;=20sharing=20pinned?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The filter-group registry holds groups weakly (the filter->group table is the only strong reference), sweeps dead entries amortised and counts only live shapes against the 4096 cap; filters past the cap are counted on the new meter otopcua.opcua.condition_events.filter_groups_unshared. - A group validates its where clause too (restored, never-validated filters evaluate like validated ones). - Test: value-equal filters cost one projection per locale set per report. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Observability/OtOpcUaTelemetry.cs | 9 ++++ .../OtOpcUaNodeManager.cs | 9 +++- .../ProjectedEventDelivery.cs | 50 ++++++++++++++++--- .../ProjectedEventDeliveryWireTests.cs | 33 ++++++++++++ 4 files changed, 93 insertions(+), 8 deletions(-) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index b46012110..1bce3a838 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -177,6 +177,15 @@ public static class OtOpcUaTelemetry /// item had already replayed a newer state (Gitea #854) — one per item and event. Every other subscriber still /// receives them; this only stops a refresh being overtaken by older events. /// + /// + /// Event filters given a private projection group because the server-wide registry of distinct live filter shapes is + /// full (Gitea #855 perf). Delivery is unchanged; only the per-report sharing of where-clause / field projection + /// stops for those filters, so a steadily rising count means more per-item work under the node-manager lock. + /// + public static readonly Counter ConditionEventFilterGroupsUnshared = + Meter.CreateCounter("otopcua.opcua.condition_events.filter_groups_unshared", unit: "{filter}", + description: "Event filters not shared because the registry of live filter shapes is full."); + public static readonly Counter ConditionEventDeliveriesSuperseded = Meter.CreateCounter("otopcua.opcua.condition_events.superseded", unit: "{delivery}", description: "Deferred condition events withheld from a Server-object item whose ConditionRefresh already replayed a newer state."); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index a4284036a..5ae0c9c71 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1701,7 +1701,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } finally { - if (e is ConditionEventSnapshot snapshot) snapshot.ReleaseDeliveryCache(); + if (e is ConditionEventSnapshot snapshot && snapshot.ReleaseDeliveryCache() is { } cache) + Interlocked.Add(ref _projectionsCreated, cache.ProjectionsCreated); } } @@ -1722,6 +1723,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } private long _projectedDeliveries; + private long _projectionsCreated; + + /// Shared (filter group, locales) projections created across reports (test observation: sharing happens). + internal long ProjectionsCreated => Interlocked.Read(ref _projectionsCreated); /// Items this node manager served through the projected path (test observation: it is really taken). internal long ProjectedDeliveries => Interlocked.Read(ref _projectedDeliveries); @@ -1961,7 +1966,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } /// Drops the delivery cache once the walk is over (the snapshot stays referenced from the item queues). - public void ReleaseDeliveryCache() => Volatile.Write(ref _deliveryCache, null); + public EventDeliveryCache? ReleaseDeliveryCache() => Interlocked.Exchange(ref _deliveryCache, null); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs index 5ef544255..3f6733f74 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -2,6 +2,7 @@ using System.Linq.Expressions; using System.Reflection; using System.Runtime.CompilerServices; using Opc.Ua; +using ZB.MOM.WW.OtOpcUa.Commons.Observability; using Opc.Ua.Server; using SdkMonitoredItem = Opc.Ua.Server.MonitoredItem; @@ -106,12 +107,16 @@ internal sealed class EventDeliveryCache lock (_sync) return _queued.Add(item); } + /// Distinct (filter group, locales) projections this report created (test observation: sharing happens). + public int ProjectionsCreated { get; private set; } + /// The projection of the event for a filter group and locale list, created on first use. internal Projection ProjectionFor(FilterGroup group, string locales, Func create) { lock (_sync) { if (_projections.TryGetValue((group, locales), out var cached)) return cached; + ProjectionsCreated++; var projection = create(); _projections[(group, locales)] = projection; return projection; @@ -150,6 +155,9 @@ internal sealed class FilterGroup Representative = (EventFilter)CoreUtils.Clone(filter)!; // The SDK's GetRejectedSelectClauses: validation depends on the namespaces and type tree, never on the locales. var context = new FilterContext(server.NamespaceUris, server.TypeTree, server.Telemetry); + // The where clause's operands too (their parsed index ranges), so a group created from a never-validated filter + // (a restored item) evaluates exactly as one created from a validated filter would. + Representative.WhereClause?.Validate(context); var clauses = Representative.SelectClauses; Rejected = new bool[clauses.Count]; for (var i = 0; i < Rejected.Length; i++) @@ -184,14 +192,22 @@ internal static class ProjectedEventDelivery /// Whether every SDK member the fast path needs was found (pinned by a guard test). internal static bool Available => s_itemLock is not null && s_eventQueue is not null && s_filterToUse is not null; + // filter object -> group: the only STRONG reference to a group (a group lives exactly as long as some filter in use maps + // to it). The registry below holds groups weakly, so a shape no item uses any more drops out of it and of the count. private static readonly ConditionalWeakTable s_groups = new(); private static readonly object s_registryLock = new(); - private static readonly Dictionary> s_registry = new(StringComparer.Ordinal); + private static readonly Dictionary>> s_registry = new(StringComparer.Ordinal); private static int s_registered; + private static int s_sweepAt = 1024; - /// Distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing). + /// Live distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing). internal const int MaxSharedFilterGroups = 4096; + private static long s_projections; + + /// Where-clause evaluations + field projections performed (test observation: sharing really happens). + internal static long Projections => Interlocked.Read(ref s_projections); + /// /// The SDK's QueueEvent(instance, bypassFilter: false) for one condition snapshot, with the projection taken /// from . Returns false (nothing done) when the item is not in a shape this reproduces. @@ -231,6 +247,7 @@ internal static class ProjectedEventDelivery /// The SDK's CanSendFilteredAlarm without filtered retain: the where clause, for one group and locale list. private static bool Passes(IServerInternal server, FilterGroup group, string[]? locales, IFilterTarget e) { + Interlocked.Increment(ref s_projections); lock (group.Gate) { return group.Representative.WhereClause.Evaluate(ContextFor(server, locales), e); @@ -285,28 +302,49 @@ internal static class ProjectedEventDelivery { if (s_registry.TryGetValue(key, out var bucket)) { - foreach (var group in bucket) + foreach (var reference in bucket) { - if (Utils.IsEqual(group.Representative, f)) return group; + if (reference.TryGetTarget(out var group) && Utils.IsEqual(group.Representative, f)) return group; } } + if (s_registered >= s_sweepAt) SweepRegistry(); + var created = new FilterGroup(f, server); if (s_registered >= MaxSharedFilterGroups) { + OtOpcUaTelemetry.ConditionEventFilterGroupsUnshared.Add(1); Interlocked.Increment(ref s_unsharedGroups); return created; } s_registered++; - (bucket ??= s_registry[key] = []).Add(created); + (bucket ??= s_registry[key] = []).Add(new WeakReference(created)); return created; } }); + /// Drops registry entries whose group is gone (no filter in use maps to it), recounting the live ones. Under the registry lock. + private static void SweepRegistry() + { + var live = 0; + foreach (var key in s_registry.Keys.ToList()) + { + var bucket = s_registry[key]; + bucket.RemoveAll(r => !r.TryGetTarget(out _)); + if (bucket.Count == 0) s_registry.Remove(key); + else live += bucket.Count; + } + + s_registered = live; + // Amortised: the next sweep only once as many shapes again have been registered (never more often than every 1024). + s_sweepAt = Math.Max(1024, live * 2); + } + private static long s_unsharedGroups; - /// Filters given a private group because the shared registry is full (observability; sharing stops, delivery is unchanged). + /// Filters given a private group because the shared registry is full of LIVE shapes (also the meter + /// otopcua.opcua.condition_events.filter_groups_unshared; delivery is unchanged, only sharing stops). internal static long UnsharedFilterGroups => Interlocked.Read(ref s_unsharedGroups); /// A cheap bucket key over the select clauses (equality within a bucket is decided by Utils.IsEqual). diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs index 617539a6f..e244f0e23 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ProjectedEventDeliveryWireTests.cs @@ -295,6 +295,39 @@ public sealed class ProjectedEventDeliveryWireTests Narrow(800).WhereClause.GetHashCode().ShouldNotBe(Narrow(900).WhereClause.GetHashCode()); } + /// + /// Sharing really happens (not only equivalence): three items with value-equal filters on one session and one on a + /// session with other locales cost TWO projections per report (one per locale set), not four. + /// + [Fact] + public async Task Value_equal_filters_share_one_projection_per_locale_set() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Projected855Share", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var eqA = ServerEventRig.UnsNodeId(session, EqA); + var items = Enumerable.Range(0, 3).Select(_ => RecordedEvents.AddEventItem(subscription, eqA)).ToList(); + await subscription.ApplyChangesAsync(ct); + + using var german = await rig.ConnectAsync(anonymous: false, ct, preferredLocales: ["de-DE"]); + var germanEvents = new RecordedEvents(); + var germanSubscription = await RecordedEvents.SubscribeAsync(german, germanEvents, ct); + var g1 = RecordedEvents.AddEventItem(germanSubscription, ServerEventRig.UnsNodeId(german, EqA)); + await germanSubscription.ApplyChangesAsync(ct); + + var before = rig.Nm.ProjectionsCreated; + const int Reports = 5; + for (var i = 0; i < Reports; i++) rig.Project(native: true, State($"x{i}")); + (rig.Nm.ProjectionsCreated - before).ShouldBe(2 * Reports, "one projection per (filter group, locale set) per report"); + await RecordedEvents.WaitUntilAsync(() => items.All(m => events.Has(m, $"x{Reports - 1}")) && germanEvents.Has(g1, $"x{Reports - 1}")); + } + private static AlarmConditionSnapshot State(string message) => new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message); From 74b2ce170b5962ec68657cb7ee778066e7599b47 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 14:31:46 -0500 Subject: [PATCH 5/5] =?UTF-8?q?fix(opcua):=20#855=20perf=20review=204=20?= =?UTF-8?q?=E2=80=94=20registry=20sweep=20threshold=20capped;=20re-sweep?= =?UTF-8?q?=20at=20the=20cap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The sweep threshold never rises above the 4096 cap (a threshold the count could not reach stopped sweeping, and sharing, for good); at the cap a sweep for dead shapes runs after every 256 refusals. - The per-report "queued" mark is taken after the filter group resolves, so a failure there leaves the item unmarked. - The bucket key also carries TypeDefinitionId and IndexRange. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../ProjectedEventDelivery.cs | 32 ++++++++++++++----- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs index 3f6733f74..0f1aa859d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/ProjectedEventDelivery.cs @@ -199,6 +199,10 @@ internal static class ProjectedEventDelivery private static readonly Dictionary>> s_registry = new(StringComparer.Ordinal); private static int s_registered; private static int s_sweepAt = 1024; + private static int s_unsharedSinceSweep; + + /// At the cap, refusals between two sweeps for dead shapes. + private const int UnsharedResweepInterval = 256; /// Live distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing). internal const int MaxSharedFilterGroups = 4096; @@ -225,10 +229,9 @@ internal static class ProjectedEventDelivery var handler = s_eventQueue!(item); if (s_filterToUse!(item) is not EventFilter filter) return false; if (handler is null) return true; // Disabled: the SDK queues nothing (no handler) - if (!cache.TryMarkQueued(item)) return true; // the SDK's duplicate check (see EventDeliveryCache) - var locales = item.Session?.PreferredLocales; // string[]?, exactly what the SDK hands its FilterContext and TranslateValue var group = GroupOf(filter, server); + if (!cache.TryMarkQueued(item)) return true; // the SDK's duplicate check (see EventDeliveryCache) var projection = cache.ProjectionFor(group, LocalesKey(locales), () => new EventDeliveryCache.Projection( () => Passes(server, group, locales, e), () => Project(server, group, locales, e))); if (!projection.Passes || handler.SetQueueOverflowIfFull()) return true; @@ -313,9 +316,21 @@ internal static class ProjectedEventDelivery var created = new FilterGroup(f, server); if (s_registered >= MaxSharedFilterGroups) { - OtOpcUaTelemetry.ConditionEventFilterGroupsUnshared.Add(1); - Interlocked.Increment(ref s_unsharedGroups); - return created; + // Full: re-check for dead shapes after every UnsharedResweepInterval refusals, so a registry that filled + // once does not refuse sharing for good, and one that is genuinely full of live shapes is not swept on + // every new filter. + if (++s_unsharedSinceSweep >= UnsharedResweepInterval) + { + s_unsharedSinceSweep = 0; + SweepRegistry(); + } + + if (s_registered >= MaxSharedFilterGroups) + { + OtOpcUaTelemetry.ConditionEventFilterGroupsUnshared.Add(1); + Interlocked.Increment(ref s_unsharedGroups); + return created; + } } s_registered++; @@ -337,8 +352,9 @@ internal static class ProjectedEventDelivery } s_registered = live; - // Amortised: the next sweep only once as many shapes again have been registered (never more often than every 1024). - s_sweepAt = Math.Max(1024, live * 2); + // Amortised: the next sweep only once as many shapes again have been registered (never more often than every 1024), + // and never above the cap — a threshold the count can no longer reach would stop sweeping, and sharing, for good. + s_sweepAt = Math.Min(MaxSharedFilterGroups, Math.Max(1024, live * 2)); } private static long s_unsharedGroups; @@ -361,7 +377,7 @@ internal static class ProjectedEventDelivery var path = new List(); foreach (var name in clause.BrowsePath) path.Add(name.ToString()); - parts.Add(string.Join("/", path) + "@" + clause.AttributeId); + parts.Add(clause.TypeDefinitionId + ":" + string.Join("/", path) + "@" + clause.AttributeId + "[" + clause.IndexRange + "]"); } return string.Join("|", parts);