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);