dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge #855 folder event delivery perf harness (Category=Perf, opt-in)

Browse Source 
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 13:21:27 -05:00
parent ca6d131078 a8f4be616b
commit 50b7b030a9
1 changed files with 350 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventDeliveryPerfTests.cs
+350
View File
@@ -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;
/// <summary>
/// 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 <c>Lock</c> (per-item verdict, <c>QueueEvent</c> with the SDK's
/// filter evaluation, duplicate scan over the last 1000 queued events and a <c>List.RemoveAt(0)</c> at a full queue).
/// Before #855 that work ran on the SDK's per-node channel consumer, off <c>Lock</c>. This harness reports, per alarm
/// report, the <c>Lock</c> 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 (<c>i=2253</c>) subscriber while the folder's subscribers
/// are stalled (the ScadaBridge shape).
/// </summary>
/// <remarks>
/// <para>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 <c>OTOPCUA_EVENT_PERF=1</c>.</para>
/// <para>Run: <c>OTOPCUA_EVENT_PERF=1 dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests -c Release
/// --filter "FullyQualifiedName~FolderEventDeliveryPerfTests" --logger "console;verbosity=detailed"</c>.
/// <c>OTOPCUA_EVENT_PERF_REPORTS</c> sets the measured reports per case (default 1000) and
/// <c>OTOPCUA_EVENT_PERF_OUT</c> appends each result row to a file.</para>
/// <para>The rig is the same one <see cref="FolderEventFloodWireTests"/> uses and existed before #855, so this file
/// compiles against the pre-#855 tree for a before/after comparison.</para>
/// </remarks>
[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)");
}
/// <summary>
/// Lock hold per report and (publishing only) the time until the LAST of the N folder items received the event.
/// </summary>
[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);
/// <summary>
/// A publishing Server-object (<c>i=2253</c>) subscriber on its own session, while N folder items on another
/// session are stalled with full queues. N = 0 is the baseline.
/// </summary>
[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);
/// <summary>
/// Full-queue sensitivity: N = 50 stalled items at different <c>QueueSize</c>s. At a full queue every
/// <c>QueueEvent</c> scans the last min(QueueSize, 1000) queued events for a duplicate and shifts the whole queue
/// (<c>List.RemoveAt(0)</c>), so the per-item cost grows with the queue, not with N.
/// </summary>
[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<uint>();
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();
}
}
/// <summary>
/// Gives an off-<c>Lock</c> consumer (pre-#855) time to drain what was just reported; post-#855 nothing is pending.
/// </summary>
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<Subscription> 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,
};
/// <summary>Client-side receipt times of the measured (<c>p{i}</c>) events, per probe.</summary>
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<uint> FolderHandles { get; set; } = new HashSet<uint>();
public uint? ServerHandle { get; set; }
public void OnEvents(Subscription subscription, EventNotificationList notification, ArrayOf<string> 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();
}
}
}
}