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

test(abcip): read-only poll-latency comparison harness for the T21 press gate (proven on ab_server)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-07 19:26:23 -04:00
parent 3ea72f29dd
commit 2dc55ecb8f
7 changed files with 495 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/2026-10-07-native-ab-transport-gate-results.md
+18
View File
@@ -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 The ab_server controllogix matrix is read-only and unaffected: 15 passed and 11 skipped with
`AB_SERVER_COMPOSE_PROFILE=controllogix`. `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) ## AdminUI /run (T19)
Run 2026-10-07 on the local docker-dev rig (OrbStack), against the ab_server `controllogix` fixture Run 2026-10-07 on the local docker-dev rig (OrbStack), against the ab_server `controllogix` fixture
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs
+69 -23
View File
@@ -44,6 +44,11 @@ internal sealed class DifferentialResult
public required int? ConnectionsBaseline { get; init; } public required int? ConnectionsBaseline { get; init; }
public required int? ConnectionsBeforeNative { get; init; } public required int? ConnectionsBeforeNative { get; init; }
public required TimeSpan LibplctagDrainWait { get; init; } public required TimeSpan LibplctagDrainWait { get; init; }
/// <summary>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.</summary>
public required IReadOnlySet<IPAddress> Addresses { get; init; }
public required IReadOnlySet<IPEndPoint>? BaselineEndpoints { get; init; }
} }
/// <summary> /// <summary>
@@ -98,7 +103,7 @@ internal static class DifferentialRun
// ---- Barrier: libplctag's session must be gone before the native side connects. ---- // ---- Barrier: libplctag's session must be gone before the native side connects. ----
var drain = Stopwatch.StartNew(); 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(); drain.Stop();
// ---- Phase 2: native only, every operation sharing ONE connection held open by the anchor. ---- // ---- 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, ConnectionsBaseline = baseline?.Count,
ConnectionsBeforeNative = beforeNative, ConnectionsBeforeNative = beforeNative,
LibplctagDrainWait = drain.Elapsed, 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 /// 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. /// every libplctag operation on ONE session — the mirror of the native anchor lease.
/// </summary> /// </summary>
internal static async Task<(TransportObservations Observations, IReadOnlyList<SelectedTag> Selection)> RunLibplctagPhaseAsync( internal static Task<(TransportObservations Observations, IReadOnlyList<SelectedTag> Selection)> RunLibplctagPhaseAsync(
DifferentialTarget target, CancellationToken ct) DifferentialTarget target, CancellationToken ct)
{ {
const string phase = "libplctag"; const string phase = "libplctag";
var lib = new TransportObservations(); var lib = new TransportObservations();
var factories = AbCipTransportFactories.For(AbCipTransport.Libplctag); 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);
}
/// <summary>
/// Holds one libplctag handle on <paramref name="anchorTag"/> for the whole of <paramref name="body"/>, so every
/// libplctag operation in it shares ONE session and ONE Forward Open. A failure to create the anchor aborts
/// before <paramref name="body"/> runs; the anchor is disposed when <paramref name="body"/> ends, either way.
/// Shared by the differential (anchor <c>@tags</c>) and the latency comparison (anchor = its first tag).
/// </summary>
internal static async Task<T> WithLibplctagAnchorAsync<T>(
string phase, DifferentialTarget target, string anchorTag, Func<Task<T>> body, CancellationToken ct)
{
var factories = AbCipTransportFactories.For(AbCipTransport.Libplctag);
using var anchor = factories.Tags.Create(target.Params(anchorTag));
try try
{ {
await anchor.InitializeAsync(ct).ConfigureAwait(false); await anchor.InitializeAsync(ct).ConfigureAwait(false);
} }
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested) 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); return await body().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);
} }
/// <summary>The native phase alone; internal so the offline abort test can drive it against a fake endpoint.</summary> /// <summary>The native phase alone; internal so the offline abort test can drive it against a fake endpoint.</summary>
@@ -152,6 +175,27 @@ internal static class DifferentialRun
const string phase = "native"; const string phase = "native";
var native = new TransportObservations(); var native = new TransportObservations();
var factories = AbCipTransportFactories.For(AbCipTransport.Native); 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);
}
/// <summary>
/// Runs <paramref name="body"/> 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 <paramref name="body"/> 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
/// <c>finally</c>. Shared by the differential and the latency comparison.
/// </summary>
/// <returns>The body's result and the Forward Opens counted between the reset and the anchor release.</returns>
internal static async Task<(T Result, int ForwardOpens)> WithNativeAnchorAsync<T>(
string phase, DifferentialTarget target, Func<Action<string>, Task<T>> body)
{
var anchorParams = target.Params("@tags"); var anchorParams = target.Params("@tags");
void CheckOpens(string operation) void CheckOpens(string operation)
@@ -165,15 +209,15 @@ internal static class DifferentialRun
Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0); Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0);
Volatile.Write(ref CipConnection.OpenAttemptBudget, Volatile.Read(ref CipConnection.OpenAttempts) + 1); Volatile.Write(ref CipConnection.OpenAttemptBudget, Volatile.Read(ref CipConnection.OpenAttempts) + 1);
T result;
int forwardOpens; int forwardOpens;
try 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)); var anchor = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(anchorParams), NativeAbCipDevice.Options(anchorParams));
try try
{ {
await BrowseAsync(phase, target, factories.Enumerators, native, CheckOpens, ct).ConfigureAwait(false); result = await body(CheckOpens).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);
} }
finally finally
{ {
@@ -185,7 +229,7 @@ internal static class DifferentialRun
{ {
Volatile.Write(ref CipConnection.OpenAttemptBudget, -1); Volatile.Write(ref CipConnection.OpenAttemptBudget, -1);
} }
return (native, forwardOpens); return (result, forwardOpens);
} }
private static async Task BrowseAsync( private static async Task BrowseAsync(
@@ -422,11 +466,11 @@ internal static class DifferentialRun
_ => true, _ => 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)}. " + new($"Differential ABORTED in the {phase} phase at {operation}: {Describe(ex)}. " +
"Stopped on the first connection-level failure; nothing was retried.", 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; target.PaceMs > 0 ? Task.Delay(target.PaceMs, ct) : Task.CompletedTask;
/// <summary>Resolves the endpoint once, bounded (an IP literal needs no lookup).</summary> /// <summary>Resolves the endpoint once, bounded (an IP literal needs no lookup).</summary>
@@ -448,14 +492,16 @@ internal static class DifferentialRun
} }
/// <summary> /// <summary>
/// libplctag closes its session on a background thread some time after the last handle is destroyed. /// A transport closes its session asynchronously (libplctag on a background thread some time after the last
/// Wait until no TCP connection to the endpoint remains that was not already there before the run /// handle is destroyed). Wait until no TCP connection to the endpoint remains that was not already there
/// (compared by local endpoint, not by count — review r1 M6), so the two transports are never /// 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. /// connected to the controller at the same time.
/// </summary> /// </summary>
/// <returns>The connection count observed when native work may start (null when unobservable).</returns> /// <param name="whose">Names the session waited for, in the abort message.</param>
private static async Task<int?> WaitForLibplctagSessionToCloseAsync( /// <returns>The connection count observed when the next phase may start (null when unobservable).</returns>
DifferentialTarget target, IReadOnlySet<IPAddress> addresses, IReadOnlySet<IPEndPoint>? baseline, CancellationToken ct) internal static async Task<int?> WaitForSessionsToCloseAsync(
DifferentialTarget target, IReadOnlySet<IPAddress> addresses, IReadOnlySet<IPEndPoint>? baseline, string whose,
CancellationToken ct)
{ {
if (baseline is null) if (baseline is null)
{ {
@@ -471,8 +517,8 @@ internal static class DifferentialRun
if (added.Count == 0) return now.Count; if (added.Count == 0) return now.Count;
if (deadline.Elapsed > LibplctagDrainBudget) if (deadline.Elapsed > LibplctagDrainBudget)
throw new DifferentialAbortedException( throw new DifferentialAbortedException(
$"libplctag's session to {target.Host}:{target.Port} was still open {LibplctagDrainBudget.TotalSeconds:0}s after its " + $"{whose}'s session to {target.Host}:{target.Port} was still open {LibplctagDrainBudget.TotalSeconds:0}s after it was " +
$"handles were disposed (local {string.Join(", ", added)}); refusing to start the native phase alongside it."); $"released (local {string.Join(", ", added)}); refusing to start the next phase alongside it.");
await Task.Delay(TimeSpan.FromMilliseconds(250), ct).ConfigureAwait(false); await Task.Delay(TimeSpan.FromMilliseconds(250), ct).ConfigureAwait(false);
} }
} }
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs
+22 -2
View File
@@ -102,7 +102,8 @@ public sealed class DifferentialSafetyTests
[Fact] [Fact]
public void Volume_knobs_fail_fast_above_the_ceiling_instead_of_clamping() 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 try
{ {
Environment.SetEnvironmentVariable(DifferentialTarget.EndpointVar, "127.0.0.1:1"); 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.MaxTemplatesVar, null);
Environment.SetEnvironmentVariable(DifferentialTarget.PaceVar, 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.MaxTags.ShouldBe(500);
t.MaxTemplates.ShouldBe(DifferentialTarget.DefaultMaxTemplates); t.MaxTemplates.ShouldBe(DifferentialTarget.DefaultMaxTemplates);
t.PaceMs.ShouldBe(DifferentialTarget.DefaultPaceMs); 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<InvalidOperationException>(DifferentialTarget.FromEnvironment).Message.ShouldContain(DifferentialTarget.LatencyTagsVar);
Environment.SetEnvironmentVariable(DifferentialTarget.LatencyTagsVar, "200");
Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassesVar, "51");
Should.Throw<InvalidOperationException>(DifferentialTarget.FromEnvironment).Message.ShouldContain(DifferentialTarget.LatencyPassesVar);
Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassesVar, "50");
Environment.SetEnvironmentVariable(DifferentialTarget.LatencyPassGapVar, "10001");
Should.Throw<InvalidOperationException>(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 finally
{ {
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialTarget.cs
+23 -2
View File
@@ -14,6 +14,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;
/// <item><c>AB_DIFF_MAX_TEMPLATES</c> — cap on template reads per transport, default 50, at most 500.</item> /// <item><c>AB_DIFF_MAX_TEMPLATES</c> — cap on template reads per transport, default 50, at most 500.</item>
/// <item><c>AB_DIFF_PACE_MS</c> — delay before every harness-issued request, default 20 ms; 0 for ab_server.</item> /// <item><c>AB_DIFF_PACE_MS</c> — delay before every harness-issued request, default 20 ms; 0 for ab_server.</item>
/// <item><c>AB_DIFF_COUNT_OPENS</c> — <c>1</c> ⇒ assert exactly one native Forward Open over the run.</item> /// <item><c>AB_DIFF_COUNT_OPENS</c> — <c>1</c> ⇒ assert exactly one native Forward Open over the run.</item>
/// <item><c>AB_LAT_TAGS</c> — latency comparison: tags read per pass, default 100, at most 200 (higher fails fast).</item>
/// <item><c>AB_LAT_PASSES</c> — latency comparison: passes per transport, default 30, at most 50.</item>
/// <item><c>AB_LAT_PASS_GAP_MS</c> — latency comparison: pause between passes, default 200 ms, at most 10000.</item>
/// </list> /// </list>
/// </summary> /// </summary>
internal sealed record DifferentialTarget( internal sealed record DifferentialTarget(
@@ -25,7 +28,10 @@ internal sealed record DifferentialTarget(
int MaxTags, int MaxTags,
bool CountOpens, bool CountOpens,
int MaxTemplates = DifferentialTarget.DefaultMaxTemplates, 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 EndpointVar = "AB_DIFF_ENDPOINT";
public const string PathVar = "AB_DIFF_PATH"; 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 CountOpensVar = "AB_DIFF_COUNT_OPENS";
public const string MaxTemplatesVar = "AB_DIFF_MAX_TEMPLATES"; public const string MaxTemplatesVar = "AB_DIFF_MAX_TEMPLATES";
public const string PaceVar = "AB_DIFF_PACE_MS"; 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";
/// <summary>Latency comparison ceilings: a production press is the target, and each pass is one burst.</summary>
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;
/// <summary>Hard ceiling on <c>AB_DIFF_MAX_TAGS</c> / <c>AB_DIFF_MAX_TEMPLATES</c>: a production press is the target.</summary> /// <summary>Hard ceiling on <c>AB_DIFF_MAX_TAGS</c> / <c>AB_DIFF_MAX_TEMPLATES</c>: a production press is the target.</summary>
public const int Ceiling = 500; public const int Ceiling = 500;
@@ -86,8 +102,13 @@ internal sealed record DifferentialTarget(
if (tags is { Count: > Ceiling }) if (tags is { Count: > Ceiling })
throw new InvalidOperationException($"{TagsVar} names {tags.Count} tags; the ceiling is {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"; 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);
} }
/// <summary>Reads an integer knob; out of range FAILS rather than clamping, so a typo never silently widens a press run.</summary> /// <summary>Reads an integer knob; out of range FAILS rather than clamping, so a typo never silently widens a press run.</summary>
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs
+117
View File
@@ -0,0 +1,117 @@
using System.Globalization;
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;
/// <summary>
/// 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.
/// <para>Gated by the same variables as the differential (<c>AB_DIFF_ENDPOINT</c> absent ⇒ skip), in the same
/// serialized collection, and read-only under the same source/IL guard
/// (<see cref="TransportDifferentialTests.The_harness_is_read_only_by_construction"/>). 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: <c>AB_LAT_TAGS</c>, <c>AB_LAT_PASSES</c>, <c>AB_LAT_PASS_GAP_MS</c> (see
/// <see cref="DifferentialTarget"/>).</para>
/// </summary>
[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}");
}
/// <summary>Offline: the tag pick takes atomic scalars only, and repeats a short set deliberately and visibly.</summary>
[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<LatencyPass> 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);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs
+240
View File
@@ -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;
/// <summary>One timed burst: every tag read once, concurrently.</summary>
/// <param name="Requests">Native only: <see cref="CipConnection.RequestsSent"/> delta over the pass (keep-alives included).</param>
/// <param name="TagFailures">Reads that ended with a tag-level status (a connection-level one aborts instead).</param>
internal sealed record LatencyPass(int Index, TimeSpan Wall, int? Requests, int TagFailures);
/// <summary>The outcome of the latency comparison.</summary>
internal sealed class LatencyResult
{
public required DifferentialTarget Target { get; init; }
/// <summary>The fixed set read every pass, in order (a name appears more than once only when <see cref="Repeated"/>).</summary>
public required IReadOnlyList<SelectedTag> Tags { get; init; }
public required int DistinctTags { get; init; }
public required bool Repeated { get; init; }
/// <summary>Names libplctag could not read at setup (tag-level status): dropped from BOTH transports' sets.</summary>
public required IReadOnlyList<string> Dropped { get; init; }
public required IReadOnlyList<LatencyPass> Libplctag { get; init; }
public required IReadOnlyList<LatencyPass> Native { get; init; }
public required int NativeForwardOpens { get; init; }
/// <summary>Native requests spent creating the runtimes (one read each, before any timed pass).</summary>
public required int NativeSetupRequests { get; init; }
public required List<string> Problems { get; init; }
}
/// <summary>
/// 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.
/// <para>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.</para>
/// <para>Per transport: create and read every tag once (untimed, paced by <c>AB_DIFF_PACE_MS</c>), then
/// <c>AB_LAT_PASSES</c> timed passes, each issuing all <c>AB_LAT_TAGS</c> reads at once (<c>Task.WhenAll</c>)
/// so the native shared scheduler can pack them into Multiple Service Packets as libplctag packs internally,
/// with <c>AB_LAT_PASS_GAP_MS</c> of quiet between passes.</para>
/// </summary>
internal static class PollLatencyRun
{
private static readonly HashSet<AbCipDataType> Atomic =
[
AbCipDataType.Bool, AbCipDataType.SInt, AbCipDataType.Int, AbCipDataType.DInt, AbCipDataType.LInt,
AbCipDataType.USInt, AbCipDataType.UInt, AbCipDataType.UDInt, AbCipDataType.ULInt,
AbCipDataType.Real, AbCipDataType.LReal,
];
/// <summary>
/// The fixed tag set: atomic, non-array scalars, from the differential's libplctag browse — or, when
/// <c>AB_DIFF_TAGS</c> is set, those of its names whose browsed type is an atomic scalar. When fewer
/// distinct tags exist than <c>AB_LAT_TAGS</c>, 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.
/// </summary>
internal static (IReadOnlyList<SelectedTag> 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<LatencyResult> 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<string>();
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<SelectedTag>)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<string>();
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,
};
}
/// <summary>Create + first read of one tag. Null on a tag-level refusal; a connection-level failure aborts.</summary>
private static async Task<IAbCipTagRuntime?> 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<List<LatencyPass>> RunPassesAsync(
string phase, DifferentialTarget target, List<(SelectedTag Tag, IAbCipTagRuntime Runtime)> runtimes,
bool countRequests, Action<string>? afterEach, CancellationToken ct)
{
var passes = new List<LatencyPass>(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<Exception?> 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;
}
}
/// <summary>Nearest-rank percentile (p in (0,100]) of the pass wall times, in milliseconds.</summary>
internal static double Percentile(IEnumerable<TimeSpan> 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];
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/TransportDifferentialTests.cs
+6
View File
@@ -169,6 +169,10 @@ public sealed class TransportDifferentialTests(DifferentialRunFixture fixture)
var dir = Path.GetDirectoryName(ThisFile())!; var dir = Path.GetDirectoryName(ThisFile())!;
var sources = Directory.GetFiles(dir, "*.cs"); var sources = Directory.GetFiles(dir, "*.cs");
sources.ShouldContain(f => Path.GetFileName(f) == nameof(TransportDifferentialTests) + ".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 var textHits = (from f in sources
let text = File.ReadAllText(f) let text = File.ReadAllText(f)
from token in forbidden from token in forbidden
@@ -179,6 +183,8 @@ public sealed class TransportDifferentialTests(DifferentialRunFixture fixture)
var harnessTypes = typeof(TransportDifferentialTests).Assembly.GetTypes() var harnessTypes = typeof(TransportDifferentialTests).Assembly.GetTypes()
.Where(t => t.Namespace == typeof(TransportDifferentialTests).Namespace).ToList(); .Where(t => t.Namespace == typeof(TransportDifferentialTests).Namespace).ToList();
harnessTypes.ShouldContain(typeof(TransportDifferentialTests)); harnessTypes.ShouldContain(typeof(TransportDifferentialTests));
harnessTypes.ShouldContain(typeof(PollLatencyComparisonTests));
harnessTypes.ShouldContain(typeof(PollLatencyRun));
var callHits = (from t in harnessTypes var callHits = (from t in harnessTypes
from m in MethodsOf(t) from m in MethodsOf(t)
from callee in CalleesOf(m) from callee in CalleesOf(m)