diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs new file mode 100644 index 000000000..25b14788e --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs @@ -0,0 +1,350 @@ +using System.Diagnostics; +using System.Globalization; +using System.Text; +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; + +/// +/// Gitea #855 cost measurement: since #855 a condition event is queued to every event item on an equipment folder by +/// the reporter itself, under the node manager's Lock (per-item verdict, QueueEvent with the SDK's +/// filter evaluation, duplicate scan over the last 1000 queued events and a List.RemoveAt(0) at a full queue). +/// Before #855 that work ran on the SDK's per-node channel consumer, off Lock. This harness reports, per alarm +/// report, the Lock hold time and the end-to-end delivery latency, for one equipment folder carrying N event +/// items, ACL on / off, with the folder's subscribers publishing or stalled (publishing disabled, item queues full and +/// discarding oldest) — and the latency of a Server-object (i=2253) subscriber while the folder's subscribers +/// are stalled (the ScadaBridge shape). +/// +/// +/// Measurement only: nothing is asserted beyond the harness's own sanity checks. Opt-in and never part of a +/// default run: every case skips unless OTOPCUA_EVENT_PERF=1. +/// Run: OTOPCUA_EVENT_PERF=1 dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests -c Release +/// --filter "FullyQualifiedName~FolderEventDeliveryPerfTests" --logger "console;verbosity=detailed". +/// OTOPCUA_EVENT_PERF_REPORTS sets the measured reports per case (default 1000) and +/// OTOPCUA_EVENT_PERF_OUT appends each result row to a file. +/// The rig is the same one uses and existed before #855, so this file +/// compiles against the pre-#855 tree for a before/after comparison. +/// +[Trait("Category", "Perf")] +public sealed class FolderEventDeliveryPerfTests +{ + private const string OptInVariable = "OTOPCUA_EVENT_PERF"; + private const string EqA = ServerEventRig.EqA; + private const uint DefaultQueueSize = 1000; + private const double PublishingIntervalMs = 50; + + private static int Reports => + int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_EVENT_PERF_REPORTS"), out var r) && r > 0 ? r : 1000; + + private static void RequireOptIn() + { + if (Environment.GetEnvironmentVariable(OptInVariable) != "1") + Assert.Skip($"perf harness: set {OptInVariable}=1 to run (not part of a default run)"); + } + + /// + /// Lock hold per report and (publishing only) the time until the LAST of the N folder items received the event. + /// + [Theory] + [InlineData(1, false, false)] + [InlineData(10, false, false)] + [InlineData(50, false, false)] + [InlineData(200, false, false)] + [InlineData(1, true, false)] + [InlineData(10, true, false)] + [InlineData(50, true, false)] + [InlineData(200, true, false)] + [InlineData(1, false, true)] + [InlineData(10, false, true)] + [InlineData(50, false, true)] + [InlineData(200, false, true)] + [InlineData(1, true, true)] + [InlineData(10, true, true)] + [InlineData(50, true, true)] + [InlineData(200, true, true)] + public Task Folder_items(int n, bool acl, bool stalled) => RunAsync(n, acl, stalled, serverObjectProbe: false); + + /// + /// A publishing Server-object (i=2253) subscriber on its own session, while N folder items on another + /// session are stalled with full queues. N = 0 is the baseline. + /// + [Theory] + [InlineData(0, false)] + [InlineData(50, false)] + [InlineData(200, false)] + [InlineData(0, true)] + [InlineData(50, true)] + [InlineData(200, true)] + public Task Server_object_subscriber_while_folder_items_stall(int n, bool acl) => + RunAsync(n, acl, stalled: true, serverObjectProbe: true); + + /// + /// Full-queue sensitivity: N = 50 stalled items at different QueueSizes. At a full queue every + /// QueueEvent scans the last min(QueueSize, 1000) queued events for a duplicate and shifts the whole queue + /// (List.RemoveAt(0)), so the per-item cost grows with the queue, not with N. + /// + [Theory] + [InlineData(10u)] + [InlineData(100u)] + [InlineData(1000u)] + [InlineData(10000u)] + public Task Stalled_items_queue_size_sensitivity(uint queueSize) => + RunAsync(50, acl: false, stalled: true, serverObjectProbe: false, queueSize); + + private static async Task RunAsync( + int n, bool acl, bool stalled, bool serverObjectProbe, uint queueSize = DefaultQueueSize) + { + RequireOptIn(); + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync($"Perf855-{n}", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(acl, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + var reports = Reports; + var reportedAt = new long[reports]; + var receipts = new Receipts(reports); + + ISession? folderSession = null; + ISession? serverSession = null; + try + { + Subscription? folderSubscription = null; + var folderHandles = new HashSet(); + if (n > 0) + { + folderSession = await rig.ConnectAsync(anonymous: false, ct); + folderSubscription = await SubscribeAsync(folderSession, receipts, isServer: false, ct); + for (var i = 0; i < n; i++) + { + var item = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + item.QueueSize = queueSize; + folderHandles.Add(item.ClientHandle); + } + + await folderSubscription.ApplyChangesAsync(ct); + folderSubscription.MonitoredItems.ShouldAllBe(m => m.Status.Error == null); + // The server's revision (MaxEventQueueSize) — what the measured queues actually hold. + queueSize = folderSubscription.MonitoredItems.First().Status.QueueSize; + } + + receipts.FolderHandles = folderHandles; + + MonitoredItem? onServer = null; + if (serverObjectProbe) + { + serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverSubscription = await SubscribeAsync(serverSession, receipts, isServer: true, ct); + onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + onServer.Status.Error.ShouldBeNull(); + receipts.ServerHandle = onServer.ClientHandle; + } + + if (stalled && folderSubscription is not null) + { + await folderSubscription.SetPublishingModeAsync(false, ct); + + // Fill every folder item's queue to QueueSize (and past it, so the measured reports run at a full queue, + // discarding oldest), then let any off-Lock consumer (pre-#855: the SDK's channel task) catch up. + for (var i = 0; i < queueSize + 50; i++) rig.Project(native: true, State($"w{i}")); + await SettleAsync(rig, n, ct); + } + + // Warm-up (JIT, first-verdict caches) — not measured. + for (var i = 0; i < 50; i++) rig.Project(native: true, State($"x{i}")); + await SettleAsync(rig, n, ct); + + var holds = new double[reports]; + for (var i = 0; i < reports; i++) + { + var state = State($"p{i}"); + long t0, t1; + lock (rig.Nm.Lock) + { + t0 = Stopwatch.GetTimestamp(); + reportedAt[i] = t0; + rig.Project(native: true, state); + t1 = Stopwatch.GetTimestamp(); + } + + holds[i] = Stopwatch.GetElapsedTime(t0, t1).TotalMicroseconds; + // ~1 kHz report rate: an alarm storm, not a single burst, so a consumer that keeps up is not penalised. + Thread.Sleep(1); + } + + // Wait for the tail of every probe that is expected to receive. + var expectFolder = n > 0 && !stalled; + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); + while (DateTime.UtcNow < deadline + && ((expectFolder && !receipts.FolderComplete(reports - 1, n)) + || (serverObjectProbe && !receipts.ServerHas(reports - 1)))) + { + await Task.Delay(50, ct); + } + + var deferredPeak = rig.Nm.DeferredEventReports; + var row = new StringBuilder(); + row.Append(CultureInfo.InvariantCulture, + $"{(serverObjectProbe ? "i=2253" : "folder"),-6} N={n,-3} acl={(acl ? "on " : "off")} {(stalled ? "stalled " : "publishing")} q={queueSize,-5} | "); + row.Append(CultureInfo.InvariantCulture, $"lock hold us p50 {Pct(holds, 50),8:F1} p90 {Pct(holds, 90),8:F1} p99 {Pct(holds, 99),8:F1} max {holds.Max(),9:F1} mean {holds.Average(),8:F1}"); + row.Append(CultureInfo.InvariantCulture, $" | per-item mean {(n > 0 ? holds.Average() / n : 0),6:F2} us"); + if (expectFolder) + { + var lat = receipts.FolderLatenciesMs(reportedAt, n); + row.Append(CultureInfo.InvariantCulture, + $" | folder all-N latency ms p50 {Pct(lat, 50),6:F1} p99 {Pct(lat, 99),6:F1} max {Max(lat),7:F1} (got {lat.Length}/{reports})"); + } + + if (serverObjectProbe) + { + var lat = receipts.ServerLatenciesMs(reportedAt); + row.Append(CultureInfo.InvariantCulture, + $" | i=2253 latency ms p50 {Pct(lat, 50),6:F1} p99 {Pct(lat, 99),6:F1} max {Max(lat),7:F1} (got {lat.Length}/{reports})"); + } + + row.Append(CultureInfo.InvariantCulture, $" | deferred-now {deferredPeak}"); + Report(row.ToString()); + } + finally + { + folderSession?.Dispose(); + serverSession?.Dispose(); + } + } + + /// + /// Gives an off-Lock consumer (pre-#855) time to drain what was just reported; post-#855 nothing is pending. + /// + private static async Task SettleAsync(ServerEventRig rig, int n, CancellationToken ct) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(60); + while (rig.Nm.DeferredEventReports > 0 && DateTime.UtcNow < deadline) await Task.Delay(50, ct); + await Task.Delay(TimeSpan.FromMilliseconds(500 + (10 * n)), ct); + } + + private static async Task SubscribeAsync(ISession session, Receipts receipts, bool isServer, CancellationToken ct) + { + var subscription = new Subscription(session.DefaultSubscription) + { + PublishingInterval = (int)PublishingIntervalMs, + MaxNotificationsPerPublish = 0, + }; + subscription.FastEventCallback = receipts.OnEvents; + session.AddSubscription(subscription); + await subscription.CreateAsync(ct); + return subscription; + } + + private static double Pct(double[] values, double p) + { + if (values.Length == 0) return double.NaN; + var sorted = values.Order().ToArray(); + var idx = (int)Math.Ceiling(p / 100.0 * sorted.Length) - 1; + return sorted[Math.Clamp(idx, 0, sorted.Length - 1)]; + } + + private static double Max(double[] values) => values.Length == 0 ? double.NaN : values.Max(); + + private static void Report(string text) + { + TestContext.Current.TestOutputHelper?.WriteLine(text); + if (Environment.GetEnvironmentVariable("OTOPCUA_EVENT_PERF_OUT") is { Length: > 0 } path) + File.AppendAllText(path, text + Environment.NewLine); + } + + private static AlarmConditionSnapshot State(string message) => + new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message); + + 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, + }; + + /// Client-side receipt times of the measured (p{i}) events, per probe. + private sealed class Receipts(int reports) + { + private readonly object _gate = new(); + private readonly int[] _folderCount = new int[reports]; + private readonly long[] _folderLast = new long[reports]; + private readonly long[] _serverFirst = new long[reports]; + + public IReadOnlySet FolderHandles { get; set; } = new HashSet(); + + public uint? ServerHandle { get; set; } + + public void OnEvents(Subscription subscription, EventNotificationList notification, ArrayOf stringTable) + { + var now = Stopwatch.GetTimestamp(); + lock (_gate) + { + foreach (var e in notification.Events) + { + var message = RecordedEvents.Message(e.EventFields); + if (message is not { Length: > 1 } || message[0] != 'p') continue; + if (!int.TryParse(message.AsSpan(1), NumberStyles.Integer, CultureInfo.InvariantCulture, out var i) + || i < 0 || i >= _folderCount.Length) + { + continue; + } + + if (ServerHandle == e.ClientHandle) + { + if (_serverFirst[i] == 0) _serverFirst[i] = now; + } + else if (FolderHandles.Contains(e.ClientHandle)) + { + _folderCount[i]++; + _folderLast[i] = now; + } + } + } + } + + public bool FolderComplete(int i, int n) + { + lock (_gate) return _folderCount[i] >= n; + } + + public bool ServerHas(int i) + { + lock (_gate) return _serverFirst[i] != 0; + } + + public double[] FolderLatenciesMs(long[] reportedAt, int n) + { + lock (_gate) + { + return Enumerable.Range(0, reportedAt.Length) + .Where(i => _folderCount[i] >= n) + .Select(i => Stopwatch.GetElapsedTime(reportedAt[i], _folderLast[i]).TotalMilliseconds) + .ToArray(); + } + } + + public double[] ServerLatenciesMs(long[] reportedAt) + { + lock (_gate) + { + return Enumerable.Range(0, reportedAt.Length) + .Where(i => _serverFirst[i] != 0) + .Select(i => Stopwatch.GetElapsedTime(reportedAt[i], _serverFirst[i]).TotalMilliseconds) + .ToArray(); + } + } + } +}