Joseph Doherty
357 lines
16 KiB
C#
357 lines
16 KiB
C#
using System.Collections;
|
|
using System.Diagnostics;
|
|
using System.Net;
|
|
using System.Net.NetworkInformation;
|
|
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 tag the harness reads, with the type it decodes under (null ⇒ raw comparison only).</summary>
|
|
internal sealed record SelectedTag(string Name, AbCipDataType? Type, int ElementCount, bool IsArray, string Origin);
|
|
|
|
/// <summary>What one transport observed for one tag.</summary>
|
|
internal sealed record ReadObservation(int Status, string? Error, byte[]? Raw, object? Decoded)
|
|
{
|
|
public bool Succeeded => Status == (int)AbStatus.Ok && Error is null;
|
|
}
|
|
|
|
/// <summary>What one transport observed over its whole phase.</summary>
|
|
internal sealed class TransportObservations
|
|
{
|
|
public List<AbCipDiscoveredTag> Browse { get; } = [];
|
|
public string? BrowseError { get; set; }
|
|
public Dictionary<uint, AbCipUdtShape?> Shapes { get; } = [];
|
|
public Dictionary<uint, string> TemplateErrors { get; } = [];
|
|
public Dictionary<string, ReadObservation> Reads { get; } = new(StringComparer.Ordinal);
|
|
}
|
|
|
|
/// <summary>The outcome of one whole differential run.</summary>
|
|
internal sealed class DifferentialResult
|
|
{
|
|
public required DifferentialTarget Target { get; init; }
|
|
public required TransportObservations Libplctag { get; init; }
|
|
public required TransportObservations Native { get; init; }
|
|
public required IReadOnlyList<SelectedTag> Selection { get; init; }
|
|
|
|
/// <summary>Native Forward Opens counted between the counter reset and the anchor release.</summary>
|
|
public required int NativeForwardOpens { get; init; }
|
|
|
|
/// <summary>TCP connections to the endpoint before libplctag started, and once its handles were disposed.</summary>
|
|
public required int? ConnectionsBaseline { get; init; }
|
|
public required int? ConnectionsBeforeNative { get; init; }
|
|
public required TimeSpan LibplctagDrainWait { get; init; }
|
|
}
|
|
|
|
/// <summary>
|
|
/// Runs the whole comparison ONCE: every libplctag operation first (handles disposed), then — only
|
|
/// once libplctag's session is gone — every native operation under one anchor lease. Reads only:
|
|
/// browse listings, template reads and tag reads. Nothing on this path encodes or sends a write.
|
|
/// </summary>
|
|
internal static class DifferentialRun
|
|
{
|
|
private const int MaxArrays = 20;
|
|
private const int MaxUdtMembers = 20;
|
|
private const int MaxTemplates = 256;
|
|
|
|
/// <summary>Upper bound on waiting for libplctag's session to close before native work may start.</summary>
|
|
private static readonly TimeSpan LibplctagDrainBudget = TimeSpan.FromSeconds(30);
|
|
|
|
/// <summary>Settle used when the platform cannot enumerate TCP connections.</summary>
|
|
private static readonly TimeSpan FallbackSettle = TimeSpan.FromSeconds(5);
|
|
|
|
public static async Task<DifferentialResult> ExecuteAsync(DifferentialTarget target, CancellationToken ct)
|
|
{
|
|
var baseline = CountConnections(target);
|
|
|
|
// ---- Phase 1: libplctag only. Every handle is disposed before the phase ends. ----
|
|
// libplctag shares one session (and one Forward Open) among the handles alive at the same time, and
|
|
// closes it when the last one is destroyed. The per-tag create/read/dispose below would therefore open
|
|
// 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.
|
|
var lib = new TransportObservations();
|
|
var libFactories = AbCipTransportFactories.For(AbCipTransport.Libplctag);
|
|
IReadOnlyList<SelectedTag> selection;
|
|
using (var libAnchor = libFactories.Tags.Create(target.Params("@tags")))
|
|
{
|
|
try
|
|
{
|
|
await libAnchor.InitializeAsync(ct).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested)
|
|
{
|
|
throw new InvalidOperationException(
|
|
$"libplctag could not open its anchor session to {target}: {Describe(ex)}", ex);
|
|
}
|
|
await BrowseAsync(target, libFactories.Enumerators, lib, ct).ConfigureAwait(false);
|
|
await ReadTemplatesAsync(target, libFactories.Templates, lib, ct).ConfigureAwait(false);
|
|
selection = Select(target, lib);
|
|
await ReadTagsAsync(target, libFactories.Tags, selection, lib, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
// ---- Barrier: libplctag's session must be gone before the native side connects. ----
|
|
var drain = Stopwatch.StartNew();
|
|
var beforeNative = await WaitForLibplctagSessionToCloseAsync(target, baseline, ct).ConfigureAwait(false);
|
|
drain.Stop();
|
|
|
|
// ---- Phase 2: native only, every operation sharing ONE connection held open by the anchor. ----
|
|
var native = new TransportObservations();
|
|
var nativeFactories = AbCipTransportFactories.For(AbCipTransport.Native);
|
|
var anchorParams = target.Params("@tags");
|
|
Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0);
|
|
int opens;
|
|
var anchor = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(anchorParams), NativeAbCipDevice.Options(anchorParams));
|
|
try
|
|
{
|
|
await BrowseAsync(target, nativeFactories.Enumerators, native, ct).ConfigureAwait(false);
|
|
await ReadTemplatesAsync(target, nativeFactories.Templates, native, ct).ConfigureAwait(false);
|
|
await ReadTagsAsync(target, nativeFactories.Tags, selection, native, ct).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
opens = Volatile.Read(ref CipConnection.ForwardOpenCount);
|
|
await anchor.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
|
|
return new DifferentialResult
|
|
{
|
|
Target = target,
|
|
Libplctag = lib,
|
|
Native = native,
|
|
Selection = selection,
|
|
NativeForwardOpens = opens,
|
|
ConnectionsBaseline = baseline,
|
|
ConnectionsBeforeNative = beforeNative,
|
|
LibplctagDrainWait = drain.Elapsed,
|
|
};
|
|
}
|
|
|
|
private static async Task BrowseAsync(
|
|
DifferentialTarget target, IAbCipTagEnumeratorFactory factory, TransportObservations into, CancellationToken ct)
|
|
{
|
|
using var enumerator = factory.Create();
|
|
try
|
|
{
|
|
await foreach (var tag in enumerator.EnumerateAsync(target.Params("@tags"), ct).ConfigureAwait(false))
|
|
into.Browse.Add(tag);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
|
|
catch (Exception ex)
|
|
{
|
|
into.BrowseError = Describe(ex);
|
|
}
|
|
}
|
|
|
|
/// <summary>Reads every template id the browse surfaced, then every nested id those shapes name.</summary>
|
|
private static async Task ReadTemplatesAsync(
|
|
DifferentialTarget target, IAbCipTemplateReaderFactory factory, TransportObservations into, CancellationToken ct)
|
|
{
|
|
var queue = new Queue<uint>(into.Browse
|
|
.Where(t => t.TemplateInstanceId is not null)
|
|
.Select(t => t.TemplateInstanceId!.Value)
|
|
.Distinct()
|
|
.Order());
|
|
using var reader = factory.Create();
|
|
while (queue.Count > 0 && into.Shapes.Count + into.TemplateErrors.Count < MaxTemplates)
|
|
{
|
|
var id = queue.Dequeue();
|
|
if (into.Shapes.ContainsKey(id) || into.TemplateErrors.ContainsKey(id)) continue;
|
|
try
|
|
{
|
|
var shape = CipTemplateObjectDecoder.Decode(await reader.ReadAsync(target.Params($"@udt/{id}"), id, ct).ConfigureAwait(false));
|
|
into.Shapes[id] = shape;
|
|
foreach (var nested in shape?.Members ?? [])
|
|
if (nested.NestedTemplateId is { } n) queue.Enqueue(n);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
|
|
catch (Exception ex)
|
|
{
|
|
into.TemplateErrors[id] = Describe(ex);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// The tags to read, chosen from the libplctag browse (the run's first phase), so both transports
|
|
/// read exactly the same list.
|
|
/// </summary>
|
|
internal static IReadOnlyList<SelectedTag> Select(DifferentialTarget target, TransportObservations browsed)
|
|
{
|
|
var byName = new Dictionary<string, SelectedTag>(StringComparer.Ordinal);
|
|
var scalars = new List<SelectedTag>();
|
|
var arrays = new List<SelectedTag>();
|
|
var members = new List<SelectedTag>();
|
|
|
|
foreach (var tag in browsed.Browse.Where(t => !t.IsSystemTag))
|
|
{
|
|
var name = tag.ProgramScope is null ? tag.Name : $"Program:{tag.ProgramScope}.{tag.Name}";
|
|
var type = tag.DataType;
|
|
AbCipUdtShape? shape = null;
|
|
if (type == AbCipDataType.Structure)
|
|
{
|
|
shape = tag.TemplateInstanceId is { } id ? browsed.Shapes.GetValueOrDefault(id) : null;
|
|
// A string-family struct (STRING, STR0016, ...) is read and presented as one String value.
|
|
if (shape is { IsStringFamily: true }) type = AbCipDataType.String;
|
|
}
|
|
|
|
if (type != AbCipDataType.Structure)
|
|
{
|
|
var selected = new SelectedTag(name, type, tag.IsArray ? tag.ElementCount : 1, tag.IsArray,
|
|
tag.IsArray ? "array" : "scalar");
|
|
byName.TryAdd(name, selected);
|
|
(tag.IsArray ? arrays : scalars).Add(selected);
|
|
continue;
|
|
}
|
|
|
|
if (tag.IsArray || shape is null) continue;
|
|
foreach (var member in shape.Members)
|
|
{
|
|
var memberType = member.DataType;
|
|
if (memberType == AbCipDataType.Structure)
|
|
{
|
|
var nested = member.NestedTemplateId is { } n ? browsed.Shapes.GetValueOrDefault(n) : null;
|
|
if (nested is not { IsStringFamily: true }) continue;
|
|
memberType = AbCipDataType.String;
|
|
}
|
|
// Hidden host bytes of packed BOOL members ("ZZZZZZZZZZ..." in Logix) are not addressable.
|
|
if (member.Name.StartsWith("ZZZZZZZZZZ", StringComparison.Ordinal)) continue;
|
|
var isArray = member.ArrayLength > 1;
|
|
var selected = new SelectedTag($"{name}.{member.Name}", memberType, isArray ? member.ArrayLength : 1,
|
|
isArray, "udt member");
|
|
byName.TryAdd(selected.Name, selected);
|
|
members.Add(selected);
|
|
}
|
|
}
|
|
|
|
IEnumerable<SelectedTag> chosen = target.Tags is { } explicitNames
|
|
? explicitNames.Select(n => byName.GetValueOrDefault(n) ?? new SelectedTag(n, null, 1, false, "explicit (type unknown)"))
|
|
: scalars.Concat(arrays.Take(MaxArrays)).Concat(members.Take(MaxUdtMembers));
|
|
return chosen.DistinctBy(t => t.Name).Take(target.MaxTags).ToList();
|
|
}
|
|
|
|
private static async Task ReadTagsAsync(
|
|
DifferentialTarget target, IAbCipTagFactory factory, IReadOnlyList<SelectedTag> selection,
|
|
TransportObservations into, CancellationToken ct)
|
|
{
|
|
foreach (var tag in selection)
|
|
{
|
|
ct.ThrowIfCancellationRequested();
|
|
var runtime = factory.Create(target.Params(tag.Name, tag.ElementCount, tag.IsArray, tag.Type));
|
|
try
|
|
{
|
|
string? error = null;
|
|
try
|
|
{
|
|
await runtime.InitializeAsync(ct).ConfigureAwait(false);
|
|
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
|
|
catch (Exception ex)
|
|
{
|
|
error = Describe(ex);
|
|
}
|
|
|
|
var status = runtime.GetStatus();
|
|
byte[]? raw = null;
|
|
object? decoded = null;
|
|
if (error is null)
|
|
{
|
|
raw = Try(() => RawBufferOf(runtime).ToArray());
|
|
if (tag.Type is { } type)
|
|
decoded = Try(() => tag.IsArray ? runtime.DecodeArray(type, tag.ElementCount) : runtime.DecodeValue(type, null));
|
|
}
|
|
into.Reads[tag.Name] = new ReadObservation(status, error, raw, decoded);
|
|
}
|
|
finally
|
|
{
|
|
runtime.Dispose();
|
|
}
|
|
}
|
|
}
|
|
|
|
private static ReadOnlyMemory<byte> RawBufferOf(IAbCipTagRuntime runtime) => runtime switch
|
|
{
|
|
LibplctagTagRuntime l => l.RawBuffer,
|
|
NativeAbCipTagRuntime n => n.RawBuffer,
|
|
_ => throw new InvalidOperationException($"Unexpected runtime {runtime.GetType().Name}."),
|
|
};
|
|
|
|
/// <summary>
|
|
/// libplctag closes its session on a background thread some time after the last handle is
|
|
/// destroyed. Wait until the endpoint's TCP connection count is back at the pre-run baseline, so
|
|
/// the two transports are never connected to the controller at the same time.
|
|
/// </summary>
|
|
/// <returns>The connection count observed when native work may start (null when unobservable).</returns>
|
|
private static async Task<int?> WaitForLibplctagSessionToCloseAsync(
|
|
DifferentialTarget target, int? baseline, CancellationToken ct)
|
|
{
|
|
if (baseline is null)
|
|
{
|
|
await Task.Delay(FallbackSettle, ct).ConfigureAwait(false);
|
|
return null;
|
|
}
|
|
var deadline = Stopwatch.StartNew();
|
|
while (true)
|
|
{
|
|
var now = CountConnections(target);
|
|
if (now is null || now <= baseline) return now;
|
|
if (deadline.Elapsed > LibplctagDrainBudget)
|
|
throw new InvalidOperationException(
|
|
$"libplctag's session to {target.Host}:{target.Port} was still open {LibplctagDrainBudget.TotalSeconds:0}s after its " +
|
|
$"handles were disposed ({now} connection(s), baseline {baseline}); refusing to start the native phase alongside it.");
|
|
await Task.Delay(TimeSpan.FromMilliseconds(250), ct).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
/// <summary>Open TCP connections from this machine to the target, or null when the OS cannot say.</summary>
|
|
internal static int? CountConnections(DifferentialTarget target)
|
|
{
|
|
try
|
|
{
|
|
var addresses = Dns.GetHostAddresses(target.Host).Select(a => a.MapToIPv6()).ToHashSet();
|
|
return IPGlobalProperties.GetIPGlobalProperties().GetActiveTcpConnections().Count(c =>
|
|
c.RemoteEndPoint.Port == target.Port
|
|
&& addresses.Contains(c.RemoteEndPoint.Address.MapToIPv6())
|
|
&& c.State is not (TcpState.Closed or TcpState.TimeWait or TcpState.Unknown));
|
|
}
|
|
catch (Exception ex) when (ex is PlatformNotSupportedException or NetworkInformationException or NotImplementedException)
|
|
{
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/// <summary>Order-insensitive multiset difference: entries of <paramref name="a"/> missing from <paramref name="b"/>.</summary>
|
|
internal static List<T> Missing<T>(IEnumerable<T> a, IEnumerable<T> b) where T : notnull
|
|
{
|
|
var counts = b.GroupBy(x => x).ToDictionary(g => g.Key, g => g.Count());
|
|
var missing = new List<T>();
|
|
foreach (var x in a)
|
|
{
|
|
if (counts.TryGetValue(x, out var c) && c > 0) counts[x] = c - 1;
|
|
else missing.Add(x);
|
|
}
|
|
return missing;
|
|
}
|
|
|
|
/// <summary>Structural equality for decoded values: arrays element-wise, NaN equal to NaN.</summary>
|
|
internal static bool ValuesEqual(object? a, object? b) =>
|
|
a is Array || b is Array
|
|
? StructuralComparisons.StructuralEqualityComparer.Equals(a, b)
|
|
: Equals(a, b);
|
|
|
|
internal static bool ShapesEqual(AbCipUdtShape? a, AbCipUdtShape? b) =>
|
|
a is null || b is null
|
|
? a is null && b is null
|
|
: a.TypeName == b.TypeName && a.TotalSize == b.TotalSize && a.Members.SequenceEqual(b.Members);
|
|
|
|
private static T? Try<T>(Func<T> f)
|
|
{
|
|
try { return f(); }
|
|
catch (Exception) { return default; }
|
|
}
|
|
|
|
private static string Describe(Exception ex) => ex is AbTransportException ate
|
|
? $"{ate.Status}: {ate.Message}"
|
|
: $"{ex.GetType().Name}: {ex.Message}";
|
|
}
|