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)