Joseph Doherty
74b2ce170b
- 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
415 lines
22 KiB
C#
415 lines
22 KiB
C#
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;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>Why.</b> Since #855 condition events are queued to folder items by the reporter under the node manager's
|
||
/// <c>Lock</c>. The SDK's <c>MonitoredItem.QueueEvent(IFilterTarget)</c> 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.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>What is reproduced, exactly.</b> <see cref="TryQueue"/> performs the SDK's <c>QueueEvent(instance, false)</c>
|
||
/// 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 (<c>FilterToUse</c>, the revised one); <c>CanSendFilteredAlarm</c>, which
|
||
/// without <c>SupportsFilteredRetain</c> on the condition is the where clause alone; <c>SetQueueOverflowIfFull</c>;
|
||
/// then <c>GetEventFields</c> (rejected select clauses ⇒ null, <c>LocalizedText</c> translated to the session's
|
||
/// preferred locales, <c>Handle</c> = the event, the item's <c>ClientHandle</c>) and the public
|
||
/// <c>QueueEvent(EventFieldList)</c>. Each item gets its OWN <see cref="EventFieldList"/> (the SDK mutates a queued
|
||
/// list's <c>ClientHandle</c> on a modify); the field values are shared (immutable).
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Duplicate check.</b> 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 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.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Fallback.</b> 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 <c>MonitoredItem</c>, a filter in use that is not an
|
||
/// <c>EventFilter</c>, a condition with <c>SupportsFilteredRetain</c>, or an event that is not a condition snapshot —
|
||
/// goes through the SDK's own <c>QueueEvent(IFilterTarget)</c>, unchanged.
|
||
/// </para>
|
||
/// </remarks>
|
||
internal sealed class EventDeliveryCache
|
||
{
|
||
private readonly object _sync = new();
|
||
private readonly HashSet<IEventMonitoredItem> _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();
|
||
|
||
/// <summary>The SDK role-permission verdict for an item, computed once per (session, effective identity).</summary>
|
||
/// <param name="item">The event item.</param>
|
||
/// <param name="compute">The SDK check (<c>ValidateEventRolePermissions</c>) for that item.</param>
|
||
/// <returns><see langword="true"/> when the item's identity may receive the event.</returns>
|
||
/// <remarks>
|
||
/// The SDK check builds an <c>OperationContext</c> from the item and asks <c>ReceiveEvents</c> 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. Held for the report's walk: on a walk that
|
||
/// suspends (none of ours does) a <c>RolePermissions</c> change between two hops lands one event later.
|
||
/// </remarks>
|
||
public bool RoleAllows(IEventMonitoredItem item, Func<IEventMonitoredItem, bool> 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;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Queues the event to the item through the projected fast path when it has exactly the shape the SDK's
|
||
/// <c>QueueEvent(IFilterTarget)</c> would handle, else through that method itself.
|
||
/// </summary>
|
||
/// <param name="server">The server (namespaces, type tree, resource manager).</param>
|
||
/// <param name="item">The event item.</param>
|
||
/// <param name="e">The condition event.</param>
|
||
/// <param name="projected">False routes it through the SDK's own <c>QueueEvent</c> (the node manager's test switch).</param>
|
||
/// <returns>True when the projected path handled the item (queued, or decided not to queue exactly as the SDK would).</returns>
|
||
public bool Queue(IServerInternal server, IEventMonitoredItem item, IFilterTarget e, bool projected)
|
||
{
|
||
if (projected
|
||
&& e is InstanceStateSnapshot { Handle: ConditionState condition }
|
||
&& condition.SupportsFilteredRetain?.Value != true
|
||
&& item.GetType() == typeof(SdkMonitoredItem)
|
||
&& ProjectedEventDelivery.TryQueue(server, (SdkMonitoredItem)item, e, this))
|
||
{
|
||
return true;
|
||
}
|
||
|
||
item.QueueEvent(e);
|
||
return false;
|
||
}
|
||
|
||
/// <summary>Marks the item as queued for this report; false when it already was (the SDK's duplicate check).</summary>
|
||
internal bool TryMarkQueued(IEventMonitoredItem item)
|
||
{
|
||
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;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// The fields are shared by every item queued from this projection: <see cref="ArrayOf{T}"/> 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.
|
||
/// </remarks>
|
||
internal sealed class Projection(Func<bool> passes, Func<ArrayOf<Variant>> fields)
|
||
{
|
||
private readonly Lazy<bool> _passes = new(passes, LazyThreadSafetyMode.ExecutionAndPublication);
|
||
private readonly Lazy<ArrayOf<Variant>> _fields = new(fields, LazyThreadSafetyMode.ExecutionAndPublication);
|
||
|
||
public bool Passes => _passes.Value;
|
||
|
||
public ArrayOf<Variant> Fields => _fields.Value;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// A set of value-equal <see cref="EventFilter"/>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: <c>Validate</c> mutates a select clause).
|
||
/// </summary>
|
||
internal sealed class FilterGroup
|
||
{
|
||
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);
|
||
// 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++)
|
||
{
|
||
var clause = clauses[i];
|
||
Rejected[i] = clause is null || (!clause.Validated && ServiceResult.IsBad(clause.Validate(context, 0)));
|
||
}
|
||
}
|
||
|
||
/// <summary>The group's own copy of the filter; evaluated only under <see cref="Gate"/>.</summary>
|
||
public EventFilter Representative { get; }
|
||
|
||
/// <summary>Per select clause: rejected (always projected as null), as the SDK decides once per filter.</summary>
|
||
public bool[] Rejected { get; }
|
||
|
||
/// <summary>Serialises evaluation of <see cref="Representative"/> (two reports may project it at once).</summary>
|
||
public object Gate { get; } = new();
|
||
}
|
||
|
||
/// <summary>The SDK internals the projected fast path reads, resolved once by reflection — see <see cref="EventDeliveryCache"/>.</summary>
|
||
internal static class ProjectedEventDelivery
|
||
{
|
||
private static readonly Func<SdkMonitoredItem, Lock?>? s_itemLock = FieldGetter<Lock>("m_lock");
|
||
private static readonly Func<SdkMonitoredItem, EventQueueHandler?>? s_eventQueue = FieldGetter<EventQueueHandler>("m_eventQueueHandler");
|
||
private static readonly Func<SdkMonitoredItem, MonitoringFilter?>? s_filterToUse = PropertyGetter<MonitoringFilter>("FilterToUse");
|
||
|
||
private static long s_projectedQueued;
|
||
|
||
/// <summary>Events queued through the projected path since start (test observation: the fast path is really taken).</summary>
|
||
internal static long ProjectedQueued => Interlocked.Read(ref s_projectedQueued);
|
||
|
||
/// <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<WeakReference<FilterGroup>>> s_registry = new(StringComparer.Ordinal);
|
||
private static int s_registered;
|
||
private static int s_sweepAt = 1024;
|
||
private static int s_unsharedSinceSweep;
|
||
|
||
/// <summary>At the cap, refusals between two sweeps for dead shapes.</summary>
|
||
private const int UnsharedResweepInterval = 256;
|
||
|
||
/// <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.
|
||
/// </summary>
|
||
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)
|
||
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;
|
||
|
||
Interlocked.Increment(ref s_projectedQueued);
|
||
item.QueueEvent(new EventFieldList
|
||
{
|
||
ClientHandle = item.ClientHandle,
|
||
Handle = e,
|
||
EventFields = projection.Fields, // shared, read-only (see Projection)
|
||
});
|
||
return true;
|
||
}
|
||
}
|
||
|
||
/// <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);
|
||
}
|
||
}
|
||
|
||
/// <summary>The SDK's <c>GetEventFields</c> values, for one group and locale list.</summary>
|
||
private static ArrayOf<Variant> 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;
|
||
}
|
||
}
|
||
|
||
/// <summary>The SDK's per-item <c>FilterContext</c>: the server's namespaces and type tree, the session's locales (null included).</summary>
|
||
private static FilterContext ContextFor(IServerInternal server, string[]? locales) =>
|
||
new(server.NamespaceUris, server.TypeTree, locales!, server.Telemetry);
|
||
|
||
/// <summary>The canonical group of a filter object: value-equal filters share one (decided once per filter object).</summary>
|
||
private static FilterGroup GroupOf(EventFilter filter, IServerInternal server) =>
|
||
s_groups.GetValue(filter, 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))
|
||
{
|
||
foreach (var reference in bucket)
|
||
{
|
||
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)
|
||
{
|
||
// 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++;
|
||
(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),
|
||
// 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;
|
||
|
||
/// <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>
|
||
private static string SelectKey(EventFilter filter)
|
||
{
|
||
var parts = new List<string>();
|
||
foreach (var clause in filter.SelectClauses)
|
||
{
|
||
if (clause is null)
|
||
{
|
||
parts.Add("-");
|
||
continue;
|
||
}
|
||
|
||
var path = new List<string>();
|
||
foreach (var name in clause.BrowsePath) path.Add(name.ToString());
|
||
parts.Add(clause.TypeDefinitionId + ":" + string.Join("/", path) + "@" + clause.AttributeId + "[" + clause.IndexRange + "]");
|
||
}
|
||
|
||
return string.Join("|", parts);
|
||
}
|
||
|
||
/// <summary>An unambiguous key for a locale list (length-prefixed entries; null distinct from empty).</summary>
|
||
private static readonly ConditionalWeakTable<string[], string> s_localeKeys = new();
|
||
|
||
/// <summary>An unambiguous key for a locale list, cached per array (a session's array is fixed for its lifetime).</summary>
|
||
internal static string LocalesKey(string[]? locales) =>
|
||
locales is null ? "null" : s_localeKeys.GetValue(locales, static l => BuildLocalesKey(l));
|
||
|
||
private static string BuildLocalesKey(string[] locales)
|
||
{
|
||
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<SdkMonitoredItem, T?>? FieldGetter<T>(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<Func<SdkMonitoredItem, T?>>(Expression.Convert(Expression.Field(item, field), typeof(T)), item).Compile();
|
||
}
|
||
|
||
private static Func<SdkMonitoredItem, T?>? PropertyGetter<T>(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<Func<SdkMonitoredItem, T?>>();
|
||
}
|
||
}
|