fix(opcua): #855 perf review 3 — weak filter-group registry; where clause validated; sharing pinned

- 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
This commit is contained in:
Joseph Doherty
2026-11-11 14:20:19 -05:00
parent b77bb2ded1
commit 733dfe94da
4 changed files with 93 additions and 8 deletions
@@ -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.
/// </summary>
/// <summary>
/// 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.
/// </summary>
public static readonly Counter<long> ConditionEventFilterGroupsUnshared =
Meter.CreateCounter<long>("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<long> ConditionEventDeliveriesSuperseded =
Meter.CreateCounter<long>("otopcua.opcua.condition_events.superseded", unit: "{delivery}",
description: "Deferred condition events withheld from a Server-object item whose ConditionRefresh already replayed a newer state.");
@@ -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;
/// <summary>Shared (filter group, locales) projections created across reports (test observation: sharing happens).</summary>
internal long ProjectionsCreated => Interlocked.Read(ref _projectionsCreated);
/// <summary>Items this node manager served through the projected path (test observation: it is really taken).</summary>
internal long ProjectedDeliveries => Interlocked.Read(ref _projectedDeliveries);
@@ -1961,7 +1966,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
}
/// <summary>Drops the delivery cache once the walk is over (the snapshot stays referenced from the item queues).</summary>
public void ReleaseDeliveryCache() => Volatile.Write(ref _deliveryCache, null);
public EventDeliveryCache? ReleaseDeliveryCache() => Interlocked.Exchange(ref _deliveryCache, null);
}
/// <summary>
@@ -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);
}
/// <summary>Distinct (filter group, locales) projections this report created (test observation: sharing happens).</summary>
public int ProjectionsCreated { get; private set; }
/// <summary>The projection of the event for a filter group and locale list, created on first use.</summary>
internal Projection ProjectionFor(FilterGroup group, string locales, Func<Projection> 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
/// <summary>Whether every SDK member the fast path needs was found (pinned by a guard test).</summary>
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<EventFilter, FilterGroup> s_groups = new();
private static readonly object s_registryLock = new();
private static readonly Dictionary<string, List<FilterGroup>> s_registry = new(StringComparer.Ordinal);
private static readonly Dictionary<string, List<WeakReference<FilterGroup>>> s_registry = new(StringComparer.Ordinal);
private static int s_registered;
private static int s_sweepAt = 1024;
/// <summary>Distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing).</summary>
/// <summary>Live distinct filter shapes kept for sharing; past this a new filter gets a group of its own (no sharing).</summary>
internal const int MaxSharedFilterGroups = 4096;
private static long s_projections;
/// <summary>Where-clause evaluations + field projections performed (test observation: sharing really happens).</summary>
internal static long Projections => Interlocked.Read(ref s_projections);
/// <summary>
/// The SDK's <c>QueueEvent(instance, bypassFilter: false)</c> for one condition snapshot, with the projection taken
/// from <paramref name="cache"/>. Returns false (nothing done) when the item is not in a shape this reproduces.
@@ -231,6 +247,7 @@ internal static class ProjectedEventDelivery
/// <summary>The SDK's <c>CanSendFilteredAlarm</c> without filtered retain: the where clause, for one group and locale list.</summary>
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<FilterGroup>(created));
return created;
}
});
/// <summary>Drops registry entries whose group is gone (no filter in use maps to it), recounting the live ones. Under the registry lock.</summary>
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;
/// <summary>Filters given a private group because the shared registry is full (observability; sharing stops, delivery is unchanged).</summary>
/// <summary>Filters given a private group because the shared registry is full of LIVE shapes (also the meter
/// <c>otopcua.opcua.condition_events.filter_groups_unshared</c>; delivery is unchanged, only sharing stops).</summary>
internal static long UnsharedFilterGroups => Interlocked.Read(ref s_unsharedGroups);
/// <summary>A cheap bucket key over the select clauses (equality within a bucket is decided by <c>Utils.IsEqual</c>).</summary>