From 2dc55ecb8f58d6b47912eec709ba7accdb3d6274 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 7 Oct 2026 19:26:23 -0400 Subject: [PATCH] test(abcip): read-only poll-latency comparison harness for the T21 press gate (proven on ab_server) --- ...-10-07-native-ab-transport-gate-results.md | 18 ++ .../Differential/DifferentialRun.cs | 92 +++++-- .../Differential/DifferentialSafetyTests.cs | 24 +- .../Differential/DifferentialTarget.cs | 25 +- .../PollLatencyComparisonTests.cs | 117 +++++++++ .../Differential/PollLatencyRun.cs | 240 ++++++++++++++++++ .../TransportDifferentialTests.cs | 6 + 7 files changed, 495 insertions(+), 27 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs diff --git a/docs/plans/2026-10-07-native-ab-transport-gate-results.md b/docs/plans/2026-10-07-native-ab-transport-gate-results.md index 3d2570e2..c6bf5d33 100644 --- a/docs/plans/2026-10-07-native-ab-transport-gate-results.md +++ b/docs/plans/2026-10-07-native-ab-transport-gate-results.md @@ -185,6 +185,24 @@ see the same split on the press. The ab_server controllogix matrix is read-only and unaffected: 15 passed and 11 skipped with `AB_SERVER_COMPOSE_PROFILE=controllogix`. +### Latency harness proven on ab_server (T21 Step 1) + +`Differential/PollLatencyComparisonTests.cs` (+ `PollLatencyRun.cs`) ran 2026-10-07 against ab_server +(`controllogix` image, host port 44828 because an unrelated outbound socket held 44818 on the docker host), +under the same `--filter "FullyQualifiedName~Differential"` invocation as the differential. The mechanics work: +the timed phase reuses the differential's browse, the anchors, the open budget and the barriers; **1** native Forward Open +in the latency phase; 30 passes per transport, 0 tag failures; native packs 100 concurrent reads into **6** requests per pass. +ab_server has 7 atomic scalars, so the 100-tag set was **repeated cyclically** (reported in the output). + +| Family | libplctag p50 / p95 ms | native p50 / p95 ms | native requests/pass | Criterion | +|---|---|---|---|---| +| CompactLogix (504, the press default) | 7.7 / 10.9 | 22.1 / 25.3 | 6 | **FAIL** | +| ControlLogix (4002) | 7.3 / 9.1 | 7.7 / 9.2 | 2 | pass | + +The CompactLogix result is a real finding, not a harness fault: native opens a 504-byte connection there, while libplctag +opens a Large one (see the T18 capture above), so native needs 3× the round trips per pass. The T21 press result is +recorded under `## Latency (T21)` and is not filled in here. + ## AdminUI /run (T19) Run 2026-10-07 on the local docker-dev rig (OrbStack), against the ab_server `controllogix` fixture diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs index 03b72db1..89e31cbf 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs @@ -44,6 +44,11 @@ internal sealed class DifferentialResult public required int? ConnectionsBaseline { get; init; } public required int? ConnectionsBeforeNative { get; init; } public required TimeSpan LibplctagDrainWait { get; init; } + + /// The endpoint's resolved addresses, and this machine's connections to it before the run (null when + /// unobservable). Reused by the latency comparison's barriers, which must not start alongside a session. + public required IReadOnlySet Addresses { get; init; } + public required IReadOnlySet? BaselineEndpoints { get; init; } } /// @@ -98,7 +103,7 @@ internal static class DifferentialRun // ---- Barrier: libplctag's session must be gone before the native side connects. ---- var drain = Stopwatch.StartNew(); - var beforeNative = await WaitForLibplctagSessionToCloseAsync(target, addresses, baseline, ct).ConfigureAwait(false); + var beforeNative = await WaitForSessionsToCloseAsync(target, addresses, baseline, "libplctag", ct).ConfigureAwait(false); drain.Stop(); // ---- Phase 2: native only, every operation sharing ONE connection held open by the anchor. ---- @@ -114,6 +119,8 @@ internal static class DifferentialRun ConnectionsBaseline = baseline?.Count, ConnectionsBeforeNative = beforeNative, LibplctagDrainWait = drain.Elapsed, + Addresses = addresses, + BaselineEndpoints = baseline, }; } @@ -123,26 +130,42 @@ internal static class DifferentialRun /// a fresh session per tag; an anchor handle on the listing pseudo-tag, held for the whole phase, keeps /// every libplctag operation on ONE session — the mirror of the native anchor lease. /// - internal static async Task<(TransportObservations Observations, IReadOnlyList Selection)> RunLibplctagPhaseAsync( + internal static Task<(TransportObservations Observations, IReadOnlyList Selection)> RunLibplctagPhaseAsync( DifferentialTarget target, CancellationToken ct) { const string phase = "libplctag"; var lib = new TransportObservations(); var factories = AbCipTransportFactories.For(AbCipTransport.Libplctag); - using var anchor = factories.Tags.Create(target.Params("@tags")); + return WithLibplctagAnchorAsync(phase, target, "@tags", async () => + { + await BrowseAsync(phase, target, factories.Enumerators, lib, afterEach: null, ct).ConfigureAwait(false); + await ReadTemplatesAsync(phase, target, factories.Templates, lib, afterEach: null, ct).ConfigureAwait(false); + var selection = Select(target, lib); + await ReadTagsAsync(phase, target, factories.Tags, selection, lib, afterEach: null, ct).ConfigureAwait(false); + return ((TransportObservations)lib, selection); + }, ct); + } + + /// + /// Holds one libplctag handle on for the whole of , so every + /// libplctag operation in it shares ONE session and ONE Forward Open. A failure to create the anchor aborts + /// before runs; the anchor is disposed when ends, either way. + /// Shared by the differential (anchor @tags) and the latency comparison (anchor = its first tag). + /// + internal static async Task WithLibplctagAnchorAsync( + string phase, DifferentialTarget target, string anchorTag, Func> body, CancellationToken ct) + { + var factories = AbCipTransportFactories.For(AbCipTransport.Libplctag); + using var anchor = factories.Tags.Create(target.Params(anchorTag)); try { await anchor.InitializeAsync(ct).ConfigureAwait(false); } catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested) { - throw Abort(phase, "anchor session (@tags)", ex); + throw Abort(phase, $"anchor session ({anchorTag})", ex); } - await BrowseAsync(phase, target, factories.Enumerators, lib, afterEach: null, ct).ConfigureAwait(false); - await ReadTemplatesAsync(phase, target, factories.Templates, lib, afterEach: null, ct).ConfigureAwait(false); - var selection = Select(target, lib); - await ReadTagsAsync(phase, target, factories.Tags, selection, lib, afterEach: null, ct).ConfigureAwait(false); - return (lib, selection); + return await body().ConfigureAwait(false); } /// The native phase alone; internal so the offline abort test can drive it against a fake endpoint. @@ -152,6 +175,27 @@ internal static class DifferentialRun const string phase = "native"; var native = new TransportObservations(); var factories = AbCipTransportFactories.For(AbCipTransport.Native); + var (_, forwardOpens) = await WithNativeAnchorAsync(phase, target, async checkOpens => + { + await BrowseAsync(phase, target, factories.Enumerators, native, checkOpens, ct).ConfigureAwait(false); + await ReadTemplatesAsync(phase, target, factories.Templates, native, checkOpens, ct).ConfigureAwait(false); + await ReadTagsAsync(phase, target, factories.Tags, selection, native, checkOpens, ct).ConfigureAwait(false); + return true; + }).ConfigureAwait(false); + return (native, forwardOpens); + } + + /// + /// Runs with every native operation sharing ONE connection, held open by an anchor + /// lease: the process-wide Forward Open counter is reset, the open-attempt budget is set to exactly one + /// more attempt, and receives a check to call after each operation that aborts the + /// moment a second Forward Open reaches the wire. The anchor is released and the budget restored in + /// finally. Shared by the differential and the latency comparison. + /// + /// The body's result and the Forward Opens counted between the reset and the anchor release. + internal static async Task<(T Result, int ForwardOpens)> WithNativeAnchorAsync( + string phase, DifferentialTarget target, Func, Task> body) + { var anchorParams = target.Params("@tags"); void CheckOpens(string operation) @@ -165,15 +209,15 @@ internal static class DifferentialRun Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0); Volatile.Write(ref CipConnection.OpenAttemptBudget, Volatile.Read(ref CipConnection.OpenAttempts) + 1); + T result; int forwardOpens; try { + // Acquire does no I/O; the key and options carry no tag-specific field, so every lease shares this entry. var anchor = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(anchorParams), NativeAbCipDevice.Options(anchorParams)); try { - await BrowseAsync(phase, target, factories.Enumerators, native, CheckOpens, ct).ConfigureAwait(false); - await ReadTemplatesAsync(phase, target, factories.Templates, native, CheckOpens, ct).ConfigureAwait(false); - await ReadTagsAsync(phase, target, factories.Tags, selection, native, CheckOpens, ct).ConfigureAwait(false); + result = await body(CheckOpens).ConfigureAwait(false); } finally { @@ -185,7 +229,7 @@ internal static class DifferentialRun { Volatile.Write(ref CipConnection.OpenAttemptBudget, -1); } - return (native, forwardOpens); + return (result, forwardOpens); } private static async Task BrowseAsync( @@ -422,11 +466,11 @@ internal static class DifferentialRun _ => true, }; - private static DifferentialAbortedException Abort(string phase, string operation, Exception ex) => + internal static DifferentialAbortedException Abort(string phase, string operation, Exception ex) => new($"Differential ABORTED in the {phase} phase at {operation}: {Describe(ex)}. " + "Stopped on the first connection-level failure; nothing was retried.", ex); - private static Task PaceAsync(DifferentialTarget target, CancellationToken ct) => + internal static Task PaceAsync(DifferentialTarget target, CancellationToken ct) => target.PaceMs > 0 ? Task.Delay(target.PaceMs, ct) : Task.CompletedTask; /// Resolves the endpoint once, bounded (an IP literal needs no lookup). @@ -448,14 +492,16 @@ internal static class DifferentialRun } /// - /// libplctag closes its session on a background thread some time after the last handle is destroyed. - /// Wait until no TCP connection to the endpoint remains that was not already there before the run - /// (compared by local endpoint, not by count — review r1 M6), so the two transports are never + /// A transport closes its session asynchronously (libplctag on a background thread some time after the last + /// handle is destroyed). Wait until no TCP connection to the endpoint remains that was not already there + /// before the run (compared by local endpoint, not by count — review r1 M6), so two sessions are never /// connected to the controller at the same time. /// - /// The connection count observed when native work may start (null when unobservable). - private static async Task WaitForLibplctagSessionToCloseAsync( - DifferentialTarget target, IReadOnlySet addresses, IReadOnlySet? baseline, CancellationToken ct) + /// Names the session waited for, in the abort message. + /// The connection count observed when the next phase may start (null when unobservable). + internal static async Task WaitForSessionsToCloseAsync( + DifferentialTarget target, IReadOnlySet addresses, IReadOnlySet? baseline, string whose, + CancellationToken ct) { if (baseline is null) { @@ -471,8 +517,8 @@ internal static class DifferentialRun if (added.Count == 0) return now.Count; if (deadline.Elapsed > LibplctagDrainBudget) throw new DifferentialAbortedException( - $"libplctag's session to {target.Host}:{target.Port} was still open {LibplctagDrainBudget.TotalSeconds:0}s after its " + - $"handles were disposed (local {string.Join(", ", added)}); refusing to start the native phase alongside it."); + $"{whose}'s session to {target.Host}:{target.Port} was still open {LibplctagDrainBudget.TotalSeconds:0}s after it was " + + $"released (local {string.Join(", ", added)}); refusing to start the next phase alongside it."); await Task.Delay(TimeSpan.FromMilliseconds(250), ct).ConfigureAwait(false); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs index 6f48e563..5a6bbd3a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs @@ -102,7 +102,8 @@ public sealed class DifferentialSafetyTests [Fact] public void Volume_knobs_fail_fast_above_the_ceiling_instead_of_clamping() { - var saved = Snapshot(DifferentialTarget.EndpointVar, DifferentialTarget.MaxTagsVar, DifferentialTarget.MaxTemplatesVar, DifferentialTarget.PaceVar); + var saved = Snapshot(DifferentialTarget.EndpointVar, DifferentialTarget.MaxTagsVar, DifferentialTarget.MaxTemplatesVar, DifferentialTarget.PaceVar, + DifferentialTarget.LatencyTagsVar, DifferentialTarget.LatencyPassesVar, DifferentialTarget.LatencyPassGapVar); try { Environment.SetEnvironmentVariable(DifferentialTarget.EndpointVar, "127.0.0.1:1"); @@ -115,10 +116,29 @@ public sealed class DifferentialSafetyTests Environment.SetEnvironmentVariable(DifferentialTarget.MaxTemplatesVar, null); Environment.SetEnvironmentVariable(DifferentialTarget.PaceVar, null); - var t = DifferentialTarget.FromEnvironment()!; + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyTagsVar, null); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassesVar, null); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassGapVar, null); + var t =DifferentialTarget.FromEnvironment()!; t.MaxTags.ShouldBe(500); t.MaxTemplates.ShouldBe(DifferentialTarget.DefaultMaxTemplates); t.PaceMs.ShouldBe(DifferentialTarget.DefaultPaceMs); + t.LatencyTags.ShouldBe(100); + t.LatencyPasses.ShouldBe(30); + t.LatencyPassGapMs.ShouldBe(200); + + // Latency knobs (Task 21): the same fail-fast rule, with tighter ceilings. + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyTagsVar, "201"); + Should.Throw(DifferentialTarget.FromEnvironment).Message.ShouldContain(DifferentialTarget.LatencyTagsVar); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyTagsVar, "200"); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassesVar, "51"); + Should.Throw(DifferentialTarget.FromEnvironment).Message.ShouldContain(DifferentialTarget.LatencyPassesVar); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassesVar, "50"); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassGapVar, "10001"); + Should.Throw(DifferentialTarget.FromEnvironment).Message.ShouldContain(DifferentialTarget.LatencyPassGapVar); + Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassGapVar, "0"); + var lat = DifferentialTarget.FromEnvironment()!; + (lat.LatencyTags, lat.LatencyPasses, lat.LatencyPassGapMs).ShouldBe((200, 50, 0)); } finally { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialTarget.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialTarget.cs index c72617ea..c19c630b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialTarget.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialTarget.cs @@ -14,6 +14,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential; /// AB_DIFF_MAX_TEMPLATES — cap on template reads per transport, default 50, at most 500. /// AB_DIFF_PACE_MS — delay before every harness-issued request, default 20 ms; 0 for ab_server. /// AB_DIFF_COUNT_OPENS — 1 ⇒ assert exactly one native Forward Open over the run. +/// AB_LAT_TAGS — latency comparison: tags read per pass, default 100, at most 200 (higher fails fast). +/// AB_LAT_PASSES — latency comparison: passes per transport, default 30, at most 50. +/// AB_LAT_PASS_GAP_MS — latency comparison: pause between passes, default 200 ms, at most 10000. /// /// internal sealed record DifferentialTarget( @@ -25,7 +28,10 @@ internal sealed record DifferentialTarget( int MaxTags, bool CountOpens, int MaxTemplates = DifferentialTarget.DefaultMaxTemplates, - int PaceMs = DifferentialTarget.DefaultPaceMs) + int PaceMs = DifferentialTarget.DefaultPaceMs, + int LatencyTags = DifferentialTarget.DefaultLatencyTags, + int LatencyPasses = DifferentialTarget.DefaultLatencyPasses, + int LatencyPassGapMs = DifferentialTarget.DefaultLatencyPassGapMs) { public const string EndpointVar = "AB_DIFF_ENDPOINT"; public const string PathVar = "AB_DIFF_PATH"; @@ -35,6 +41,16 @@ internal sealed record DifferentialTarget( public const string CountOpensVar = "AB_DIFF_COUNT_OPENS"; public const string MaxTemplatesVar = "AB_DIFF_MAX_TEMPLATES"; public const string PaceVar = "AB_DIFF_PACE_MS"; + public const string LatencyTagsVar = "AB_LAT_TAGS"; + public const string LatencyPassesVar = "AB_LAT_PASSES"; + public const string LatencyPassGapVar = "AB_LAT_PASS_GAP_MS"; + + /// Latency comparison ceilings: a production press is the target, and each pass is one burst. + public const int LatencyTagsCeiling = 200; + public const int LatencyPassesCeiling = 50; + public const int DefaultLatencyTags = 100; + public const int DefaultLatencyPasses = 30; + public const int DefaultLatencyPassGapMs = 200; /// Hard ceiling on AB_DIFF_MAX_TAGS / AB_DIFF_MAX_TEMPLATES: a production press is the target. public const int Ceiling = 500; @@ -86,8 +102,13 @@ internal sealed record DifferentialTarget( if (tags is { Count: > Ceiling }) throw new InvalidOperationException($"{TagsVar} names {tags.Count} tags; the ceiling is {Ceiling}."); + var latencyTags = BoundedInt(LatencyTagsVar, DefaultLatencyTags, 1, LatencyTagsCeiling); + var latencyPasses = BoundedInt(LatencyPassesVar, DefaultLatencyPasses, 1, LatencyPassesCeiling); + var latencyGap = BoundedInt(LatencyPassGapVar, DefaultLatencyPassGapMs, 0, 10_000); + var countOpens = Environment.GetEnvironmentVariable(CountOpensVar) == "1"; - return new DifferentialTarget(host, port, path, family, tags, max, countOpens, maxTemplates, pace); + return new DifferentialTarget(host, port, path, family, tags, max, countOpens, maxTemplates, pace, + latencyTags, latencyPasses, latencyGap); } /// Reads an integer knob; out of range FAILS rather than clamping, so a typo never silently widens a press run. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs new file mode 100644 index 00000000..3397ce98 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs @@ -0,0 +1,117 @@ +using System.Globalization; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential; + +/// +/// Task 21 latency comparison: a fixed set of atomic tags polled through libplctag, then through the native +/// transport, each pass issuing every read at once. Pass criterion: native p95 ≤ libplctag p95 × 1.10. +/// Gated by the same variables as the differential (AB_DIFF_ENDPOINT absent ⇒ skip), in the same +/// serialized collection, and read-only under the same source/IL guard +/// (). It runs AFTER the +/// differential's one run, reusing its browse for tag selection, so the target sees one listing per transport, +/// not two. Knobs: AB_LAT_TAGS, AB_LAT_PASSES, AB_LAT_PASS_GAP_MS (see +/// ). +/// +[Collection(DifferentialCollection.Name)] +[Trait("Category", "LiveIntegration")] +public sealed class PollLatencyComparisonTests(DifferentialRunFixture fixture) +{ + private const double Tolerance = 1.10; + private static readonly TimeSpan RunBudget = TimeSpan.FromMinutes(10); + + [Fact] + public async Task Native_poll_p95_is_within_ten_percent_of_libplctag() + { + var target = DifferentialTarget.FromEnvironment(); + if (target is null) Assert.Skip(DifferentialTarget.SkipReason); + var differential = await fixture.GetAsync(); + + using var cts = new CancellationTokenSource(RunBudget); + var r = await PollLatencyRun.ExecuteAsync(target, differential, cts.Token); + + Log($"## Latency — {r.Target}"); + Log($"Tags per pass: {r.Tags.Count} ({r.DistinctTags} distinct" + + (r.Repeated ? $", REPEATED cyclically to reach {DifferentialTarget.LatencyTagsVar}={r.Target.LatencyTags}" : "") + + $"); passes per transport: {r.Target.LatencyPasses}; gap between passes: {r.Target.LatencyPassGapMs} ms."); + Log("Types: " + string.Join(", ", r.Tags.GroupBy(t => t.Type).Select(g => $"{g.Key} {g.Count()}"))); + if (r.Dropped.Count > 0) Log($"Dropped (libplctag refused at setup, removed from both sets): {string.Join(", ", r.Dropped)}"); + Log($"Native requests spent on setup (one read per tag): {r.NativeSetupRequests}; native Forward Opens: {r.NativeForwardOpens}."); + Log(""); + + var libP50 = PollLatencyRun.Percentile(r.Libplctag.Select(p => p.Wall), 50); + var libP95 = PollLatencyRun.Percentile(r.Libplctag.Select(p => p.Wall), 95); + var natP50 = PollLatencyRun.Percentile(r.Native.Select(p => p.Wall), 50); + var natP95 = PollLatencyRun.Percentile(r.Native.Select(p => p.Wall), 95); + var limit = libP95 * Tolerance; + + Log("| Transport | Passes | p50 ms | p95 ms | min ms | max ms | CIP requests per pass (median) | Tag failures |"); + Log("|---|---|---|---|---|---|---|---|"); + Log(Row("libplctag", r.Libplctag, libP50, libP95)); + Log(Row("native", r.Native, natP50, natP95)); + Log(""); + Log($"Criterion: native p95 {Ms(natP95)} ms ≤ libplctag p95 {Ms(libP95)} ms × {Tolerance:0.00} = {Ms(limit)} ms → " + + (natP95 <= limit ? "PASS" : "**FAIL**")); + Log(""); + Log("| Pass | libplctag ms | native ms | native requests |"); + Log("|---|---|---|---|"); + for (var i = 0; i < Math.Max(r.Libplctag.Count, r.Native.Count); i++) + { + var lib = i < r.Libplctag.Count ? r.Libplctag[i] : null; + var nat = i < r.Native.Count ? r.Native[i] : null; + Log($"| {i + 1} | {(lib is null ? "-" : Ms(lib.Wall.TotalMilliseconds))} | {(nat is null ? "-" : Ms(nat.Wall.TotalMilliseconds))} | " + + $"{nat?.Requests?.ToString(CultureInfo.InvariantCulture) ?? "-"} |"); + } + foreach (var p in r.Problems) Log($"- PROBLEM: {p}"); + + r.Problems.ShouldBeEmpty(); + r.Libplctag.Count.ShouldBe(r.Target.LatencyPasses); + r.Native.Count.ShouldBe(r.Target.LatencyPasses); + r.Libplctag.Sum(p => p.TagFailures).ShouldBe(0, "every timed libplctag read must succeed"); + r.Native.Sum(p => p.TagFailures).ShouldBe(0, "every timed native read must succeed"); + if (r.Target.CountOpens) r.NativeForwardOpens.ShouldBe(1); + natP95.ShouldBeLessThanOrEqualTo(limit, $"native p95 must be ≤ libplctag p95 × {Tolerance:0.00}"); + } + + /// Offline: the tag pick takes atomic scalars only, and repeats a short set deliberately and visibly. + [Fact] + public void Tag_pick_takes_atomic_scalars_only_and_repeats_a_short_set_visibly() + { + var browsed = new TransportObservations(); + browsed.Browse.Add(new AbCipDiscoveredTag("D", null, AbCipDataType.DInt, false)); + browsed.Browse.Add(new AbCipDiscoveredTag("R", null, AbCipDataType.Real, false)); + browsed.Browse.Add(new AbCipDiscoveredTag("S", null, AbCipDataType.String, false)); + browsed.Browse.Add(new AbCipDiscoveredTag("T", null, AbCipDataType.Dt, false)); + browsed.Browse.Add(new AbCipDiscoveredTag("A", null, AbCipDataType.DInt, false, ElementCount: 4, IsArray: true)); + browsed.Browse.Add(new AbCipDiscoveredTag("Sys", null, AbCipDataType.DInt, false, IsSystemTag: true)); + var target = new DifferentialTarget("127.0.0.1", 1, "1,0", AbCipPlcFamily.CompactLogix, null, 200, false, LatencyTags: 5); + + var (tags, distinct, repeated) = PollLatencyRun.PickTags(target, browsed); + distinct.ShouldBe(2); + repeated.ShouldBeTrue(); + tags.Select(t => t.Name).ShouldBe(["D", "R", "D", "R", "D"]); + + var (one, oneDistinct, oneRepeated) = PollLatencyRun.PickTags(target with { LatencyTags = 1 }, browsed); + one.ShouldHaveSingleItem().Name.ShouldBe("D"); + oneDistinct.ShouldBe(1); + oneRepeated.ShouldBeFalse(); + + PollLatencyRun.Percentile([TimeSpan.FromMilliseconds(3), TimeSpan.FromMilliseconds(1), TimeSpan.FromMilliseconds(2)], 50) + .ShouldBe(2); + PollLatencyRun.Percentile(Enumerable.Range(1, 20).Select(i => TimeSpan.FromMilliseconds(i)), 95).ShouldBe(19); + } + + private static string Row(string name, IReadOnlyList passes, double p50, double p95) + { + var walls = passes.Select(p => p.Wall.TotalMilliseconds).ToList(); + var requests = passes.Where(p => p.Requests is not null).Select(p => p.Requests!.Value).Order().ToList(); + var medianRequests = requests.Count == 0 ? "n/a" : requests[requests.Count / 2].ToString(CultureInfo.InvariantCulture); + return $"| {name} | {passes.Count} | {Ms(p50)} | {Ms(p95)} | {Ms(walls.Min())} | {Ms(walls.Max())} | {medianRequests} | " + + $"{passes.Sum(p => p.TagFailures)} |"; + } + + private static string Ms(double ms) => ms.ToString("0.0", CultureInfo.InvariantCulture); + + private static void Log(string line) => TestContext.Current.TestOutputHelper?.WriteLine(line); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs new file mode 100644 index 00000000..6a95ecb3 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs @@ -0,0 +1,240 @@ +using System.Diagnostics; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential; + +/// One timed burst: every tag read once, concurrently. +/// Native only: delta over the pass (keep-alives included). +/// Reads that ended with a tag-level status (a connection-level one aborts instead). +internal sealed record LatencyPass(int Index, TimeSpan Wall, int? Requests, int TagFailures); + +/// The outcome of the latency comparison. +internal sealed class LatencyResult +{ + public required DifferentialTarget Target { get; init; } + + /// The fixed set read every pass, in order (a name appears more than once only when ). + public required IReadOnlyList Tags { get; init; } + public required int DistinctTags { get; init; } + public required bool Repeated { get; init; } + + /// Names libplctag could not read at setup (tag-level status): dropped from BOTH transports' sets. + public required IReadOnlyList Dropped { get; init; } + public required IReadOnlyList Libplctag { get; init; } + public required IReadOnlyList Native { get; init; } + public required int NativeForwardOpens { get; init; } + + /// Native requests spent creating the runtimes (one read each, before any timed pass). + public required int NativeSetupRequests { get; init; } + public required List Problems { get; init; } +} + +/// +/// Task 21's poll-latency comparison, read-only and built on the differential's machinery: the same +/// endpoint parsing, the same anchors (one libplctag handle, one native lease), the same open-attempt budget, +/// the same connection barrier and the same stop-on-first-connection-failure rule. +/// Order: the differential's run (its browse supplies the tags — no second listing), a barrier until its +/// native session is gone, every libplctag pass with every handle then disposed, a barrier until libplctag's +/// session is gone, then every native pass under one anchor lease. The two transports never overlap. +/// Per transport: create and read every tag once (untimed, paced by AB_DIFF_PACE_MS), then +/// AB_LAT_PASSES timed passes, each issuing all AB_LAT_TAGS reads at once (Task.WhenAll) +/// so the native shared scheduler can pack them into Multiple Service Packets as libplctag packs internally, +/// with AB_LAT_PASS_GAP_MS of quiet between passes. +/// +internal static class PollLatencyRun +{ + private static readonly HashSet Atomic = + [ + AbCipDataType.Bool, AbCipDataType.SInt, AbCipDataType.Int, AbCipDataType.DInt, AbCipDataType.LInt, + AbCipDataType.USInt, AbCipDataType.UInt, AbCipDataType.UDInt, AbCipDataType.ULInt, + AbCipDataType.Real, AbCipDataType.LReal, + ]; + + /// + /// The fixed tag set: atomic, non-array scalars, from the differential's libplctag browse — or, when + /// AB_DIFF_TAGS is set, those of its names whose browsed type is an atomic scalar. When fewer + /// distinct tags exist than AB_LAT_TAGS, the set is REPEATED cyclically up to the count (each + /// repeat is its own runtime/handle on the same name) and reported as such — never silently. + /// + internal static (IReadOnlyList Tags, int Distinct, bool Repeated) PickTags( + DifferentialTarget target, TransportObservations browsed) + { + var candidates = DifferentialRun.Select(target with { MaxTags = DifferentialTarget.Ceiling }, browsed) + .Where(t => !t.IsArray && t.Type is { } type && Atomic.Contains(type)) + .DistinctBy(t => t.Name) + .Take(target.LatencyTags) + .ToList(); + if (candidates.Count == 0 || candidates.Count >= target.LatencyTags) return (candidates, candidates.Count, false); + var repeated = Enumerable.Range(0, target.LatencyTags).Select(i => candidates[i % candidates.Count]).ToList(); + return (repeated, candidates.Count, true); + } + + public static async Task ExecuteAsync(DifferentialTarget target, DifferentialResult differential, CancellationToken ct) + { + var (picked, distinct, repeated) = PickTags(target, differential.Libplctag); + if (picked.Count == 0) + throw new InvalidOperationException("The browse offered no atomic scalar tags to time; nothing was read."); + + // The differential's native session must be gone before libplctag connects. + await DifferentialRun.WaitForSessionsToCloseAsync( + target, differential.Addresses, differential.BaselineEndpoints, "the differential's native", ct).ConfigureAwait(false); + + // ---- libplctag: setup, passes, every handle disposed (anchor last). ---- + var dropped = new List(); + var libFactory = AbCipTransportFactories.For(AbCipTransport.Libplctag).Tags; + // Anchor on a tag the differential already read Ok through libplctag, so the anchor is never the refusal. + var anchorTag = picked.FirstOrDefault(t => differential.Libplctag.Reads.TryGetValue(t.Name, out var o) && o.Succeeded) + ?? picked[0]; + var (libPasses, tags) = await DifferentialRun.WithLibplctagAnchorAsync("latency libplctag", target, anchorTag.Name, async () => + { + var runtimes = new List<(SelectedTag Tag, IAbCipTagRuntime Runtime)>(); + try + { + foreach (var tag in picked) + { + var rt = await CreateAsync("latency libplctag", target, libFactory, tag, ct).ConfigureAwait(false); + if (rt is null) { dropped.Add(tag.Name); continue; } + runtimes.Add((tag, rt)); + } + var passes = await RunPassesAsync("latency libplctag", target, runtimes, countRequests: false, afterEach: null, ct) + .ConfigureAwait(false); + return (passes, (IReadOnlyList)runtimes.Select(r => r.Tag).ToList()); + } + finally + { + foreach (var (_, rt) in runtimes) rt.Dispose(); + } + }, ct).ConfigureAwait(false); + + // ---- Barrier: libplctag's session must be gone before the native side connects. ---- + await DifferentialRun.WaitForSessionsToCloseAsync( + target, differential.Addresses, differential.BaselineEndpoints, "libplctag", ct).ConfigureAwait(false); + + // ---- native: the same (post-drop) set, under one anchor lease. ---- + var problems = new List(); + var nativeFactory = AbCipTransportFactories.For(AbCipTransport.Native).Tags; + var ((nativePasses, setupRequests), opens) = await DifferentialRun.WithNativeAnchorAsync("latency native", target, async checkOpens => + { + var runtimes = new List<(SelectedTag Tag, IAbCipTagRuntime Runtime)>(); + try + { + var before = Volatile.Read(ref CipConnection.RequestsSent); + foreach (var tag in tags) + { + var rt = await CreateAsync("latency native", target, nativeFactory, tag, ct).ConfigureAwait(false); + checkOpens($"setup of '{tag.Name}'"); + if (rt is null) { problems.Add($"native could not read '{tag.Name}', which libplctag read"); continue; } + runtimes.Add((tag, rt)); + } + var setup = Volatile.Read(ref CipConnection.RequestsSent) - before; + var passes = await RunPassesAsync("latency native", target, runtimes, countRequests: true, checkOpens, ct) + .ConfigureAwait(false); + return (passes, setup); + } + finally + { + foreach (var (_, rt) in runtimes) rt.Dispose(); + } + }).ConfigureAwait(false); + + return new LatencyResult + { + Target = target, + Tags = tags, + DistinctTags = distinct, + Repeated = repeated, + Dropped = dropped, + Libplctag = libPasses, + Native = nativePasses, + NativeForwardOpens = opens, + NativeSetupRequests = setupRequests, + Problems = problems, + }; + } + + /// Create + first read of one tag. Null on a tag-level refusal; a connection-level failure aborts. + private static async Task CreateAsync( + string phase, DifferentialTarget target, IAbCipTagFactory factory, SelectedTag tag, CancellationToken ct) + { + var rt = factory.Create(target.Params(tag.Name, tag.ElementCount, tag.IsArray, tag.Type)); + try + { + await DifferentialRun.PaceAsync(target, ct).ConfigureAwait(false); + await rt.InitializeAsync(ct).ConfigureAwait(false); + if (rt.GetStatus() == (int)AbStatus.Ok) return rt; + } + catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested) + { + if (DifferentialRun.IsConnectionLevel(ex)) + { + rt.Dispose(); + throw DifferentialRun.Abort(phase, $"setup of '{tag.Name}'", ex); + } + } + catch + { + rt.Dispose(); + throw; + } + rt.Dispose(); + return null; + } + + private static async Task> RunPassesAsync( + string phase, DifferentialTarget target, List<(SelectedTag Tag, IAbCipTagRuntime Runtime)> runtimes, + bool countRequests, Action? afterEach, CancellationToken ct) + { + var passes = new List(target.LatencyPasses); + for (var p = 1; p <= target.LatencyPasses; p++) + { + if (p > 1 && target.LatencyPassGapMs > 0) await Task.Delay(target.LatencyPassGapMs, ct).ConfigureAwait(false); + + var requestsBefore = Volatile.Read(ref CipConnection.RequestsSent); + var clock = Stopwatch.StartNew(); + var outcomes = await Task.WhenAll(runtimes.Select(r => ReadOnceAsync(r.Runtime, ct))).ConfigureAwait(false); + clock.Stop(); + var requests = Volatile.Read(ref CipConnection.RequestsSent) - requestsBefore; + + var failures = 0; + for (var i = 0; i < outcomes.Length; i++) + { + if (outcomes[i] is { } ex) + { + if (DifferentialRun.IsConnectionLevel(ex)) + throw DifferentialRun.Abort(phase, $"pass {p}, tag '{runtimes[i].Tag.Name}'", ex); + failures++; + } + else if (runtimes[i].Runtime.GetStatus() != (int)AbStatus.Ok) + { + failures++; + } + } + afterEach?.Invoke($"pass {p}"); + passes.Add(new LatencyPass(p, clock.Elapsed, countRequests ? requests : null, failures)); + } + return passes; + } + + private static async Task ReadOnceAsync(IAbCipTagRuntime runtime, CancellationToken ct) + { + try + { + await runtime.ReadAsync(ct).ConfigureAwait(false); + return null; + } + catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested) + { + return ex; + } + } + + /// Nearest-rank percentile (p in (0,100]) of the pass wall times, in milliseconds. + internal static double Percentile(IEnumerable walls, double p) + { + var sorted = walls.Select(w => w.TotalMilliseconds).Order().ToList(); + if (sorted.Count == 0) throw new InvalidOperationException("No passes to take a percentile of."); + var rank = (int)Math.Ceiling(p / 100.0 * sorted.Count); + return sorted[Math.Clamp(rank, 1, sorted.Count) - 1]; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/TransportDifferentialTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/TransportDifferentialTests.cs index b5b33e47..d4b4494f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/TransportDifferentialTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/TransportDifferentialTests.cs @@ -169,6 +169,10 @@ public sealed class TransportDifferentialTests(DifferentialRunFixture fixture) var dir = Path.GetDirectoryName(ThisFile())!; var sources = Directory.GetFiles(dir, "*.cs"); sources.ShouldContain(f => Path.GetFileName(f) == nameof(TransportDifferentialTests) + ".cs"); + // Task 21's latency comparison and its run helper are part of the harness the press sees: pinned in scope, + // so moving either out of this folder (or namespace) fails the guard instead of silently escaping it. + sources.ShouldContain(f => Path.GetFileName(f) == nameof(PollLatencyComparisonTests) + ".cs"); + sources.ShouldContain(f => Path.GetFileName(f) == nameof(PollLatencyRun) + ".cs"); var textHits = (from f in sources let text = File.ReadAllText(f) from token in forbidden @@ -179,6 +183,8 @@ public sealed class TransportDifferentialTests(DifferentialRunFixture fixture) var harnessTypes = typeof(TransportDifferentialTests).Assembly.GetTypes() .Where(t => t.Namespace == typeof(TransportDifferentialTests).Namespace).ToList(); harnessTypes.ShouldContain(typeof(TransportDifferentialTests)); + harnessTypes.ShouldContain(typeof(PollLatencyComparisonTests)); + harnessTypes.ShouldContain(typeof(PollLatencyRun)); var callHits = (from t in harnessTypes from m in MethodsOf(t) from callee in CalleesOf(m)