Joseph Doherty
09e0fe1841
N1: linkDown is set only when NO reply arrived, i.e. a timeout, a transport exception or a connect failure. A status the CNC returned proves it answered. EW_REJECT and every unlisted rc map to BadCommunicationError, so a deterministic rc on one tag used to fail every later tag of a reachable control on every poll. The comms-classed status still degrades health. The coalescer already treated a returned refusal this way; it is unchanged. Tests: a returned EW_REJECT on tag 1 leaves tags 2 and 3 read and Good (red before). A timeout still fails the rest without reading them (positive control). Minors: - m1: the _lifecycleLock doc now covers the health transitions it also guards. - m2: the NoteFixedTreeFailed summary says a refusal credits the answer. - m3: ReadAsync stamps a failed batch's exception with the generation the batch started under, and HandlePollError degrades that generation only. - m4: the fixed-tree "blip" absence check runs unconditionally under a wide budget, and asserts that both ticks landed inside it. - m5: direct unit cases for the coalescer link-down paths. A transport fault on the first range fails the second range and the per-tag tag without reading them. A connect failure costs one connect per batch. CncRelay.Up() retries the re-bind. Refs #740 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
285 lines
15 KiB
C#
285 lines
15 KiB
C#
using Shouldly;
|
||
using Xunit;
|
||
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
|
||
|
||
/// <summary>
|
||
/// #740 / R07:P9 — <see cref="FocasDriver.ReadAsync"/> walked the whole batch sequentially, so a SLOW CNC
|
||
/// (TCP-accept-then-stall, the live 31i-B case) cost its timeout per tag per poll during which no other
|
||
/// device was read. Devices are now read concurrently, each still serialised on its own socket.
|
||
/// </summary>
|
||
[Trait("Category", "Unit")]
|
||
public sealed class FocasConcurrentDeviceReadTests
|
||
{
|
||
private const string HostA = "focas://10.0.0.5:8193";
|
||
private const string HostB = "focas://10.0.0.6:8193";
|
||
|
||
/// <summary>
|
||
/// A client that records which device it dialled. Device A's read only completes once device B's read has
|
||
/// STARTED — so a sequential walk (A first) can never finish A, while a concurrent one does. Also counts
|
||
/// the peak number of reads in flight on any one client.
|
||
/// </summary>
|
||
private sealed class OrderingClient(TaskCompletionSource bStarted, int[] global) : FakeFocasClient
|
||
{
|
||
private string _host = "";
|
||
private int _inFlight;
|
||
public int PeakInFlight;
|
||
public int Reads;
|
||
|
||
public override Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct)
|
||
{
|
||
_host = address.ToString();
|
||
return base.ConnectAsync(address, timeout, ct);
|
||
}
|
||
|
||
public override async Task<(object? value, uint status)> ReadAsync(
|
||
FocasAddress address, FocasDataType type, CancellationToken ct)
|
||
{
|
||
var now = Interlocked.Increment(ref _inFlight);
|
||
InterlockedMax(ref PeakInFlight, now);
|
||
Interlocked.Increment(ref Reads);
|
||
// global[0] = reads in flight across EVERY client, global[1] = its peak.
|
||
InterlockedMax(ref global[1], Interlocked.Increment(ref global[0]));
|
||
try
|
||
{
|
||
if (_host.Contains("10.0.0.6", StringComparison.Ordinal)) bStarted.TrySetResult();
|
||
else await bStarted.Task.WaitAsync(TimeSpan.FromSeconds(2), ct).ConfigureAwait(false);
|
||
await Task.Delay(20, ct).ConfigureAwait(false);
|
||
return ((sbyte)1, FocasStatusMapper.Good);
|
||
}
|
||
finally
|
||
{
|
||
Interlocked.Decrement(ref _inFlight);
|
||
Interlocked.Decrement(ref global[0]);
|
||
}
|
||
}
|
||
|
||
private static void InterlockedMax(ref int target, int value)
|
||
{
|
||
int seen;
|
||
while ((seen = Volatile.Read(ref target)) < value
|
||
&& Interlocked.CompareExchange(ref target, value, seen) != seen) { }
|
||
}
|
||
}
|
||
|
||
/// <summary>A stalled device no longer holds every other device's read hostage — and each device stays serialised.</summary>
|
||
[Fact]
|
||
public async Task Devices_are_read_concurrently_and_each_stays_serialised()
|
||
{
|
||
var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||
var global = new int[2];
|
||
var factory = new FakeFocasClientFactory { Customise = () => new OrderingClient(bStarted, global) };
|
||
var drv = new FocasDriver(new FocasDriverOptions
|
||
{
|
||
Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
|
||
RawTags =
|
||
[
|
||
FocasRawTags.Entry(new FocasTagDefinition("raw/A1", "A", "R100", FocasDataType.Byte)),
|
||
FocasRawTags.Entry(new FocasTagDefinition("raw/A2", "A", "R101", FocasDataType.Byte)),
|
||
FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R100", FocasDataType.Byte)),
|
||
],
|
||
Probe = new FocasProbeOptions { Enabled = false },
|
||
// Off so every tag takes the per-tag path — this pins device-level concurrency, not coalescing.
|
||
PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = false },
|
||
}, "focas-p9", factory);
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
|
||
var snaps = await drv.ReadAsync(["raw/A1", "raw/A2", "raw/B1"], CancellationToken.None);
|
||
|
||
snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.Good,
|
||
"device A's read can only finish once device B's has started — a sequential walk times A out");
|
||
factory.Clients.Count.ShouldBe(2);
|
||
var clients = factory.Clients.Cast<OrderingClient>().ToList();
|
||
// Every client really served reads (2 on A, 1 on B), so "peak 1" is measured, not vacuous …
|
||
clients.Select(c => c.Reads).OrderBy(n => n).ShouldBe([1, 2]);
|
||
clients.ShouldAllBe(c => c.PeakInFlight == 1,
|
||
"one device's socket is still request→response: never two reads in flight on it");
|
||
// … and the two devices overlapped: two reads were in flight at once across the clients.
|
||
global[1].ShouldBe(2, "the devices must be read at the same time, not one after the other");
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
|
||
/// <summary>A device that accepted TCP and then stopped answering: every read hangs until the call timeout.</summary>
|
||
private sealed class StalledClient : FakeFocasClient
|
||
{
|
||
public int Reads;
|
||
|
||
public override async Task<(object? value, uint status)> ReadAsync(
|
||
FocasAddress address, FocasDataType type, CancellationToken ct)
|
||
{
|
||
Interlocked.Increment(ref Reads);
|
||
await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
|
||
return (null, FocasStatusMapper.Good);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// #740 review I1 — <c>Task.WhenAll</c> waits for the slowest device, so concurrency alone still made a
|
||
/// stalled device cost (its tag count × the call timeout) per batch, and the healthy device's values waited
|
||
/// that long. After the first timeout on a device its remaining tags in the batch fail fast: the batch is
|
||
/// bounded by about ONE call timeout and the healthy device's values are Good.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_stalled_device_costs_one_call_timeout_per_batch_not_one_per_tag()
|
||
{
|
||
var callTimeout = TimeSpan.FromMilliseconds(200);
|
||
var factory = new FakeFocasClientFactory();
|
||
factory.Customise = () => factory.Clients.Count == 0 ? new StalledClient() : new FakeFocasClient { Values = { ["R100"] = (sbyte)9 } };
|
||
var stalledTags = Enumerable.Range(0, 10)
|
||
.Select(i => FocasRawTags.Entry(new FocasTagDefinition($"raw/A{i}", "A", $"R{100 + i}", FocasDataType.Byte)));
|
||
var drv = new FocasDriver(new FocasDriverOptions
|
||
{
|
||
Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
|
||
RawTags = [.. stalledTags, FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R100", FocasDataType.Byte))],
|
||
Probe = new FocasProbeOptions { Enabled = false },
|
||
PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = false },
|
||
Timeout = callTimeout,
|
||
}, "focas-stall", factory);
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
// Connect device A first so it gets the stalled client (connect order = client order).
|
||
await drv.ReadAsync(["raw/A0"], CancellationToken.None);
|
||
|
||
var refs = Enumerable.Range(0, 10).Select(i => $"raw/A{i}").Append("raw/B1").ToArray();
|
||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||
var snaps = await drv.ReadAsync(refs, CancellationToken.None);
|
||
sw.Stop();
|
||
|
||
sw.Elapsed.ShouldBeLessThan(callTimeout * 3, "one stalled device must cost ~one call timeout, not ten");
|
||
snaps[10].StatusCode.ShouldBe(FocasStatusMapper.Good);
|
||
snaps[10].Value.ShouldBe((sbyte)9);
|
||
snaps.Take(10).ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
|
||
((StalledClient)factory.Clients[0]).Reads.ShouldBe(2, "one read in the warm-up batch, one in the measured batch");
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
|
||
/// <summary>Counts per-tag reads; one address can be made to time out (throw) or answer a status.</summary>
|
||
private sealed class CountingClient : FakeFocasClient
|
||
{
|
||
public int Reads;
|
||
public string? TimeoutAt;
|
||
|
||
public override async Task<(object? value, uint status)> ReadAsync(
|
||
FocasAddress address, FocasDataType type, CancellationToken ct)
|
||
{
|
||
Interlocked.Increment(ref Reads);
|
||
if (address.Canonical == TimeoutAt)
|
||
{
|
||
await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
|
||
}
|
||
return await base.ReadAsync(address, type, ct).ConfigureAwait(false);
|
||
}
|
||
}
|
||
|
||
private static FocasDriver OneDevice(FakeFocasClientFactory factory, bool coalesce, params string[] addresses) => new(
|
||
new FocasDriverOptions
|
||
{
|
||
Devices = [new FocasDeviceOptions(HostA, DeviceName: "A")],
|
||
RawTags = [.. addresses.Select(a => FocasRawTags.Entry(new FocasTagDefinition($"raw/{a}", "A", a, FocasDataType.Byte)))],
|
||
Probe = new FocasProbeOptions { Enabled = false },
|
||
PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = coalesce },
|
||
Timeout = TimeSpan.FromMilliseconds(200),
|
||
}, "focas-linkdown", factory);
|
||
|
||
/// <summary>
|
||
/// #740 review round 2 N1 — a status the CNC RETURNED (EW_REJECT and every unlisted rc map to
|
||
/// BadCommunicationError) proves the device answered, so it must not end the device's batch: the tags after it
|
||
/// still read. Before the fix, a deterministic rc on the first tag starved every later tag forever.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task A_returned_comms_status_does_not_fail_the_rest_of_the_batch()
|
||
{
|
||
var client = new CountingClient { Values = { ["R101"] = (sbyte)1, ["R102"] = (sbyte)2 } };
|
||
client.ReadStatuses["R100"] = FocasStatusMapper.MapFocasReturn(13); // EW_REJECT ⇒ BadCommunicationError
|
||
var drv = OneDevice(new FakeFocasClientFactory { Customise = () => client }, coalesce: false, "R100", "R101", "R102");
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
|
||
var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/R102"], CancellationToken.None);
|
||
|
||
snaps[0].StatusCode.ShouldBe(FocasStatusMapper.BadCommunicationError);
|
||
snaps[1].StatusCode.ShouldBe(FocasStatusMapper.Good);
|
||
snaps[2].Value.ShouldBe((sbyte)2);
|
||
client.Reads.ShouldBe(3, "every tag was really read");
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
|
||
/// <summary>Positive control for N1: a TIMEOUT (no reply at all) still fails the rest of the batch without reading it.</summary>
|
||
[Fact]
|
||
public async Task A_timeout_still_fails_the_rest_of_the_batch_fast()
|
||
{
|
||
var client = new CountingClient { TimeoutAt = "R100", Values = { ["R101"] = (sbyte)1 } };
|
||
var drv = OneDevice(new FakeFocasClientFactory { Customise = () => client }, coalesce: false, "R100", "R101", "R102");
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
|
||
var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/R102"], CancellationToken.None);
|
||
|
||
snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
|
||
client.Reads.ShouldBe(1, "only the tag that timed out touched the wire");
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Review round 2 m5 — the coalescer's link-down paths: a transport fault on the first merged range fails the
|
||
/// second range WITHOUT a second range read, and the per-tag tags of the same device without a per-tag read.
|
||
/// </summary>
|
||
[Fact]
|
||
public async Task Coalescer_transport_fault_fails_later_ranges_and_per_tag_tags_without_reading_them()
|
||
{
|
||
var client = new CountingClient { ThrowOnPmcRange = true };
|
||
// Two ranges far apart (gap > MaxGapBytes) + one lone tag that takes the per-tag path.
|
||
var drv = OneDevice(new FakeFocasClientFactory { Customise = () => client }, coalesce: true,
|
||
"R100", "R101", "R500", "R501", "D900");
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
|
||
var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/R500", "raw/R501", "raw/D900"], CancellationToken.None);
|
||
|
||
snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
|
||
client.PmcRangeReads.Count.ShouldBe(1, "the second range must not pay another transport attempt");
|
||
client.Reads.ShouldBe(0, "the per-tag tag must not pay one either");
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
|
||
/// <summary>Review round 2 m5 — a coalescer CONNECT failure fails the per-tag tags of the device without a second connect.</summary>
|
||
[Fact]
|
||
public async Task Coalescer_connect_failure_fails_per_tag_tags_without_reconnecting()
|
||
{
|
||
var factory = new FakeFocasClientFactory { Customise = () => new CountingClient { ThrowOnConnect = true } };
|
||
var drv = OneDevice(factory, coalesce: true, "R100", "R101", "D900");
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
|
||
var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/D900"], CancellationToken.None);
|
||
|
||
snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
|
||
factory.Clients.Count.ShouldBe(1, "one connect attempt per device per batch");
|
||
factory.Clients.Sum(c => c.ConnectCount).ShouldBe(1);
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
|
||
/// <summary>Results land at their own indices whatever order the devices finish in.</summary>
|
||
[Fact]
|
||
public async Task Results_keep_their_request_order_across_devices()
|
||
{
|
||
var factory = new FakeFocasClientFactory();
|
||
var drv = new FocasDriver(new FocasDriverOptions
|
||
{
|
||
Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
|
||
RawTags =
|
||
[
|
||
FocasRawTags.Entry(new FocasTagDefinition("raw/A1", "A", "R100", FocasDataType.Byte)),
|
||
FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R200", FocasDataType.Byte)),
|
||
],
|
||
Probe = new FocasProbeOptions { Enabled = false },
|
||
}, "focas-p9-order", factory)
|
||
;
|
||
factory.Customise = () => new FakeFocasClient { Values = { ["R100"] = (sbyte)1, ["R200"] = (sbyte)2 } };
|
||
await drv.InitializeAsync("{}", CancellationToken.None);
|
||
|
||
var snaps = await drv.ReadAsync(["raw/B1", "nope", "raw/A1"], CancellationToken.None);
|
||
|
||
snaps[0].Value.ShouldBe((sbyte)2);
|
||
snaps[1].StatusCode.ShouldBe(FocasStatusMapper.BadNodeIdUnknown);
|
||
snaps[2].Value.ShouldBe((sbyte)1);
|
||
await drv.ShutdownAsync(CancellationToken.None);
|
||
}
|
||
}
|