Joseph Doherty
241 lines
12 KiB
C#
241 lines
12 KiB
C#
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];
|
|
}
|
|
}
|