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

fix(ablegacy): isolate a dead device — parallel devices, per-device backoff, single-flight connect

Browse Source 
Review 07 P7 (High): reads walked every reference of every device serially
with no backoff, so a dead device cost Timeout per tag per poll (50 tags x 2 s
= 100 s) and starved every healthy device long enough to trip the comms-loss
watchdog against a reachable one.

- ReadAsync buckets references by device and reads up to 8 devices
  concurrently; each device's tags are still read in order.
- Each device has a ConnectionBackoff (1 s doubling to 30 s). Two consecutive
  comms faults mark it down: the rest of that pass and every later pass fail
  fast with BadCommunicationError and touch no wire until the window expires or
  any answer (data path or probe) resets it. One timing-out tag does not
  blacken its answering siblings.
- Cold connects are single-flight per device (ConnectGate), replacing the
  per-key creation locks; a cached handle never touches the gate.
- A status/exception proving the device ANSWERED with a deterministic refusal
  (not found, not allowed, out of bounds, type/size mismatch) is per-tag Bad,
  keeps its handle, warns once per tag and advances LastSuccessfulRead instead
  of evicting and degrading. Writes classify the same way.
- Health is written once per read: Degraded if any device faulted, Healthy if
  anything read Good, LastSuccessfulRead advanced when any device answered.

Adds an env-gated ab_server case with a blackholed sibling device.

