Merge #855 shared per-report event projection: Lock hold back near pre-#855 at N=50
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
@@ -100,3 +100,37 @@ 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), 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;
|
||||
`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`, 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, 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.
|
||||
|
||||
@@ -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.");
|
||||
|
||||
@@ -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,75 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
await DeliverConditionEventAsync(condition, parent, e).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private async ValueTask DeliverReportAsync(AlarmConditionState condition, IFilterTarget e)
|
||||
{
|
||||
try
|
||||
{
|
||||
await DeliverConditionEventAsync(condition, condition, e).ConfigureAwait(false);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (e is ConditionEventSnapshot snapshot && snapshot.ReleaseDeliveryCache() is { } cache)
|
||||
Interlocked.Add(ref _projectionsCreated, cache.ProjectionsCreated);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>QueueEvent</c>.
|
||||
/// </summary>
|
||||
private void QueueToItem(IEventMonitoredItem item, IFilterTarget 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;
|
||||
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);
|
||||
|
||||
/// <summary>
|
||||
/// Whether condition events are queued through the report's shared projection (#855 perf, default) or through the
|
||||
/// SDK's own <c>QueueEvent(IFilterTarget)</c> per item, with the role and grant checks asked per item (the
|
||||
/// pre-optimisation path). A test switch: the two must deliver identical events.
|
||||
/// </summary>
|
||||
internal bool UseProjectedEventDelivery { get; set; } = true;
|
||||
|
||||
/// <summary>The SDK's role-permission verdict for one item, shared per (session, identity) within a report (#855 perf).</summary>
|
||||
private bool RoleAllows(IEventMonitoredItem item, IFilterTarget e) =>
|
||||
UseProjectedEventDelivery && e is ConditionEventSnapshot snapshot
|
||||
? snapshot.DeliveryCache.RoleAllows(item, i => !ServiceResult.IsBad(ValidateEventRolePermissions(i, e)))
|
||||
: !ServiceResult.IsBad(ValidateEventRolePermissions(item, e));
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IsConditionVisibleTo"/> per effective identity within one delivery (#855 perf): the decision reads the
|
||||
/// item's identity only, and a delivery's scopes are fixed.
|
||||
/// </summary>
|
||||
private bool IsConditionVisibleTo(
|
||||
IEventMonitoredItem item, List<NodeId> scopes, Dictionary<object, bool> perIdentity)
|
||||
{
|
||||
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;
|
||||
return visible;
|
||||
}
|
||||
|
||||
/// <summary>Marks an event whose remaining delivery will run off <c>Lock</c> (#855) — before the walk suspends.</summary>
|
||||
private static void MarkDeferred(IFilterTarget e)
|
||||
{
|
||||
@@ -1756,14 +1825,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
scopes = ConditionVisibilityScopes(live);
|
||||
}
|
||||
|
||||
var visibleByIdentity = scopes is null ? null : new Dictionary<object, bool>(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 +1947,26 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3
|
||||
get => _deferred;
|
||||
set => _deferred = value;
|
||||
}
|
||||
|
||||
private EventDeliveryCache? _deliveryCache;
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>Lock</c>); dropped when the
|
||||
/// walk ends.
|
||||
/// </summary>
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Drops the delivery cache once the walk is over (the snapshot stays referenced from the item queues).</summary>
|
||||
public EventDeliveryCache? ReleaseDeliveryCache() => Interlocked.Exchange(ref _deliveryCache, null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -5841,14 +5931,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<object, bool>(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 +5975,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 +5998,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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,414 @@
|
||||
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?>>();
|
||||
}
|
||||
}
|
||||
+413
@@ -0,0 +1,413 @@
|
||||
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.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using MonitoredItem = Opc.Ua.Client.MonitoredItem;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// #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 <c>QueueEvent(IFilterTarget)</c> 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 <c>UseProjectedEventDelivery</c> switch, which also turns the verdict sharing off) and
|
||||
/// compare every item's stream.
|
||||
/// </summary>
|
||||
public sealed class ProjectedEventDeliveryWireTests
|
||||
{
|
||||
private const string EqA = ServerEventRig.EqA;
|
||||
|
||||
private static readonly ushort[] Severities = [700, 900, 300, 850, 801, 800, 999, 950];
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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, 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);
|
||||
|
||||
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();
|
||||
|
||||
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)
|
||||
{
|
||||
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));
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
[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<List<string>> 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();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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"));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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 <see cref="EventFieldList"/> must take part in
|
||||
/// that exactly like the SDK's own (each on a fresh server).
|
||||
/// </summary>
|
||||
[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<List<string>> 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");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
[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());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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);
|
||||
|
||||
/// <summary>
|
||||
/// The projected path rests on PRIVATE members of the pinned SDK's <c>MonitoredItem</c> (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
|
||||
/// <c>QueueEvent</c> (correct, slower) — this fails first, so the upgrade is looked at.
|
||||
/// </summary>
|
||||
[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)");
|
||||
|
||||
/// <summary>One phase's events of an item, with the per-event identity (EventId, the phase prefix) removed.</summary>
|
||||
private static List<string> Normalised(IReadOnlyList<ArrayOf<Variant>> 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;
|
||||
}
|
||||
|
||||
/// <summary>EventId, Message, Severity — with the optional where clause Severity > <paramref name="severityAbove"/>.</summary>
|
||||
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();
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>OfType AlarmConditionType: EventId, Message, Retain, a field the event lacks (or, alternatively, Severity).</summary>
|
||||
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));
|
||||
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;
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -298,7 +298,7 @@ internal sealed class ServerEventRig : IAsyncDisposable
|
||||
.Invoke(server, [on]);
|
||||
}
|
||||
|
||||
public async Task<ISession> ConnectAsync(bool anonymous, CancellationToken ct, bool encrypted = false)
|
||||
public async Task<ISession> 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<string>)locales : default,
|
||||
ct: ct);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user