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

fix(ablegacy): build-and-swap device/tag tables; teardown safe vs Dispose and abandoned init

Browse Source 
Review 07 H13: _devices and _tagsByRawPath were plain dictionaries cleared and
refilled in place while the retained poll loop, the actor's GetHostStatuses and
server reads enumerated them. Both are now built locally and published by
reference under a lifecycle lock; teardown swaps in empty tables.

Lane A contract: ShutdownAsync and Dispose each retire a disjoint set of device
states, so Dispose running concurrently with a still-joining ShutdownAsync is
safe. A lifecycle epoch makes an InitializeAsync that the runtime abandoned
(overlapping the next ShutdownAsync) refuse to publish its tables, start probe
loops or write health. A runtime cached on an already-retired device state by an
in-flight read is disposed instead of leaked, and late reads/writes no longer
overwrite shutdown's health. Per-key semaphores are no longer disposed under
callers that may still hold them.

Refs #738
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:04:51 -04:00
parent 8a4cf8b823
commit 92171a01ec
4 changed files with 504 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs
+225 -137
View File
@@ -23,16 +23,30 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// <summary>Mutable so <see cref="InitializeAsync"/> can adopt a re-parsed config on reinitialize.
/// Only ever written on the init path, before any reader/session uses it.</summary>
private AbLegacyDriverOptions _options;
private volatile AbLegacyDriverOptions _options;
private readonly string _driverInstanceId;
private readonly IAbLegacyTagFactory _tagFactory;
private readonly ILogger<AbLegacyDriver> _logger;
private readonly PollGroupEngine _poll;
private readonly Dictionary<string, DeviceState> _devices = new(StringComparer.OrdinalIgnoreCase);
// Device table, keyed by HostAddress. BUILD-AND-SWAP (review 07 H13): InitializeAsync builds a fresh
// dictionary and publishes it by reference; teardown swaps in an empty one. A published dictionary is never
// mutated, so the retained poll loop, the actor's GetHostStatuses and a server read can enumerate whatever
// snapshot they hold without "Collection was modified". volatile = the publish is visible to every reader.
private volatile Dictionary<string, DeviceState> _devices = new(StringComparer.OrdinalIgnoreCase);
// v3 authored-tag table, keyed by RawPath (the tag identity + driver wire reference). RawPath is
// case-sensitive ordinal. Built in InitializeAsync from _options.RawTags via the pure mapper.
private readonly Dictionary<string, AbLegacyTagDefinition> _tagsByRawPath = new(StringComparer.Ordinal);
// case-sensitive ordinal. Built in InitializeAsync from _options.RawTags via the pure mapper, and published
// by reference alongside _devices (same build-and-swap rule).
private volatile Dictionary<string, AbLegacyTagDefinition> _tagsByRawPath = new(StringComparer.Ordinal);
// Serialises every PUBLISH/RETIRE of the two tables with the lifecycle epoch. Never held across an await.
private readonly object _lifecycleLock = new();
// Bumped by every InitializeAsync start and every teardown. An initialise only publishes if the epoch is still
// the one it took — so an InitializeAsync the runtime abandoned (Lane A: it times the attempt out and calls
// ShutdownAsync) cannot resurrect its tables and probe loops behind a driver the runtime believes stopped.
private long _lifecycleEpoch;
// Resolves a read/write/subscribe RawPath reference to its tag definition via a single hit on the
// authored _tagsByRawPath table (v3: the reference is always a RawPath; a miss is a miss).
@@ -121,9 +135,12 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
internal void HandlePollError(Exception ex)
{
_logger.LogWarning(ex, "AbLegacy poll reader failed. Driver={DriverInstanceId}", _driverInstanceId);
var current = _health;
if (current.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message);
lock (_lifecycleLock)
{
var current = _health;
if (current.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message);
}
}
/// <inheritdoc />
@@ -145,53 +162,62 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// <inheritdoc />
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
long epoch;
lock (_lifecycleLock) epoch = ++_lifecycleEpoch;
// InitializeAsync can run on a live driver without a teardown first (the IDriver contract
// allows it, and a reconnect retry can overlap a running attempt). Release the previous device states' probe loops + runtimes BEFORE the
// overwrite below, or they are orphaned beyond the reach of ShutdownAsync. Deliberately NOT
// new table is published, or they are orphaned beyond the reach of ShutdownAsync. Deliberately NOT
// ShutdownAsync: that also drops _poll and every subscription, which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
await ReleaseDeviceStatesAsync(RetireTables(bumpEpoch: false), "re-initialise").ConfigureAwait(false);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Initializing, null, null));
_logger.LogInformation(
"AbLegacy driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}",
_driverInstanceId, _transportDescription, _tagFactory.GetType().Name);
// Built locally and published by reference at the end (build-and-swap): nothing a concurrent reader holds
// is ever mutated.
var devices = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
var tags = new Dictionary<string, AbLegacyTagDefinition>(StringComparer.Ordinal);
try
{
// re-parse the supplied DriverConfig so a config change delivered through the IDriver
// contract is honoured. Without this the driver keeps serving the options it was CONSTRUCTED
// with, so an operator's edit is discarded while the deployment still seals green.
if (HasConfigBody(driverConfigJson))
_options = AbLegacyDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson);
var options = HasConfigBody(driverConfigJson)
? AbLegacyDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson)
: _options;
// Native is opt-in for every family (the default is libplctag); an operator may pick it explicitly
// even for PLC-5 / LogixPccc. Allowed, but the limitation is client-visible, so name it ONCE per instance —
// not on every init, which includes every comms-loss reconnect retry.
if (!_limitedFamilyWarned
&& _transport == AbLegacyTransport.Native
&& AbLegacyTransportResolver.NativeLimitedFamilies(_options.Devices) is { Count: > 0 } limited)
&& AbLegacyTransportResolver.NativeLimitedFamilies(options.Devices) is { Count: > 0 } limited)
{
_limitedFamilyWarned = true;
_logger.LogWarning(
"AbLegacy driver {DriverInstanceId} runs the Native transport against {Families} device(s); native " +
"support is limited there (PLC-5: N-file reads only; LogixPccc: framing unverified). Set " +
"support is limited there (PLC-5: N-file reads only, every write refused; LogixPccc: framing unverified). Set " +
"\"transport\": \"Libplctag\" to use the fallback stack.",
_driverInstanceId, string.Join(", ", limited));
}
foreach (var device in _options.Devices)
foreach (var device in options.Devices)
{
var addr = AbLegacyHostAddress.TryParse(device.HostAddress)
?? throw new InvalidOperationException(
$"AbLegacy device has invalid HostAddress '{device.HostAddress}' — expected 'ab://gateway[:port]/cip-path'.");
var profile = AbLegacyPlcFamilyProfile.ForFamily(device.PlcFamily);
_devices[device.HostAddress] = new DeviceState(addr, device, profile);
devices[device.HostAddress] = new DeviceState(addr, device, profile);
}
// Build the RawPath → definition table from the authored raw tags. Each entry's TagConfig
// blob is mapped by the pure factory; the entry's WriteIdempotent flag is threaded onto the
// def (it lives on the RawTagEntry, not inside the blob), and its DeviceName is resolved to
// the owning device's host so the existing DeviceHostAddress-keyed routing continues to work.
// A mapper miss is skipped (logged), never thrown — a bad tag must not fail the whole init.
_skippedTags.BeginRebuild(_options.RawTags.Count);
foreach (var entry in _options.RawTags)
_skippedTags.BeginRebuild(options.RawTags.Count);
foreach (var entry in options.RawTags)
{
if (!AbLegacyTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var def))
{
@@ -201,10 +227,10 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
_driverInstanceId, entry.RawPath);
continue;
}
_tagsByRawPath[entry.RawPath] = def with
tags[entry.RawPath] = def with
{
WriteIdempotent = entry.WriteIdempotent,
DeviceHostAddress = ResolveDeviceHost(entry.DeviceName),
DeviceHostAddress = ResolveDeviceHost(options, entry.DeviceName),
};
}
_skippedTags.Commit();
@@ -213,9 +239,9 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
// and String (ST-file) are not supported by all PCCC families; reject them early
// so a misconfigured tag fails at init time with a clear message rather than
// surfacing an opaque comms error at runtime.
foreach (var tag in _tagsByRawPath.Values)
foreach (var tag in tags.Values)
{
if (!_devices.TryGetValue(tag.DeviceHostAddress, out var deviceForTag)) continue;
if (!devices.TryGetValue(tag.DeviceHostAddress, out var deviceForTag)) continue;
var profile = deviceForTag.Profile;
if (tag.DataType == AbLegacyDataType.Long && !profile.SupportsLongFile)
throw new InvalidOperationException(
@@ -227,68 +253,107 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
$"(family {deviceForTag.Options.PlcFamily}) does not support ST-files.");
}
// Probe loops — one per device when enabled + probe address configured.
if (_options.Probe.Enabled && !string.IsNullOrWhiteSpace(_options.Probe.ProbeAddress))
// Publish, start the probe loops and go Healthy as ONE step under the lifecycle lock, and only if no
// teardown (or newer initialise) ran since this attempt began. Starting the loops inside the lock
// means a teardown that swaps this table out always sees their tasks and joins them.
lock (_lifecycleLock)
{
foreach (var state in _devices.Values)
if (epoch != _lifecycleEpoch)
throw new OperationCanceledException(
$"AbLegacy driver {_driverInstanceId} initialise was superseded by a shutdown or a newer initialise.");
_options = options;
_tagsByRawPath = tags;
_devices = devices;
_resolver.Clear(); // drop transient parses made against the previous table
if (options.Probe.Enabled && !string.IsNullOrWhiteSpace(options.Probe.ProbeAddress))
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
foreach (var state in devices.Values)
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
var probe = options.Probe;
state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, probe, ct), ct);
}
}
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
}
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
}
catch (Exception ex)
{
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
// Release what this attempt built. Nothing was published (the publish is the last statement of the
// try and runs entirely under the lock), so no loop is running against these states.
foreach (var state in devices.Values) state.Retire();
if (ex is OperationCanceledException && epoch != Volatile.Read(ref _lifecycleEpoch))
{
_logger.LogInformation(
"AbLegacy driver {DriverInstanceId} initialise abandoned: superseded before it could publish.",
_driverInstanceId);
throw;
}
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Faulted, null, ex.Message));
_logger.LogError(ex,
"AbLegacy driver initialise failed. Driver={DriverInstanceId}", _driverInstanceId);
// Release cached state created before the failure so a caller who catches and abandons
// (rather than retrying via ReinitializeAsync) doesn't leave CancellationTokenSources and
// libplctag handles alive. No probe-loop join here: the loops start as the LAST statement of
// the try and nothing after that can throw, so a failure never has a running loop to join.
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
_tagsByRawPath.Clear();
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
throw;
}
}
/// <summary>
/// Cancels, bounded-joins and disposes every existing device state, then clears the table.
/// Device-state part of <see cref="ShutdownAsync"/> only (no poll / subscription teardown).
/// </summary>
private async Task ReleaseDeviceStatesAsync()
/// <summary>Writes <paramref name="health"/> only while <paramref name="epoch"/> is still the current lifecycle
/// epoch, so a superseded initialise never overwrites the health of the teardown that superseded it.</summary>
private void SetHealthIfCurrent(long epoch, DriverHealth health)
{
if (_devices.Count == 0) return;
foreach (var state in _devices.Values)
lock (_lifecycleLock)
{
if (epoch == _lifecycleEpoch) _health = health;
}
}
/// <summary>The current lifecycle epoch; a read/write captures it so a late result cannot rewrite health after a teardown.</summary>
private long CurrentEpoch => Volatile.Read(ref _lifecycleEpoch);
/// <summary>
/// Atomically swaps the device table (and, for a teardown, the tag table) for an empty one and returns the
/// device states that were published, for the caller to release. Concurrent teardowns (ShutdownAsync racing
/// Dispose) each get a DISJOINT set, so no state is released twice and neither enumerates a table the other
/// is clearing.
/// </summary>
/// <param name="bumpEpoch">True for a teardown: supersedes any in-flight initialise.</param>
/// <returns>The device states that were published.</returns>
private List<DeviceState> RetireTables(bool bumpEpoch)
{
lock (_lifecycleLock)
{
if (bumpEpoch)
{
_lifecycleEpoch++;
_tagsByRawPath = new Dictionary<string, AbLegacyTagDefinition>(StringComparer.Ordinal);
}
var retired = _devices.Values.ToList();
_devices = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
return retired;
}
}
/// <summary>
/// Cancels, bounded-joins and retires the given (already unpublished) device states.
/// </summary>
/// <param name="states">The states to release.</param>
/// <param name="phase">"shutdown" / "re-initialise", for the join-timeout warning.</param>
private async Task ReleaseDeviceStatesAsync(List<DeviceState> states, string phase)
{
if (states.Count == 0) return;
foreach (var state in states)
try { state.ProbeCts?.Cancel(); } catch { }
var loops = _devices.Values.Select(d => d.ProbeTask).OfType<Task>().ToList();
var loops = states.Select(d => d.ProbeTask).OfType<Task>().ToList();
if (loops.Count > 0)
{
var all = Task.WhenAll(loops);
var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false);
if (finished != all)
_logger.LogWarning(
"AbLegacy re-initialise: {Count} probe loop(s) did not stop within {Timeout}; proceeding. Driver={DriverInstanceId}",
loops.Count, LoopJoinTimeout, _driverInstanceId);
"AbLegacy {Phase}: {Count} probe loop(s) did not stop within {Timeout}; proceeding. Driver={DriverInstanceId}",
phase, loops.Count, LoopJoinTimeout, _driverInstanceId);
}
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
foreach (var state in states) state.Retire();
}
/// <inheritdoc />
@@ -305,44 +370,35 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// Synchronous bounded join of the (already-cancelled) probe loops for the sync teardown paths.
/// The loop body uses <c>ConfigureAwait(false)</c> throughout, so blocking here cannot deadlock.
/// </summary>
private void JoinProbeLoopsSync()
/// <param name="states">The retired states whose loops to join.</param>
private static void JoinProbeLoopsSync(List<DeviceState> states)
{
var loops = _devices.Values.Select(d => d.ProbeTask).OfType<Task>().ToArray();
var loops = states.Select(d => d.ProbeTask).OfType<Task>().ToArray();
if (loops.Length == 0) return;
try { Task.WaitAll(loops, LoopJoinTimeout); }
catch (AggregateException) { /* loops observe cancellation; a fault is not actionable at teardown */ }
}
/// <inheritdoc />
/// <summary>
/// Complete teardown: stops every poll subscription, cancels and (bounded) joins the probe loops, disposes
/// every cached runtime and unpublishes both tables. Safe to run concurrently with <see cref="Dispose"/> /
/// <see cref="DisposeAsync"/> (each teardown retires a disjoint set of states) and with an in-flight
/// <see cref="InitializeAsync"/>, which it supersedes: that attempt will not publish.
/// </summary>
/// <param name="cancellationToken">Unused; teardown is bounded by <see cref="LoopJoinTimeout"/>.</param>
/// <returns>A task that completes when teardown is done.</returns>
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
// Unpublish FIRST, so an in-flight initialise is superseded and new reads resolve nothing.
var retired = RetireTables(bumpEpoch: true);
await _poll.DisposeAsync().ConfigureAwait(false);
// Cancel the probe loops, JOIN them (bounded), then dispose the CTSs and runtimes.
// The loop used to be fire-and-forget, so one parked in a probe-runtime create/read could
// outlive shutdown and keep a libplctag handle alive past DisposeRuntimes.
foreach (var state in _devices.Values)
try { state.ProbeCts?.Cancel(); } catch { }
var loops = _devices.Values.Select(d => d.ProbeTask).OfType<Task>().ToList();
if (loops.Count > 0)
{
var all = Task.WhenAll(loops);
var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false);
if (finished != all)
_logger.LogWarning(
"AbLegacy shutdown: {Count} probe loop(s) did not stop within {Timeout}; proceeding with teardown. Driver={DriverInstanceId}",
loops.Count, LoopJoinTimeout, _driverInstanceId);
}
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
_tagsByRawPath.Clear();
// outlive shutdown and keep a libplctag handle alive past retirement.
await ReleaseDeviceStatesAsync(retired, "shutdown").ConfigureAwait(false);
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
lock (_lifecycleLock)
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}
/// <inheritdoc />
@@ -369,6 +425,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var epoch = CurrentEpoch;
var now = DateTime.UtcNow;
var results = new DataValueSnapshot[fullReferences.Count];
@@ -404,8 +461,8 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
{
results[i] = new DataValueSnapshot(null, res.Status, null, now);
if (res.TransportFault)
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport fault resolving indirect address for {reference}");
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport fault resolving indirect address for {reference}"));
continue;
}
def = res.Def;
@@ -443,8 +500,8 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
EvictRuntime(device, def.Name);
results[i] = new DataValueSnapshot(null,
AbLegacyStatusMapper.MapStatus(status), null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport status {status} reading {reference}");
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport status {status} reading {reference}"));
if (!device.FirstNonZeroStatusLogged)
{
device.FirstNonZeroStatusLogged = true;
@@ -459,7 +516,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
device.FirstNonZeroStatusLogged = false;
results[i] = new DataValueSnapshot(value, AbLegacyStatusMapper.Good, now, now);
_health = new DriverHealth(DriverState.Healthy, now, null);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Healthy, now, null));
}
catch (OperationCanceledException) { throw; }
catch (AbTransportException ex) when (IsLocalRefusal(ex))
@@ -474,7 +531,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
EvictRuntime(device, def.Name);
results[i] = new DataValueSnapshot(null,
AbLegacyStatusMapper.BadCommunicationError, null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message));
}
}
@@ -488,6 +545,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(writes);
var epoch = CurrentEpoch;
var results = new WriteResult[writes.Count];
for (var i = 0; i < writes.Count; i++)
@@ -530,8 +588,8 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
{
results[i] = new WriteResult(res.Status);
if (res.TransportFault)
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport fault resolving indirect address for {w.FullReference}");
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
$"transport fault resolving indirect address for {w.FullReference}"));
continue;
}
def = res.Def;
@@ -597,7 +655,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
catch (NotSupportedException nse)
{
results[i] = new WriteResult(AbLegacyStatusMapper.BadNotSupported);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, nse.Message);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, nse.Message));
}
catch (Exception ex) when (ex is FormatException or InvalidCastException)
{
@@ -618,7 +676,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);
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message));
}
}
@@ -684,7 +742,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses() =>
[.. _devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))];
private async Task ProbeLoopAsync(DeviceState state, CancellationToken ct)
private async Task ProbeLoopAsync(DeviceState state, AbLegacyProbeOptions probe, CancellationToken ct)
{
var probeParams = new AbLegacyTagCreateParams(
Gateway: state.ParsedAddress.Gateway,
@@ -693,9 +751,9 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
LibplctagPlcAttribute: state.Profile.LibplctagPlcAttribute,
// Normalised exactly like tag paths (PLC-5 octal I/O -> decimal); an unparseable probe address is passed through
// so the transport reports the failure as it always did.
TagName: AbLegacyProbeAddress.Normalise(_options.Probe.ProbeAddress, state.Options.PlcFamily)
?? _options.Probe.ProbeAddress!,
Timeout: _options.Probe.Timeout);
TagName: AbLegacyProbeAddress.Normalise(probe.ProbeAddress, state.Options.PlcFamily)
?? probe.ProbeAddress!,
Timeout: probe.Timeout);
IAbLegacyTagRuntime? probeRuntime = null;
while (!ct.IsCancellationRequested)
@@ -722,7 +780,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
try { await Task.Delay(_options.Probe.Interval, ct).ConfigureAwait(false); }
try { await Task.Delay(probe.Interval, ct).ConfigureAwait(false); }
catch (OperationCanceledException) { break; }
}
try { probeRuntime?.Dispose(); } catch { }
@@ -768,21 +826,22 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// <c>HostAddress</c> as the connection key. The HostAddress fallback is deliberate: the Client CLI
/// passes the host as the DeviceName.
/// </remarks>
/// <param name="options">The options being initialised (their device list is the routing table).</param>
/// <param name="deviceName">The owning device name delivered on the <see cref="RawTagEntry"/>.</param>
/// <returns>The matched device's <c>HostAddress</c>; empty when unroutable.</returns>
private string ResolveDeviceHost(string deviceName)
private static string ResolveDeviceHost(AbLegacyDriverOptions options, string deviceName)
{
if (!string.IsNullOrEmpty(deviceName))
{
foreach (var d in _options.Devices)
foreach (var d in options.Devices)
if (string.Equals(d.DeviceName, deviceName, StringComparison.OrdinalIgnoreCase))
return d.HostAddress;
foreach (var d in _options.Devices)
foreach (var d in options.Devices)
if (string.Equals(d.HostAddress, deviceName, StringComparison.OrdinalIgnoreCase))
return d.HostAddress;
}
// Best-available fallback: a single-device driver routes every tag to its sole device.
return _options.Devices.Count == 1 ? _options.Devices[0].HostAddress : "";
return options.Devices.Count == 1 ? options.Devices[0].HostAddress : "";
}
// ---- IPerCallHostResolver ----
@@ -1011,7 +1070,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
TagName: parentName,
Timeout: _options.Timeout));
await InitializeOrKeepRefusedAsync(runtime, () => device.ParentRuntimes[parentName] = runtime, ct)
await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(device.ParentRuntimes, parentName, runtime), ct)
.ConfigureAwait(false);
return runtime;
}
@@ -1117,7 +1176,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
$"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.Runtimes[def.Name] = runtime, ct)
await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(device.Runtimes, def.Name, runtime), ct)
.ConfigureAwait(false);
return runtime;
}
@@ -1158,6 +1217,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// </summary>
public void Dispose()
{
var retired = RetireTables(bumpEpoch: true);
// ValueTask.ConfigureAwait(false).GetAwaiter().GetResult() — drains the poll engine
// shutdown on the current thread without capturing the SynchronizationContext. The
// engine cancels every loop CTS up-front then either completes immediately (no
@@ -1165,20 +1225,13 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
// already-shutting-down loop tasks. We swallow exceptions so a buggy poll-loop teardown
// can't poison the rest of the disposal chain.
try { _poll.DisposeAsync().ConfigureAwait(false).GetAwaiter().GetResult(); } catch { }
foreach (var state in _devices.Values)
foreach (var state in retired)
try { state.ProbeCts?.Cancel(); } catch { }
JoinProbeLoopsSync();
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeRuntimes();
}
_devices.Clear();
_tagsByRawPath.Clear();
JoinProbeLoopsSync(retired);
foreach (var state in retired) state.Retire();
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
lock (_lifecycleLock)
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}
/// <summary>
@@ -1244,7 +1297,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// 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="DisposeRuntimes"/>.
/// that would be leaked on <see cref="Retire"/>.
/// </summary>
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, SemaphoreSlim> _creationLocks = new(StringComparer.OrdinalIgnoreCase);
@@ -1326,25 +1379,60 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// <summary>Tags already warned about for a malformed address (once-per-tag log).</summary>
public System.Collections.Concurrent.ConcurrentDictionary<string, byte> MalformedAddressLogged { get; } = new();
private volatile bool _retired;
/// <summary>True once this state has been unpublished and released; nothing may be cached on it any more.</summary>
public bool IsRetired => _retired;
/// <summary>
/// Disposes all cached tag runtimes.
/// Caches a freshly initialised runtime under <paramref name="key"/>. A concurrent creator that won the
/// race keeps its runtime (the loser is disposed, never overwritten-and-leaked). A runtime cached AFTER
/// <see cref="Retire"/> drained the table — a read that was already in flight when shutdown ran — is
/// removed and disposed here, because nothing would ever release it otherwise.
/// </summary>
public void DisposeRuntimes()
/// <param name="table">The runtime table (<see cref="Runtimes"/> or <see cref="ParentRuntimes"/>).</param>
/// <param name="key">The runtime key.</param>
/// <param name="runtime">The initialised runtime.</param>
public void Cache(System.Collections.Concurrent.ConcurrentDictionary<string, IAbLegacyTagRuntime> table,
string key, IAbLegacyTagRuntime runtime)
{
foreach (var r in Runtimes.Values) r.Dispose();
Runtimes.Clear();
foreach (var r in ParentRuntimes.Values) r.Dispose();
ParentRuntimes.Clear();
if (!table.TryAdd(key, runtime))
{
if (!ReferenceEquals(table.GetValueOrDefault(key), runtime))
{
try { runtime.Dispose(); } catch { /* best-effort */ }
return;
}
}
// The ConcurrentDictionary write above is a full fence, so this read cannot be satisfied before it:
// either Retire's drain sees the entry, or this check sees the retirement.
if (_retired && table.TryRemove(new KeyValuePair<string, IAbLegacyTagRuntime>(key, runtime)))
{
try { runtime.Dispose(); } catch { /* best-effort */ }
}
}
/// <summary>
/// Retires the state: marks it, disposes the probe CTS and every cached runtime. Idempotent. The
/// per-key <see cref="SemaphoreSlim"/> gates are deliberately NOT disposed: an operation that resolved
/// this state before it was unpublished may still be waiting on or releasing one, and disposing it under
/// that caller turned a clean teardown into <see cref="ObjectDisposedException"/>s. A SemaphoreSlim whose
/// <c>AvailableWaitHandle</c> was never touched holds no unmanaged resource; the gates go with the state.
/// </summary>
public void Retire()
{
_retired = true;
var cts = ProbeCts;
ProbeCts = null;
ProbeTask = null;
try { cts?.Dispose(); } catch { /* best-effort */ }
foreach (var key in Runtimes.Keys)
if (Runtimes.TryRemove(key, out var r))
try { r.Dispose(); } catch { /* best-effort */ }
foreach (var key in ParentRuntimes.Keys)
if (ParentRuntimes.TryRemove(key, out var r))
try { r.Dispose(); } catch { /* best-effort */ }
IndirectLastName.Clear();
// Dispose + clear the per-parent RMW gates and the per-runtime/creation locks so
// ReinitializeAsync cycles don't orphan their SemaphoreSlim instances (each leaks a
// wait handle once contended).
foreach (var sem in _rmwLocks.Values) sem.Dispose();
_rmwLocks.Clear();
foreach (var sem in _creationLocks.Values) sem.Dispose();
_creationLocks.Clear();
foreach (var sem in _runtimeLocks.Values) sem.Dispose();
_runtimeLocks.Clear();
}
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyBitRmwTests.cs
+1 -1
View File
@@ -205,7 +205,7 @@ public sealed class AbLegacyBitRmwTests
var firstResult = await drv.WriteAsync([new WriteRequest("Bit2", true)], CancellationToken.None);
firstResult.Single().StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
// Teardown — DisposeRuntimes must dispose + clear the RMW lock map.
// Teardown — retirement drops the device state (and its RMW lock map) with it.
await drv.DisposeAsync();
// Second session: re-init on a fresh driver instance (mirrors ReinitializeAsync behavior).
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyLifecycleConcurrencyTests.cs
+255
View File
@@ -0,0 +1,255 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.PlcFamilies;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
/// <summary>
/// Review 07 H13 + the Lane A runtime contract: the device and tag tables are build-and-swapped (never cleared
/// in place under a reader), <c>Dispose</c> may run concurrently with a still-running <c>ShutdownAsync</c>, and
/// an abandoned <c>InitializeAsync</c> that overlaps the next <c>ShutdownAsync</c> must not resurrect state.
/// </summary>
[Trait("Category", "Unit")]
public sealed class AbLegacyLifecycleConcurrencyTests
{
private const string Host = "ab://10.0.0.5/1,0";
/// <summary>A probe tag whose read parks until released, ignoring cancellation (a wedged device call).</summary>
private sealed class ParkingProbeTag(AbLegacyTagCreateParams p, Task release) : FakeAbLegacyTag(p)
{
public readonly TaskCompletionSource Entered = new(TaskCreationOptions.RunContinuationsAsynchronously);
public override async Task ReadAsync(CancellationToken ct)
{
Entered.TrySetResult();
await release.ConfigureAwait(false);
await base.ReadAsync(ct).ConfigureAwait(false);
}
}
private static async Task WaitUntilAsync(Func<bool> condition, string what)
{
var deadline = DateTime.UtcNow.AddSeconds(10);
while (!condition())
{
if (DateTime.UtcNow > deadline) throw new TimeoutException(what);
await Task.Delay(10);
}
}
/// <summary>
/// Re-initialising in a loop while other threads read the tables never throws "Collection was modified":
/// the readers see either the old or the new table, never one being cleared.
/// </summary>
[Fact]
public async Task Reinit_under_concurrent_readers_never_throws()
{
var factory = new FakeAbLegacyTagFactory { Customise = p => new FakeAbLegacyTag(p) { Value = 1 } };
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(Host), new AbLegacyDeviceOptions("ab://10.0.0.6/1,0")],
RawTags = AbLegacyRawTags.Entries(
new AbLegacyTagDefinition("X", Host, "N7:0", AbLegacyDataType.Int),
new AbLegacyTagDefinition("Y", "ab://10.0.0.6/1,0", "N7:1", AbLegacyDataType.Int)),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-swap", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
using var stop = new CancellationTokenSource();
var failures = new System.Collections.Concurrent.ConcurrentQueue<Exception>();
var reads = 0;
var readers = Enumerable.Range(0, 3).Select(n => Task.Run(async () =>
{
while (!stop.IsCancellationRequested)
{
try
{
_ = drv.GetHostStatuses();
_ = drv.ResolveHost("X");
_ = await drv.ReadAsync(["X", "Y"], CancellationToken.None);
Interlocked.Increment(ref reads);
}
catch (Exception ex) { failures.Enqueue(ex); }
}
})).ToArray();
// Keep re-initialising until the readers have demonstrably overlapped many swaps.
var deadline = DateTime.UtcNow.AddSeconds(10);
for (var i = 0; i < 100 || Volatile.Read(ref reads) < 200; i++)
{
if (DateTime.UtcNow > deadline) break;
await drv.InitializeAsync("{}", CancellationToken.None);
}
stop.Cancel();
await Task.WhenAll(readers);
failures.ShouldBeEmpty();
reads.ShouldBeGreaterThan(0); // positive control: the readers really ran against the swapping tables
await drv.ShutdownAsync(CancellationToken.None);
}
/// <summary>A re-init publishes a NEW device state; the old one is retired, not mutated in place.</summary>
[Fact]
public async Task Reinit_publishes_a_new_table_and_retires_the_old_state()
{
var factory = new FakeAbLegacyTagFactory { Customise = p => new FakeAbLegacyTag(p) { Value = 1 } };
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(Host)],
RawTags = AbLegacyRawTags.Entries(new AbLegacyTagDefinition("X", Host, "N7:0", AbLegacyDataType.Int)),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-swap", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
(await drv.ReadAsync(["X"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
var old = drv.GetDeviceState(Host)!;
var oldTag = factory.Tags["N7:0"];
await drv.InitializeAsync("{}", CancellationToken.None);
var fresh = drv.GetDeviceState(Host)!;
fresh.ShouldNotBeSameAs(old);
old.IsRetired.ShouldBeTrue();
oldTag.Disposed.ShouldBeTrue();
(await drv.ReadAsync(["X"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
await drv.ShutdownAsync(CancellationToken.None);
}
/// <summary>
/// A runtime created by a read that was already in flight when shutdown retired its device state is disposed
/// instead of being cached on the retired state (where nothing would ever release it).
/// </summary>
[Fact]
public async Task Runtime_created_after_retirement_is_disposed_not_cached()
{
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
FakeAbLegacyTag? created = null;
var factory = new FakeAbLegacyTagFactory
{
Customise = p => created = new InitParkingTag(p, entered, release.Task) { Value = 1 },
};
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(Host)],
RawTags = AbLegacyRawTags.Entries(new AbLegacyTagDefinition("X", Host, "N7:0", AbLegacyDataType.Int)),
Probe = new AbLegacyProbeOptions { Enabled = false },
}, "drv-late", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var state = drv.GetDeviceState(Host)!;
var read = drv.ReadAsync(["X"], CancellationToken.None);
await entered.Task;
await drv.ShutdownAsync(CancellationToken.None);
release.SetResult();
_ = await read;
created.ShouldNotBeNull();
created!.Disposed.ShouldBeTrue();
state.Runtimes.ShouldBeEmpty();
drv.GetHealth().State.ShouldBe(DriverState.Unknown); // the late read did not overwrite shutdown's health
}
private sealed class InitParkingTag(AbLegacyTagCreateParams p, TaskCompletionSource entered, Task release)
: FakeAbLegacyTag(p)
{
public override async Task InitializeAsync(CancellationToken ct)
{
entered.TrySetResult();
await release.ConfigureAwait(false);
await base.InitializeAsync(ct).ConfigureAwait(false);
}
}
/// <summary>
/// Lane A's final stop calls ShutdownAsync and then DisposeAsync/Dispose; under the 5 s bound Dispose can run
/// while ShutdownAsync is still joining a wedged probe loop. Both complete without throwing and every runtime
/// is released.
/// </summary>
[Fact]
public async Task Dispose_concurrent_with_running_shutdown_is_safe()
{
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
ParkingProbeTag? probe = null;
var factory = new FakeAbLegacyTagFactory
{
Customise = p => p.TagName == "S:0"
? probe = new ParkingProbeTag(p, release.Task)
: new FakeAbLegacyTag(p) { Value = 1 },
};
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(Host)],
RawTags = AbLegacyRawTags.Entries(new AbLegacyTagDefinition("X", Host, "N7:0", AbLegacyDataType.Int)),
Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20) },
}, "drv-dispose", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
(await drv.ReadAsync(["X"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
await WaitUntilAsync(() => probe is not null, "probe never created");
await probe!.Entered.Task;
var shutdown = drv.ShutdownAsync(CancellationToken.None); // parks joining the wedged probe loop
await Task.Delay(50);
shutdown.IsCompleted.ShouldBeFalse();
Should.NotThrow(() => drv.Dispose());
await Should.NotThrowAsync(async () => await drv.DisposeAsync());
release.SetResult();
await shutdown;
factory.Tags["N7:0"].Disposed.ShouldBeTrue();
probe.Disposed.ShouldBeTrue();
drv.DeviceCount.ShouldBe(0);
}
/// <summary>
/// An InitializeAsync the runtime abandoned (it is parked releasing a wedged prior probe loop) overlaps the next
/// ShutdownAsync. When it resumes it must NOT publish its tables or start probe loops — that state would sit
/// behind a driver the runtime believes is stopped. A later Initialize publishes normally (positive control).
/// </summary>
[Fact]
public async Task Abandoned_initialize_overlapping_shutdown_does_not_resurrect_state()
{
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var probes = new System.Collections.Concurrent.ConcurrentQueue<FakeAbLegacyTag>();
ParkingProbeTag? first = null;
var factory = new FakeAbLegacyTagFactory
{
Customise = p =>
{
FakeAbLegacyTag t = first is null ? first = new ParkingProbeTag(p, release.Task) : new FakeAbLegacyTag(p);
probes.Enqueue(t);
return t;
},
};
var drv = new AbLegacyDriver(new AbLegacyDriverOptions
{
Devices = [new AbLegacyDeviceOptions(Host, AbLegacyPlcFamily.Slc500)],
Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20) },
}, "drv-abandon", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
await WaitUntilAsync(() => first is not null, "probe never created");
await first!.Entered.Task;
var abandoned = drv.InitializeAsync("{}", CancellationToken.None); // parks joining the wedged loop
await Task.Delay(50);
abandoned.IsCompleted.ShouldBeFalse();
await drv.ShutdownAsync(CancellationToken.None);
release.SetResult();
await Should.ThrowAsync<OperationCanceledException>(() => abandoned);
var createdAfter = probes.Count;
await Task.Delay(150); // several probe intervals
drv.DeviceCount.ShouldBe(0);
drv.GetHealth().State.ShouldBe(DriverState.Unknown);
probes.Count.ShouldBe(createdAfter);
// Positive control: a fresh Initialize after the stop publishes and probes as normal.
await drv.InitializeAsync("{}", CancellationToken.None);
drv.DeviceCount.ShouldBe(1);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
await WaitUntilAsync(() => probes.Count > createdAfter, "fresh init never probed");
await drv.ShutdownAsync(CancellationToken.None);
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyTestLogger.cs
+23
View File
@@ -0,0 +1,23 @@
using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests;
/// <summary>Thread-safe capturing logger shared by the remediation tests (poll loops log from pool threads).</summary>
internal sealed class AbLegacyTestLogger : ILogger<AbLegacyDriver>
{
/// <summary>Every entry logged, in order.</summary>
public System.Collections.Concurrent.ConcurrentQueue<(LogLevel Level, string Message)> Entries { get; } = new();
/// <summary>The Warning-level messages logged.</summary>
public IReadOnlyList<string> Warnings => [.. Entries.Where(e => e.Level == LogLevel.Warning).Select(e => e.Message)];
/// <inheritdoc />
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
/// <inheritdoc />
public bool IsEnabled(LogLevel logLevel) => true;
/// <inheritdoc />
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter) => Entries.Enqueue((logLevel, formatter(state, exception)));
}