Refs #738
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:33:54 -04:00
parent 517837e798
commit e5527d0201
3 changed files with 738 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
+414 -160
View File
@@ -423,7 +423,50 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
// ---- IReadable ----
/// <summary>
/// How many devices one <see cref="ReadAsync"/> reads concurrently. Each device is still read serially (its
/// tags share one PCCC connection, and SLC/MicroLogix controllers accept few), so this bounds the number of
/// devices — not requests — in flight, which is what keeps a dead device from starving a healthy one.
/// </summary>
internal const int MaxConcurrentDevices = 8;
/// <summary>
/// Consecutive comms faults within one device that mark it DOWN (opening its backoff window and failing the
/// rest of the pass fast). Two, not one: a single tag that times out (an odd address, a slow file) must not
/// blacken every sibling on a device that is answering.
/// </summary>
internal const int FaultsToMarkDeviceDown = 2;
/// <summary>One resolved reference queued for its device's pass.</summary>
private sealed record ReadItem(int Index, string Reference, AbLegacyTagDefinition Def, AbLegacyAddress Parsed);
/// <summary>
/// What one wire operation proved. <see cref="Answered"/> = the device replied (a Good value OR a
/// deterministic device refusal such as "file not found") — the evidence that advances
/// <c>LastSuccessfulRead</c>; <see cref="Fault"/> = a comms failure (timeout, dropped connection, garbage
/// reply); <see cref="Throttled"/> = failed fast inside the device's backoff window (Bad, Degraded, but no new
/// evidence for the backoff); <see cref="Neutral"/> = nothing reached the wire or it says nothing about
/// reachability (a local refusal, a bad index value).
/// </summary>
private enum OpVerdict { Neutral, Answered, Fault, Throttled }
/// <summary>The health evidence one device pass produced (collected, then written once per read).</summary>
private struct DeviceReadOutcome
{
public bool AnyGood;
public bool AnyAnswered;
public string? FaultMessage;
}
/// <inheritdoc />
/// <remarks>
/// Review 07 P7: references are bucketed by device and the devices are read concurrently (at most
/// <see cref="MaxConcurrentDevices"/>); within a device the tags are read in order. A device that produces
/// <see cref="FaultsToMarkDeviceDown"/> consecutive comms faults opens its <see cref="ConnectionBackoff"/>
/// window and every remaining (and every later) read on it fails fast with <c>BadCommunicationError</c>,
/// touching no wire, until the window expires or the probe sees it answer. Health is written ONCE per call:
/// worst outcome wins, and <c>LastSuccessfulRead</c> advances when ANY device answered.
/// </remarks>
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
@@ -431,124 +474,255 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
var epoch = CurrentEpoch;
var now = DateTime.UtcNow;
var results = new DataValueSnapshot[fullReferences.Count];
var devices = _devices; // one snapshot of the published table for the whole call
var buckets = new Dictionary<DeviceState, List<ReadItem>>(ReferenceEqualityComparer.Instance);
for (var i = 0; i < fullReferences.Count; i++)
{
var reference = fullReferences[i];
if (!_resolver.TryResolve(reference, out var def))
if (!_resolver.TryResolve(reference, out var def) || !devices.TryGetValue(def.DeviceHostAddress, out var device))
{
results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNodeIdUnknown, null, now);
continue;
}
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device))
var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily);
if (parsed is null)
{
WarnMalformedOnce(device, def);
results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNodeIdUnknown, null, now);
continue;
}
if (!buckets.TryGetValue(device, out var list)) buckets[device] = list = [];
list.Add(new ReadItem(i, reference, def, parsed));
}
if (buckets.Count == 0) return results;
try
var passes = buckets.ToArray();
var outcomes = new DeviceReadOutcome[passes.Length];
using (var window = new SemaphoreSlim(MaxConcurrentDevices, MaxConcurrentDevices))
{
var tasks = new Task[passes.Length];
for (var d = 0; d < passes.Length; d++)
tasks[d] = RunPassAsync(d);
// WhenAll completes only once every pass has finished (it never short-circuits on a fault), so the
// window cannot be disposed under a pass still waiting on it.
await Task.WhenAll(tasks).ConfigureAwait(false);
async Task RunPassAsync(int d)
{
var preParsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily);
if (preParsed is null)
{
WarnMalformedOnce(device, def);
results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNodeIdUnknown, null, now);
continue;
}
if (preParsed.IsIndirect)
{
// Indirect: read the index word(s) NOW (never cached - the index can change between
// polls), compose the concrete address, then run the normal path against it.
var res = await ResolveIndirectAsync(device, def, preParsed, cancellationToken).ConfigureAwait(false);
if (res.Def is null)
{
results[i] = new DataValueSnapshot(null, res.Status, null, now);
if (res.TransportFault)
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport fault resolving indirect address for {reference}"));
continue;
}
def = res.Def;
}
var runtime = await EnsureTagRuntimeAsync(device, def, cancellationToken).ConfigureAwait(false);
int status;
object? value;
// Serialise the Read → GetStatus → DecodeValue sequence against the shared
// runtime — the server read path and the poll loop both call ReadAsync against
// the same cached libplctag Tag handle, which is not concurrency-safe.
var opLock = device.GetRuntimeLock(def.Name);
await opLock.WaitAsync(cancellationToken).ConfigureAwait(false);
await window.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await runtime.ReadAsync(cancellationToken).ConfigureAwait(false);
status = runtime.GetStatus();
var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily);
if (status != 0)
value = null;
else if (IsArrayTag(def))
value = runtime.DecodeArray(def.DataType, EffectiveArrayLength(def));
else
value = runtime.DecodeValue(def.DataType, parsed?.BitIndex);
outcomes[d] = await ReadDevicePassAsync(passes[d].Key, passes[d].Value, results, now, cancellationToken)
.ConfigureAwait(false);
}
finally
{
opLock.Release();
window.Release();
}
if (status != 0)
{
// Evict the stale handle so the next call re-creates it (mirrors AbCip).
EvictRuntime(device, def.Name);
results[i] = new DataValueSnapshot(null,
AbLegacyStatusMapper.MapStatus(status), null, now);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport status {status} reading {reference}"));
if (!device.FirstNonZeroStatusLogged)
{
device.FirstNonZeroStatusLogged = true;
_logger.LogWarning(
"AbLegacy non-zero transport status. Driver={DriverInstanceId} Device={DeviceHostAddress} Reference={Reference} Status={Status}",
_driverInstanceId, def.DeviceHostAddress, reference, status);
}
continue;
}
// Healthy read — re-arm the per-device first-failure log so a future non-zero
// status logs again rather than being suppressed by an old flag from a prior outage.
device.FirstNonZeroStatusLogged = false;
results[i] = new DataValueSnapshot(value, AbLegacyStatusMapper.Good, now, now);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Healthy, now, null));
}
catch (OperationCanceledException) { throw; }
catch (AbTransportException ex) when (IsLocalRefusal(ex))
{
// The runtime refused the operation before the wire (e.g. a native PLC-5 non-N read).
// Not a comms fault: BadNotSupported, keep the handle, leave health alone.
WarnRefusalOnce(device, def, "read", ex.Message);
results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now);
}
catch (NotSupportedException nse)
{
// A deterministic decode refusal (review 07 H16) — the same answer every poll. BadNotSupported, keep
// the handle (re-creating it only re-paid the init read), warn once, and NO health write.
WarnRefusalOnce(device, def, "read", nse.Message);
results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now);
}
catch (Exception ex)
{
// Transport exception — evict so the next read creates a fresh handle (mirrors AbCip).
EvictRuntime(device, def.Name);
results[i] = new DataValueSnapshot(null,
AbLegacyStatusMapper.BadCommunicationError, null, now);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message));
}
}
ApplyReadHealth(epoch, outcomes, now);
return results;
}
/// <summary>
/// Writes the read's health once, after every device pass: Degraded if any pass hit a comms fault, else
/// Healthy if anything read Good; <c>LastSuccessfulRead</c> advances when any device answered (a Good read or a
/// deterministic device refusal — both prove the device is reachable). A read that reached no device's
/// wire, or only drew refusals, does not change the state.
/// </summary>
private void ApplyReadHealth(long epoch, DeviceReadOutcome[] outcomes, DateTime now)
{
var anyGood = false;
var anyAnswered = false;
string? fault = null;
foreach (var o in outcomes)
{
anyGood |= o.AnyGood;
anyAnswered |= o.AnyAnswered;
fault ??= o.FaultMessage;
}
if (fault is null && !anyGood && !anyAnswered) return;
lock (_lifecycleLock)
{
if (epoch != _lifecycleEpoch) return;
var current = _health;
var lastRead = anyAnswered ? now : current.LastSuccessfulRead;
if (fault is not null)
_health = new DriverHealth(DriverState.Degraded, lastRead, fault);
else if (anyGood)
_health = new DriverHealth(DriverState.Healthy, lastRead, null);
else
_health = current with { LastSuccessfulRead = lastRead };
}
}
/// <summary>Reads one device's references in order, failing fast once the device is marked down.</summary>
private async Task<DeviceReadOutcome> ReadDevicePassAsync(
DeviceState device, List<ReadItem> items, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
{
var outcome = new DeviceReadOutcome();
foreach (var item in items)
{
if (device.IsThrottled(DateTime.UtcNow))
{
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now);
outcome.FaultMessage ??= $"device {device.Options.HostAddress} is in its reconnect backoff window";
continue;
}
var (verdict, message) = await ReadOneAsync(device, item, results, now, ct).ConfigureAwait(false);
switch (verdict)
{
case OpVerdict.Answered:
outcome.AnyAnswered = true;
if (results[item.Index].StatusCode == AbLegacyStatusMapper.Good) outcome.AnyGood = true;
device.RecordAnswered();
break;
case OpVerdict.Fault:
outcome.FaultMessage ??= message;
device.RecordCommsFault(DateTime.UtcNow);
break;
case OpVerdict.Throttled:
outcome.FaultMessage ??= message;
break;
}
}
return outcome;
}
/// <summary>Reads one reference; writes its snapshot and reports what the operation proved.</summary>
private async Task<(OpVerdict Verdict, string? Message)> ReadOneAsync(
DeviceState device, ReadItem item, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
{
var def = item.Def;
var reference = item.Reference;
try
{
if (item.Parsed.IsIndirect)
{
// Indirect: read the index word(s) NOW (never cached - the index can change between
// polls), compose the concrete address, then run the normal path against it.
var res = await ResolveIndirectAsync(device, def, item.Parsed, ct).ConfigureAwait(false);
if (res.Def is null)
{
results[item.Index] = new DataValueSnapshot(null, res.Status, null, now);
return res.TransportFault
? (OpVerdict.Fault, $"transport fault resolving indirect address for {reference}")
: (OpVerdict.Neutral, null);
}
def = res.Def;
}
var runtime = await EnsureTagRuntimeAsync(device, def, ct).ConfigureAwait(false);
int status;
object? value;
// Serialise the Read → GetStatus → DecodeValue sequence against the shared
// runtime — the server read path and the poll loop both call ReadAsync against
// the same cached libplctag Tag handle, which is not concurrency-safe.
var opLock = device.GetRuntimeLock(def.Name);
await opLock.WaitAsync(ct).ConfigureAwait(false);
try
{
await runtime.ReadAsync(ct).ConfigureAwait(false);
status = runtime.GetStatus();
var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily);
if (status != 0)
value = null;
else if (IsArrayTag(def))
value = runtime.DecodeArray(def.DataType, EffectiveArrayLength(def));
else
value = runtime.DecodeValue(def.DataType, parsed?.BitIndex);
}
finally
{
opLock.Release();
}
if (status != 0)
{
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.MapStatus(status), null, now);
if (IsDeviceRefusal((AbStatus)status))
{
// The device ANSWERED with a deterministic refusal (file/element missing, out of bounds…): per-tag
// Bad, the handle is kept (re-creating it would only re-pay the init read), warned once per tag,
// and it is evidence of reachability rather than a comms fault.
WarnRefusalOnce(device, def, "read", $"device status {(AbStatus)status}");
return (OpVerdict.Answered, null);
}
// Evict the stale handle so the next call re-creates it (mirrors AbCip).
EvictRuntime(device, def.Name);
if (!device.FirstNonZeroStatusLogged)
{
device.FirstNonZeroStatusLogged = true;
_logger.LogWarning(
"AbLegacy non-zero transport status. Driver={DriverInstanceId} Device={DeviceHostAddress} Reference={Reference} Status={Status}",
_driverInstanceId, def.DeviceHostAddress, reference, status);
}
return (OpVerdict.Fault, $"transport status {status} reading {reference}");
}
// Healthy read — re-arm the per-device first-failure log so a future non-zero
// status logs again rather than being suppressed by an old flag from a prior outage.
device.FirstNonZeroStatusLogged = false;
results[item.Index] = new DataValueSnapshot(value, AbLegacyStatusMapper.Good, now, now);
return (OpVerdict.Answered, null);
}
catch (OperationCanceledException) { throw; }
catch (ConnectionThrottledException tex)
{
// A cold connect refused inside the backoff window: no wire, but the tag has no value either.
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now);
return (OpVerdict.Throttled, tex.Message);
}
catch (AbTransportException ex) when (IsLocalRefusal(ex))
{
// The runtime refused the operation before the wire (e.g. a native PLC-5 non-N read).
// Not a comms fault: BadNotSupported, keep the handle, leave health alone.
WarnRefusalOnce(device, def, "read", ex.Message);
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now);
return (OpVerdict.Neutral, null);
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
{
// The native runtime raises a device's refusal (e.g. STS 0x20 "file not found") as an exception.
WarnRefusalOnce(device, def, "read", ex.Message);
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.MapStatus(ex.Status), null, now);
return (OpVerdict.Answered, null);
}
catch (NotSupportedException nse)
{
// A deterministic decode refusal (review 07 H16) — the same answer every poll. BadNotSupported, keep
// the handle (re-creating it only re-paid the init read), warn once, and NO health write.
WarnRefusalOnce(device, def, "read", nse.Message);
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now);
return (OpVerdict.Neutral, null);
}
catch (Exception ex)
{
// Transport exception — evict so the next read creates a fresh handle (mirrors AbCip).
EvictRuntime(device, def.Name);
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now);
return (OpVerdict.Fault, ex.Message);
}
}
/// <summary>
/// True for a status that proves the DEVICE answered and refused this one operation deterministically — the
/// file or element does not exist, the access is not allowed, the data type or size is wrong. Such a status
/// is per-tag, keeps the handle, and is not a comms fault. Everything else non-zero (timeouts, dropped or
/// refused connections, malformed or partial replies, unknown codes) is treated as a comms fault.
/// <see cref="AbStatus.Unsupported"/> is classified separately (<see cref="IsLocalRefusal"/>).
/// </summary>
/// <param name="status">The transport status.</param>
/// <returns><see langword="true"/> for a deterministic device refusal.</returns>
internal static bool IsDeviceRefusal(AbStatus status) => status is
AbStatus.NotFound or AbStatus.NotAllowed or AbStatus.OutOfBounds or AbStatus.TooLarge or AbStatus.TooSmall
or AbStatus.TypeMismatch or AbStatus.DeviceStateConflict;
// ---- IWritable ----
/// <inheritdoc />
@@ -558,6 +732,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
ArgumentNullException.ThrowIfNull(writes);
var epoch = CurrentEpoch;
var results = new WriteResult[writes.Count];
var devices = _devices; // one snapshot of the published table for the whole call
for (var i = 0; i < writes.Count; i++)
{
@@ -572,7 +747,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
results[i] = new WriteResult(AbLegacyStatusMapper.BadNotWritable);
continue;
}
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device))
if (!devices.TryGetValue(def.DeviceHostAddress, out var device))
{
results[i] = new WriteResult(AbLegacyStatusMapper.BadNodeIdUnknown);
continue;
@@ -619,14 +794,6 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
continue;
}
// PCCC bit-within-word writes — RMW against a parallel parent-word runtime (strip the /N
// bit suffix). The per-parent-word lock serialises concurrent bit writers. Applies to every
// bit-addressable PCCC file: N-file (N7:0/3), B-file (B3:0/0), and the I/O image files
// (I:0/0, O:1/2); L-file bits RMW a 32-bit parent, the rest a 16-bit word. T/C/R sub-elements
// (T4:0.DN, even when authored Bit-typed) never reach this path: AbLegacyAddress.TryParse rejects a
// sub-element combined with a /N bit index, so their BitIndex is always null. NOTE: an Input-image (I) write is sent
// to the PLC like any other write — the device drives the input image from physical inputs and
// may reject it; we surface that real PCCC status rather than pre-rejecting at the driver.
// Array writes (review 07 U10): convert and length-check BEFORE any runtime exists, then encode every
// element. A bit / sub-element array has no flat encoding.
Array? arrayValue = null;
@@ -643,6 +810,14 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
continue;
}
}
// PCCC bit-within-word writes — RMW against a parallel parent-word runtime (strip the /N
// bit suffix). The per-parent-word lock serialises concurrent bit writers. Applies to every
// bit-addressable PCCC file: N-file (N7:0/3), B-file (B3:0/0), and the I/O image files
// (I:0/0, O:1/2); L-file bits RMW a 32-bit parent, the rest a 16-bit word. T/C/R sub-elements
// (T4:0.DN, even when authored Bit-typed) never reach this path: AbLegacyAddress.TryParse rejects a
// sub-element combined with a /N bit index, so their BitIndex is always null. NOTE: an Input-image (I) write is sent
// to the PLC like any other write — the device drives the input image from physical inputs and
// may reject it; we surface that real PCCC status rather than pre-rejecting at the driver.
else if (def.DataType == AbLegacyDataType.Bit && parsed.BitIndex is int bit)
{
results[i] = new WriteResult(
@@ -669,18 +844,38 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
opLock.Release();
}
if (status != 0)
if (status == 0)
{
// Evict the stale handle so the next call re-creates it (mirrors AbCip).
EvictRuntime(device, def.Name);
device.RecordAnswered();
results[i] = new WriteResult(AbLegacyStatusMapper.Good);
}
else if (IsDeviceRefusal((AbStatus)status))
{
// The device answered and refused this write deterministically: keep the handle, warn once.
device.RecordAnswered();
WarnRefusalOnce(device, def, "write", $"device status {(AbStatus)status}");
results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(status));
}
else
{
results[i] = new WriteResult(AbLegacyStatusMapper.Good);
// Evict the stale handle so the next call re-creates it (mirrors AbCip).
EvictRuntime(device, def.Name);
device.RecordCommsFault(DateTime.UtcNow);
results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(status));
}
}
catch (OperationCanceledException) { throw; }
catch (ConnectionThrottledException)
{
// The device is in its reconnect backoff window: fail fast, no wire, no new evidence.
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
{
device.RecordAnswered();
WarnRefusalOnce(device, def, "write", ex.Message);
results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(ex.Status));
}
catch (NotSupportedException nse)
{
// Deterministic (review 07 H16): nothing reached the wire, so it says nothing about the device.
@@ -706,6 +901,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
{
// Transport exception — evict so the next write/read creates a fresh handle (mirrors AbCip).
EvictRuntime(device, def.Name);
device.RecordCommsFault(DateTime.UtcNow);
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message));
}
@@ -799,14 +995,21 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
state.ProbeInitialized = true;
}
await probeRuntime.ReadAsync(ct).ConfigureAwait(false);
success = probeRuntime.GetStatus() == 0;
var status = probeRuntime.GetStatus();
success = status == 0;
// The probe's verdict feeds the data path's backoff (review 07 P7): an answer resets the window so a
// recovered device is read again at once; a comms fault counts toward marking it down.
if (success || IsDeviceRefusal((AbStatus)status)) state.RecordAnswered();
else state.RecordCommsFault(DateTime.UtcNow);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
catch
catch (Exception ex)
{
try { probeRuntime?.Dispose(); } catch { }
probeRuntime = null;
state.ProbeInitialized = false;
if (ex is AbTransportException ate && IsDeviceRefusal(ate.Status)) state.RecordAnswered();
else if (!IsLocalRefusal(ex)) state.RecordCommsFault(DateTime.UtcNow);
}
TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
@@ -1043,10 +1246,16 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
await runtime.ReadAsync(ct).ConfigureAwait(false);
}
catch (OperationCanceledException) { throw; }
catch (ConnectionThrottledException) { throw; } // the caller answers it as a fail-fast
catch (AbTransportException ex) when (IsLocalRefusal(ex))
{
return (null, AbLegacyStatusMapper.BadNotSupported, false);
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
{
// The device answered: the index file/element does not exist. Per-tag, not a comms fault.
return (null, AbLegacyStatusMapper.MapStatus(ex.Status), false);
}
catch
{
EvictParentRuntime(device, runtimeName);
@@ -1054,6 +1263,8 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
}
var status = runtime.GetStatus();
if (status != 0 && IsDeviceRefusal((AbStatus)status))
return (null, AbLegacyStatusMapper.MapStatus(status), false);
if (status != 0)
{
EvictParentRuntime(device, runtimeName);
@@ -1093,32 +1304,56 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
private async Task<IAbLegacyTagRuntime> EnsureParentRuntimeAsync(
AbLegacyDriver.DeviceState device, string parentName, CancellationToken ct)
{
// Fast path: runtime already cached.
if (device.ParentRuntimes.TryGetValue(parentName, out var existing)) return existing;
// Slow path: serialise creation per key so concurrent callers don't each create a
// runtime and one of them gets overwritten + leaked. Only one caller initialises; the
// others find the entry on the second TryGetValue inside the lock.
var creationLock = device.GetCreationLock($"parent:{parentName}");
await creationLock.WaitAsync(ct).ConfigureAwait(false);
try
{
if (device.ParentRuntimes.TryGetValue(parentName, out existing)) return existing;
var runtime = _tagFactory.Create(new AbLegacyTagCreateParams(
var timeout = _options.Timeout;
return await CreateSingleFlightAsync(device, device.ParentRuntimes, parentName, () => _tagFactory.Create(
new AbLegacyTagCreateParams(
Gateway: device.ParsedAddress.Gateway,
Port: device.ParsedAddress.Port,
CipPath: device.EffectiveCipPath,
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
TagName: parentName,
Timeout: _options.Timeout));
await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(device.ParentRuntimes, parentName, runtime), ct)
.ConfigureAwait(false);
Timeout: timeout)), ct).ConfigureAwait(false);
}
/// <summary>
/// Thrown when a cold connect is refused because the device's <see cref="ConnectionBackoff"/> window is open
/// — a fail-fast that costs no wire traffic. Reads and writes answer it with <c>BadCommunicationError</c>.
/// </summary>
internal sealed class ConnectionThrottledException(string message) : Exception(message);
/// <summary>
/// Returns the cached runtime for <paramref name="key"/>, or creates and initialises one. The COLD path is
/// single-flight per DEVICE (<see cref="DeviceState.ConnectGate"/>, review 07 P7): concurrent callers for
/// different cold tags of one device connect one at a time, so a down device sees one connect attempt instead
/// of one per caller, and inside an open backoff window no attempt is made at all
/// (<see cref="ConnectionThrottledException"/>). The HOT path (cached handle) never touches the gate. The
/// caller records the outcome on the device's backoff (once — recording here as well would double-count).
/// </summary>
private static async Task<IAbLegacyTagRuntime> CreateSingleFlightAsync(
DeviceState device,
System.Collections.Concurrent.ConcurrentDictionary<string, IAbLegacyTagRuntime> table,
string key,
Func<IAbLegacyTagRuntime> create,
CancellationToken ct)
{
if (table.TryGetValue(key, out var existing)) return existing;
await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
try
{
// Re-check under the gate: a sibling caller may have connected this very key while we waited.
if (table.TryGetValue(key, out existing)) return existing;
if (device.IsThrottled(DateTime.UtcNow))
throw new ConnectionThrottledException(
$"AbLegacy connect for '{key}' on {device.Options.HostAddress} throttled: the device is in its reconnect backoff window.");
var runtime = create();
await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(table, key, runtime), ct).ConfigureAwait(false);
return runtime;
}
finally
{
creationLock.Release();
device.ConnectGate.Release();
}
}
@@ -1214,30 +1449,14 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
private async Task<IAbLegacyTagRuntime> EnsureTagRuntimeAsync(
DeviceState device, AbLegacyTagDefinition def, CancellationToken ct)
{
// Fast path: runtime already cached.
if (device.Runtimes.TryGetValue(def.Name, out var existing)) return existing;
// Slow path: serialise creation per tag name so concurrent callers for the same tag
// (server read path + poll loop) don't both create a runtime and one gets leaked.
var creationLock = device.GetCreationLock($"tag:{def.Name}");
await creationLock.WaitAsync(ct).ConfigureAwait(false);
try
{
if (device.Runtimes.TryGetValue(def.Name, out existing)) return existing;
var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily)
?? throw new InvalidOperationException(
$"AbLegacy tag '{def.Name}' has malformed Address '{def.Address}' for PLC family {device.Options.PlcFamily}.");
var runtime = _tagFactory.Create(CreateParamsFor(device, def, parsed));
await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(device.Runtimes, def.Name, runtime), ct)
.ConfigureAwait(false);
return runtime;
}
finally
{
creationLock.Release();
}
var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily)
?? throw new InvalidOperationException(
$"AbLegacy tag '{def.Name}' has malformed Address '{def.Address}' for PLC family {device.Options.PlcFamily}.");
var createParams = CreateParamsFor(device, def, parsed);
return await CreateSingleFlightAsync(device, device.Runtimes, def.Name, () => _tagFactory.Create(createParams), ct)
.ConfigureAwait(false);
}
/// <summary>
@@ -1329,7 +1548,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// avoids the check-then-act race present on a plain <c>Dictionary</c>: two concurrent
/// <c>EnsureTagRuntimeAsync</c> callers for the same key both miss the lookup on a
/// plain dict and both create + store a runtime, leaking the loser. Access is guarded
/// by a per-key creation semaphore (<see cref="GetCreationLock"/>) so exactly one
/// by the per-device connect gate (<see cref="ConnectGate"/>) so exactly one
/// runtime is created per tag name.
/// </summary>
public System.Collections.Concurrent.ConcurrentDictionary<string, IAbLegacyTagRuntime> Runtimes { get; } =
@@ -1348,20 +1567,56 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
new(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Per-key creation locks for <see cref="Runtimes"/> and <see cref="ParentRuntimes"/>.
/// A caller holds this before the TryGetValue + Create + InitializeAsync + TryAdd
/// sequence so that a concurrent caller waits rather than creating a duplicate runtime
/// that would be leaked on <see cref="Retire"/>.
/// Per-device connect gate (review 07 P7): every COLD runtime create + initialise on this device runs under
/// it, so connects are single-flight per device — a down device sees one attempt at a time, not one per
/// concurrent caller — and a concurrent caller for the same key finds the winner's runtime instead of
/// creating (and leaking) a duplicate. A cached runtime never touches it. Not disposed on retirement, for
/// the same reason as the per-key gates (an in-flight caller may still release it).
/// </summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, SemaphoreSlim> _creationLocks = new(StringComparer.OrdinalIgnoreCase);
public SemaphoreSlim ConnectGate { get; } = new(1, 1);
/// <summary>
/// Gets or creates the creation lock for the specified key.
/// Per-device reconnect throttle: opened after <see cref="FaultsToMarkDeviceDown"/> consecutive comms faults,
/// 1 s doubling to 30 s, reset by any answer from the device (data path or probe). Not internally
/// synchronised — every touch is under <see cref="_backoffLock"/>.
/// </summary>
/// <param name="key">The lock key.</param>
/// <returns>The semaphore slim for the key.</returns>
public SemaphoreSlim GetCreationLock(string key) =>
_creationLocks.GetOrAdd(key, _ => new SemaphoreSlim(1, 1));
private readonly ConnectionBackoff _backoff = new(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30));
private readonly object _backoffLock = new();
private int _faultStreak;
/// <summary>True while the device's backoff window is open: operations fail fast without the wire.</summary>
/// <param name="nowUtc">The current UTC time.</param>
/// <returns><see langword="true"/> when no attempt should be made now.</returns>
public bool IsThrottled(DateTime nowUtc)
{
lock (_backoffLock) return !_backoff.ShouldAttempt(nowUtc);
}
/// <summary>The device answered (a Good value or a deterministic refusal): reset the streak and the window.</summary>
public void RecordAnswered()
{
lock (_backoffLock)
{
_faultStreak = 0;
_backoff.RecordSuccess();
}
}
/// <summary>
/// A comms fault: the <see cref="FaultsToMarkDeviceDown"/>th consecutive one marks the device down — opens
/// (or, after a failed retry, lengthens) the backoff window and restarts the streak.
/// </summary>
/// <param name="nowUtc">The current UTC time (the window is measured from here).</param>
public void RecordCommsFault(DateTime nowUtc)
{
lock (_backoffLock)
{
if (++_faultStreak < FaultsToMarkDeviceDown) return;
_faultStreak = 0;
_backoff.RecordFailure(nowUtc);
}
}
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, SemaphoreSlim> _rmwLocks = new();
@@ -1445,11 +1700,10 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
public void ForgetGates(string runtimeKey)
{
_runtimeLocks.TryRemove(runtimeKey, out _);
_creationLocks.TryRemove($"tag:{runtimeKey}", out _);
}
/// <summary>The number of per-key gates currently held (runtime + creation + RMW). Tests only.</summary>
internal int GateCount => _runtimeLocks.Count + _creationLocks.Count + _rmwLocks.Count;
/// <summary>The number of per-key gates currently held (runtime + RMW). Tests only.</summary>
internal int GateCount => _runtimeLocks.Count + _rmwLocks.Count;
private volatile bool _retired;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.IntegrationTests/AbLegacyDeadDeviceIsolationTests.cs
+65
View File
@@ -0,0 +1,65 @@
using System.Diagnostics;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.IntegrationTests;
/// <summary>
/// Review 07 P7 (High), wire level: one driver serving a live <c>ab_server</c> device AND a dead device (a
/// blackholed address, so every connect waits out the full timeout). Before the fix the dead device's 20 tags
/// cost 20 x Timeout per poll and the live device was read once per cycle; now the live device reads Good in
/// the same call, the dead device costs at most two timeouts, and the next poll does not touch it at all.
/// Read-only.
/// </summary>
[Collection(AbLegacyServerCollection.Name)]
[Trait("Category", "Integration")]
[Trait("Simulator", "ab_server-PCCC")]
public sealed class AbLegacyDeadDeviceIsolationTests(AbLegacyServerFixture sim)
{
// TEST-NET-1 (RFC 5737): never routed, so a connect blackholes until the driver's timeout.
private const string DeadUri = "ab://192.0.2.1:44818/1,0";
/// <summary>The live device is served promptly while a blackholed sibling is throttled.</summary>
/// <param name="profile">The simulator profile.</param>
/// <param name="transport">The transport stack the driver runs on.</param>
[AbLegacyTheory]
[MemberData(nameof(TransportMatrix.Profiles), MemberType = typeof(TransportMatrix))]
public async Task Dead_device_does_not_stall_the_live_one(AbLegacyServerProfile profile, AbLegacyTransport transport)
{
if (sim.SkipReasonFor(profile) is { } skip) Assert.Skip(skip);
var liveUri = $"ab://{sim.Host}:{sim.PortFor(profile)}/{sim.CipPath}";
var timeout = TimeSpan.FromSeconds(2);
var dead = Enumerable.Range(0, 20)
.Select(i => new AbLegacyTagDefinition($"Dead{i}", DeadUri, $"N7:{i}", AbLegacyDataType.Int))
.ToArray();
await using var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(liveUri, profile.Family), new AbLegacyDeviceOptions(DeadUri, profile.Family)],
RawTags = AbLegacyRawTags.Entries(
[new AbLegacyTagDefinition("Live", liveUri, "N7:0", AbLegacyDataType.Int), .. dead]),
Timeout = timeout,
Probe = new AbLegacyProbeOptions { Enabled = false },
Transport = transport,
}, driverInstanceId: $"ablegacy-p7-{profile.Family}-{transport}");
var ct = TestContext.Current.CancellationToken;
await drv.InitializeAsync("{}", ct);
string[] refs = ["Live", .. dead.Select(d => d.Name)];
var sw = Stopwatch.StartNew();
var first = await drv.ReadAsync(refs, ct);
sw.Stop();
first[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good, "the live device must be served in the same poll");
first.Skip(1).ShouldAllBe(s => s.StatusCode != AbLegacyStatusMapper.Good);
sw.Elapsed.ShouldBeLessThan(timeout * 3, "a dead device may cost at most two timeouts per backoff window");
sw.Restart();
var second = await drv.ReadAsync(refs, ct);
sw.Stop();
second[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
sw.Elapsed.ShouldBeLessThan(timeout, "inside the backoff window the dead device is not dialled");
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyDeviceIsolationTests.cs
+259
View File
@@ -0,0 +1,259 @@
using System.Diagnostics;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
/// <summary>
/// Review 07 P7 (High): reads were strictly serial across every device with no per-device backoff, so one dead
/// device cost <c>Timeout</c> per tag per poll and starved every healthy device — long enough to trip the
/// comms-loss watchdog against a reachable one. Devices are now read in parallel (bounded), each device pass
/// stops paying for a dead device after two consecutive comms faults and opens a per-device
/// <see cref="ConnectionBackoff"/> window, and cold connects are single-flight per device.
/// </summary>
[Trait("Category", "Unit")]
public sealed class AbLegacyDeviceIsolationTests
{
private const string HostA = "ab://10.0.0.5/1,0";
private const string HostB = "ab://10.0.0.6/1,0";
/// <summary>A tag on a dead device: each wire operation burns <paramref name="cost"/> and then fails.</summary>
private sealed class DeadTag(AbLegacyTagCreateParams p, TimeSpan cost) : FakeAbLegacyTag(p)
{
public static int WireOps;
public override async Task InitializeAsync(CancellationToken ct)
{
Interlocked.Increment(ref WireOps);
await Task.Delay(cost, ct);
throw new System.Net.Sockets.SocketException(10060);
}
public override async Task ReadAsync(CancellationToken ct)
{
Interlocked.Increment(ref WireOps);
await Task.Delay(cost, ct);
throw new System.Net.Sockets.SocketException(10060);
}
}
private static AbLegacyDriver NewDriver(FakeAbLegacyTagFactory factory, params AbLegacyTagDefinition[] tags) =>
new(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(HostA), new AbLegacyDeviceOptions(HostB)],
RawTags = AbLegacyRawTags.Entries(tags),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-p7", factory);
private static AbLegacyTagDefinition[] Tags(string prefix, string host, int n, string file = "N7") =>
[.. Enumerable.Range(0, n).Select(i => new AbLegacyTagDefinition($"{prefix}{i}", host, $"{file}:{i * 3}", AbLegacyDataType.Int))];
[Fact]
public async Task Dead_device_costs_at_most_two_timeouts_then_fails_fast_and_never_stalls_the_healthy_one()
{
var cost = TimeSpan.FromMilliseconds(150);
var factory = new FakeAbLegacyTagFactory
{
Customise = p => p.Gateway == "10.0.0.6" ? new DeadTag(p, cost) : new FakeAbLegacyTag(p) { Value = 7 },
};
var a = Tags("A", HostA, 5);
var b = Tags("B", HostB, 20);
var drv = NewDriver(factory, [.. a, .. b]);
await drv.InitializeAsync("{}", CancellationToken.None);
var refs = a.Concat(b).Select(t => t.Name).ToList();
var before = drv.GetHealth().LastSuccessfulRead;
Interlocked.Exchange(ref DeadTag.WireOps, 0);
var sw = Stopwatch.StartNew();
var first = await drv.ReadAsync(refs, CancellationToken.None);
sw.Stop();
// Serial-with-no-backoff would pay 20 x 150 ms = 3 s on device B alone.
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(1.5));
Volatile.Read(ref DeadTag.WireOps).ShouldBeLessThanOrEqualTo(2);
first.Take(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good); // positive control: A is served
first.Skip(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.BadCommunicationError);
var health = drv.GetHealth();
health.State.ShouldBe(DriverState.Degraded);
health.LastSuccessfulRead.ShouldNotBeNull();
health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(before ?? DateTime.MinValue); // A's success advances it
// Inside B's backoff window the next poll touches B's wire not at all.
var opsAfterFirst = Volatile.Read(ref DeadTag.WireOps);
sw.Restart();
var second = await drv.ReadAsync(refs, CancellationToken.None);
sw.Stop();
Volatile.Read(ref DeadTag.WireOps).ShouldBe(opsAfterFirst);
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(500));
second.Take(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good);
await drv.ShutdownAsync(CancellationToken.None);
}
[Fact]
public async Task Devices_are_read_in_parallel()
{
// Device A's read cannot complete until device B's read has STARTED: a serial walk deadlocks here.
var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var factory = new FakeAbLegacyTagFactory
{
Customise = p => p.Gateway == "10.0.0.5"
? new GatedTag(p, wait: bStarted.Task) { Value = 1 }
: new GatedTag(p, signal: bStarted) { Value = 2 },
};
var drv = NewDriver(factory,
new AbLegacyTagDefinition("A", HostA, "N7:0", AbLegacyDataType.Int),
new AbLegacyTagDefinition("B", HostB, "N7:0", AbLegacyDataType.Int));
await drv.InitializeAsync("{}", CancellationToken.None);
var r = await drv.ReadAsync(["A", "B"], CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5));
r[0].Value.ShouldBe(1);
r[1].Value.ShouldBe(2);
await drv.ShutdownAsync(CancellationToken.None);
}
private sealed class GatedTag(AbLegacyTagCreateParams p, Task? wait = null, TaskCompletionSource? signal = null)
: FakeAbLegacyTag(p)
{
public override async Task ReadAsync(CancellationToken ct)
{
signal?.TrySetResult();
if (wait is not null) await wait.WaitAsync(ct);
await base.ReadAsync(ct);
}
}
[Fact]
public async Task One_timing_out_tag_does_not_blacken_the_rest_of_its_device()
{
var factory = new FakeAbLegacyTagFactory
{
Customise = p => p.TagName == "N7:0"
? new FakeAbLegacyTag(p) { ThrowOnRead = true, Exception = new TimeoutException() }
: new FakeAbLegacyTag(p) { Value = 5 },
};
var drv = NewDriver(factory, Tags("A", HostA, 4));
await drv.InitializeAsync("{}", CancellationToken.None);
for (var poll = 0; poll < 3; poll++)
{
var r = await drv.ReadAsync(["A0", "A1", "A2", "A3"], CancellationToken.None);
r[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
r.Skip(1).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good, $"poll {poll}");
}
drv.GetHealth().State.ShouldBe(DriverState.Degraded);
await drv.ShutdownAsync(CancellationToken.None);
}
[Fact]
public async Task Device_refusal_is_per_tag_Bad_advances_LastSuccessfulRead_and_keeps_the_handle()
{
var logger = new AbLegacyTestLogger();
var factory = new FakeAbLegacyTagFactory
{
Customise = p => new FakeAbLegacyTag(p) { Status = (int)AbStatus.NotFound },
};
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(HostA)],
RawTags = AbLegacyRawTags.Entries(new AbLegacyTagDefinition("X", HostA, "N7:0", AbLegacyDataType.Int)),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-p7", factory, logger);
await drv.InitializeAsync("{}", CancellationToken.None);
var initHealth = drv.GetHealth();
await Task.Delay(20);
var r1 = await drv.ReadAsync(["X"], CancellationToken.None);
var r2 = await drv.ReadAsync(["X"], CancellationToken.None);
r1[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadNodeIdUnknown);
r2[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadNodeIdUnknown);
var health = drv.GetHealth();
health.State.ShouldBe(DriverState.Healthy); // the device answered: not a comms fault
health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(initHealth.LastSuccessfulRead!.Value);
factory.Tags["N7:0"].InitializeCount.ShouldBe(1);
factory.Tags["N7:0"].ReadCount.ShouldBe(2); // positive control: both polls really asked the device
logger.Warnings.Count(w => w.Contains("Reference=X")).ShouldBe(1);
await drv.ShutdownAsync(CancellationToken.None);
}
[Fact]
public async Task Cold_connects_on_one_device_are_single_flight()
{
var inFlight = 0;
var maxInFlight = 0;
var factory = new FakeAbLegacyTagFactory
{
Customise = p => new CountingInitTag(p, () =>
{
var now = Interlocked.Increment(ref inFlight);
int seen;
while (now > (seen = Volatile.Read(ref maxInFlight)) && Interlocked.CompareExchange(ref maxInFlight, now, seen) != seen) { }
}, () => Interlocked.Decrement(ref inFlight)) { Value = 1 },
};
var tags = Tags("A", HostA, 6);
var drv = NewDriver(factory, tags);
await drv.InitializeAsync("{}", CancellationToken.None);
// Three concurrent callers (poll loop, server read, write) each hitting different cold tags of one device.
await Task.WhenAll(
drv.ReadAsync(["A0", "A1"], CancellationToken.None),
drv.ReadAsync(["A2", "A3"], CancellationToken.None),
drv.WriteAsync([new WriteRequest("A4", 3), new WriteRequest("A5", 4)], CancellationToken.None));
maxInFlight.ShouldBe(1);
factory.Tags.Values.Count(t => t.InitializeCount == 1).ShouldBe(6); // positive control: all six connected
await drv.ShutdownAsync(CancellationToken.None);
}
private sealed class CountingInitTag(AbLegacyTagCreateParams p, Action enter, Action exit) : FakeAbLegacyTag(p)
{
public override async Task InitializeAsync(CancellationToken ct)
{
enter();
try
{
await Task.Delay(30, ct);
await base.InitializeAsync(ct);
}
finally { exit(); }
}
}
[Fact]
public async Task A_recovered_device_is_read_again_once_its_window_expires()
{
var dead = true;
var creates = 0;
var factory = new FakeAbLegacyTagFactory
{
Customise = p =>
{
Interlocked.Increment(ref creates);
return dead
? new FakeAbLegacyTag(p) { ThrowOnInitialize = true, Exception = new System.Net.Sockets.SocketException(10061) }
: new FakeAbLegacyTag(p) { Value = 9 };
},
};
var drv = NewDriver(factory, new AbLegacyTagDefinition("B", HostB, "N7:0", AbLegacyDataType.Int));
await drv.InitializeAsync("{}", CancellationToken.None);
// Two consecutive comms faults mark the device down and open its first (1 s) window.
(await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
(await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
var attempts = Volatile.Read(ref creates);
dead = false;
// Inside the window: fails fast without a connect attempt, although the device is back.
(await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
Volatile.Read(ref creates).ShouldBe(attempts);
await Task.Delay(TimeSpan.FromMilliseconds(1100));
var recovered = await drv.ReadAsync(["B"], CancellationToken.None);
recovered[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
recovered[0].Value.ShouldBe(9);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
await drv.ShutdownAsync(CancellationToken.None);
}
}