fix(abcip): build-and-swap device tables, volatile health, async-only dispose

H13: _devices/_devicesByRoutingKey are published by reference swap under a
lifecycle lock and never mutated after publish, so the retained poll engine,
discovery and the actor's GetHostStatuses pull never read a table mid-Clear.
Every retire (init's own release, shutdown, dispose) takes the table it
replaces, so ShutdownAsync racing DisposeAsync tears each state down once, and
a generation check makes an InitializeAsync overtaken by a shutdown publish
nothing (it throws OperationCanceledException) instead of resurrecting probe
loops behind the shutdown. Probe join is one bounded 5 s wait independent of
the caller token.

H21: _health is volatile and read-modify-write updates go through a CAS.

F10: the sync Dispose() (sync-over-async) is removed; DisposeAsync is the
complete, idempotent teardown the runtime calls after ShutdownAsync.

Refs #737
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:02:00 -04:00
parent 8a4cf8b823
commit 20cb2a52d6
2 changed files with 333 additions and 99 deletions
@@ -27,7 +27,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// handles GC-collected tags.</para>
/// </remarks>
public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, IBatchSubscribable,
IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable
IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, ISkippedTagReporter, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table.</summary>
private readonly SkippedTagLedger _skippedTags = new();
@@ -98,12 +98,21 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
private readonly PollGroupEngine _poll;
// Devices keyed by host address — the primary registry (GetDeviceState, discovery, probe, diagnostics).
private readonly Dictionary<string, DeviceState> _devices = new(StringComparer.OrdinalIgnoreCase);
// BUILD-AND-SWAP, never mutated after publish: the retained poll engine, the alarm projection, discovery and
// the actor's GetHostStatuses pull all read it while a re-init / shutdown runs. Each teardown takes the table
// it replaces, so concurrent teardowns (ShutdownAsync racing DisposeAsync) own disjoint device states.
private volatile Dictionary<string, DeviceState> _devices = new(StringComparer.OrdinalIgnoreCase);
// Tag-routing index: a tag's device routing key (DeviceName ?? HostAddress, plus the HostAddress alias)
// → its DeviceState. A tag resolves to its device by DeviceName-or-host so multi-device routing keys on
// the RawPath's device segment. The HostAddress alias is deliberate (Client CLI + tests route by host);
// _options.Devices is already the merged per-device DeviceConfig.
private readonly Dictionary<string, DeviceState> _devicesByRoutingKey = new(StringComparer.OrdinalIgnoreCase);
// _options.Devices is already the merged per-device DeviceConfig. Swapped with _devices, same rule.
private volatile Dictionary<string, DeviceState> _devicesByRoutingKey = new(StringComparer.OrdinalIgnoreCase);
// Serialises table publish/retire. _lifecycleGeneration is bumped by every retire (an init's own release, a
// shutdown, a dispose); an init publishes only if no retire happened after its own, so an abandoned init that a
// shutdown overtook can never resurrect devices or probe loops after that shutdown returned.
private readonly object _lifecycleLock = new();
private int _lifecycleGeneration;
// v3 authored RawPath → definition table (parents + fanned-out UDT members), keyed case-sensitive
// ordinal. Built in InitializeAsync from _options.RawTags via the pure mapper.
private volatile Dictionary<string, AbCipTagDefinition> _tagsByRawPath = new(StringComparer.Ordinal);
@@ -133,7 +142,24 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// <returns><see langword="true"/> when a definition exists for <paramref name="rawPath"/>.</returns>
internal bool TryGetTagDefinition(string rawPath, out AbCipTagDefinition definition) =>
_tagsByRawPath.TryGetValue(rawPath, out definition!);
private DriverHealth _health = new(DriverState.Unknown, null, null);
// volatile: written from the read batch, the write path, the poll engine's error hook and the lifecycle, and
// read from the actor thread (GetHealth). Read-modify-write updates go through UpdateHealth (CAS) so a
// concurrent update is never lost.
private volatile DriverHealth _health = new(DriverState.Unknown, null, null);
/// <summary>Atomically replaces <see cref="_health"/> with <paramref name="update"/> applied to its current value.</summary>
/// <param name="update">Computes the new health from the current one.</param>
private void UpdateHealth(Func<DriverHealth, DriverHealth> update)
{
while (true)
{
var current = _health;
var next = update(current);
#pragma warning disable CS0420 // a volatile field passed by ref to Interlocked keeps its full-fence semantics
if (ReferenceEquals(Interlocked.CompareExchange(ref _health, next, current), current)) return;
#pragma warning restore CS0420
}
}
/// <summary>Occurs when a subscribed tag's value changes.</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
@@ -270,9 +296,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
internal void HandlePollError(Exception ex)
{
_logger.LogWarning(ex, "AbCip poll reader failed. Driver={DriverInstanceId}", _driverInstanceId);
var current = _health;
if (current.State != DriverState.Faulted)
_health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message);
UpdateHealth(current => current.State == DriverState.Faulted
? current
: new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message));
}
/// <summary>
@@ -364,48 +390,58 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract
// allows it, and a reconnect retry can overlap a running attempt). Release the previous device states' probe loops + handles before
// the overwrite below. Deliberately NOT ShutdownAsync: that also drops _poll and the subscriptions.
await ReleaseDeviceStatesAsync(cancellationToken).ConfigureAwait(false);
_health = new DriverHealth(DriverState.Initializing, null, null);
// allows it, and a reconnect retry can overlap a running attempt). Retire the previous device states' probe
// loops + handles before building new ones. Deliberately NOT ShutdownAsync: that also drops _poll and the
// subscriptions. The generation this release produced is what the publish below is checked against.
var generation = await ReleaseDeviceStatesAsync().ConfigureAwait(false);
lock (_lifecycleLock)
{
if (generation == _lifecycleGeneration)
_health = new DriverHealth(DriverState.Initializing, null, null);
}
_logger.LogInformation(
"AbCip driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}",
_driverInstanceId, _transportDescription, _tagFactory.GetType().Name);
Dictionary<string, DeviceState>? builtDevices = null;
try
{
var options = _options;
var adoptedConfig = false;
if (!string.IsNullOrWhiteSpace(driverConfigJson))
{
var parsed = AbCipDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson);
if (parsed.Devices.Count > 0 || parsed.RawTags.Count > 0)
{
_options = parsed;
_alarmProjection = new AbCipAlarmProjection(this, _options.AlarmPollInterval, _logger);
options = parsed;
adoptedConfig = true;
}
}
foreach (var device in _options.Devices)
builtDevices = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
var builtRoutes = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
foreach (var device in options.Devices)
{
var addr = AbCipHostAddress.TryParse(device.HostAddress)
?? throw new InvalidOperationException(
$"AbCip device has invalid HostAddress '{device.HostAddress}' — expected 'ab://gateway[:port]/cip-path'.");
var profile = AbCipPlcFamilyProfile.ForFamily(device.PlcFamily);
var state = new DeviceState(addr, device, profile);
_devices[device.HostAddress] = state;
builtDevices[device.HostAddress] = state;
// Route index: a tag reaches this device by matching its DeviceName routing key OR its
// host address. Devices with no explicit DeviceName are identified purely by host today.
_devicesByRoutingKey[device.HostAddress] = state;
builtRoutes[device.HostAddress] = state;
if (!string.IsNullOrEmpty(device.DeviceName))
_devicesByRoutingKey[device.DeviceName] = state;
builtRoutes[device.DeviceName] = state;
}
// Build the RawPath → definition table from the authored raw tags. Each entry's TagConfig blob
// is mapped by the pure factory; the entry's DeviceName (routing key) + WriteIdempotent flag are
// threaded onto the def (they live on the RawTagEntry, not inside the blob). A mapper miss is
// skipped (logged), never thrown — a bad tag must not fail the whole driver init.
_skippedTags.BeginRebuild(_options.RawTags.Count);
_skippedTags.BeginRebuild(options.RawTags.Count);
var builtTagsByRawPath = new Dictionary<string, AbCipTagDefinition>(StringComparer.Ordinal);
var builtDeclaredTags = new List<AbCipTagDefinition>();
foreach (var entry in _options.RawTags)
foreach (var entry in options.RawTags)
{
if (!AbCipTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var mapped))
{
@@ -457,24 +493,43 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
}
}
}
// Publish the finished tables by reference swap — readers see the old or the new, never a half-built one.
_tagsByRawPath = builtTagsByRawPath;
_declaredTags = builtDeclaredTags;
_skippedTags.Commit();
// Probe loops — one per device when enabled + a ProbeTagPath is configured.
if (_options.Probe.Enabled && !string.IsNullOrWhiteSpace(_options.Probe.ProbeTagPath))
// Publish everything under the lifecycle lock, and only if no shutdown / other init retired the tables
// after this init's own release — readers see the old tables or the new ones, never a half-built one,
// and an overtaken init leaves nothing running behind the shutdown that overtook it.
lock (_lifecycleLock)
{
foreach (var state in _devices.Values)
if (generation != _lifecycleGeneration)
throw new OperationCanceledException(
$"AbCip driver {_driverInstanceId} initialization was superseded by a shutdown or a newer initialization.");
_options = options;
if (adoptedConfig)
_alarmProjection = new AbCipAlarmProjection(this, options.AlarmPollInterval, _logger);
_devices = builtDevices;
_devicesByRoutingKey = builtRoutes;
_tagsByRawPath = builtTagsByRawPath;
_declaredTags = builtDeclaredTags;
_skippedTags.Commit();
// Probe loops — one per device when enabled + a ProbeTagPath is configured. Started inside the lock so
// a retire that follows this publish always sees (and joins) them.
if (options.Probe.Enabled && !string.IsNullOrWhiteSpace(options.Probe.ProbeTagPath))
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
// Keep the loop Task so ShutdownAsync can await its clean exit before
// disposing the CTS / handles the loop is still using.
state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
foreach (var state in builtDevices.Values)
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
// Keep the loop Task so a teardown can await its clean exit before
// disposing the CTS / handles the loop is still using.
state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
}
}
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
}
else if (_options.Probe.Enabled && _devices.Count > 0)
builtDevices = null; // published: owned by the next retire now
if (options.Probe.Enabled && string.IsNullOrWhiteSpace(options.Probe.ProbeTagPath) && options.Devices.Count > 0)
{
// Probe is Enabled but no ProbeTagPath is configured. Without a
// tag path the loop has nothing to read, so HostState would stay Unknown forever
@@ -486,51 +541,84 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
"as Unknown until a ProbeTagPath is set or Probe.Enabled is set to false.",
_driverInstanceId);
}
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
}
catch (Exception ex)
{
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
// Never-published states hold no probe loop and no runtime yet; dispose for symmetry.
if (builtDevices is not null)
foreach (var state in builtDevices.Values) state.DisposeHandles();
if (ex is OperationCanceledException)
{
_logger.LogInformation("AbCip driver {DriverInstanceId} initialization abandoned: {Reason}",
_driverInstanceId, ex.Message);
throw;
}
lock (_lifecycleLock)
{
// A superseded init must not overwrite the health of whatever superseded it.
if (generation == _lifecycleGeneration)
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
}
_logger.LogError(ex, "AbCip driver {DriverInstanceId} failed to initialize", _driverInstanceId);
throw;
}
}
/// <summary>Bound on joining the retired probe loops — matches the fleet's "≤ 5 s join" teardown rule.</summary>
private static readonly TimeSpan ProbeJoinTimeout = TimeSpan.FromSeconds(5);
/// <summary>
/// Cancels, joins (bounded) and disposes every existing device state's probe loop + handles, then
/// clears the device/route tables. Device part of <see cref="ShutdownAsync"/> only.
/// Retires the published device states: swaps empty device/route/tag tables in under the lifecycle lock
/// (bumping the generation), then cancels, joins (bounded, independent of any caller token — a cancelled
/// caller must not dispose handles under a probe loop that is still running) and disposes the states it took.
/// Concurrent callers each take a disjoint table, so every state is torn down exactly once. The device part of
/// <see cref="ShutdownAsync"/> and the first step of <see cref="InitializeAsync"/>.
/// </summary>
private async Task ReleaseDeviceStatesAsync(CancellationToken cancellationToken)
/// <returns>The lifecycle generation this retire produced.</returns>
private async Task<int> ReleaseDeviceStatesAsync()
{
// The projection is replaced (not reused) when a config body is adopted; dispose the old one so its
// poll loop does not outlive it and raise duplicate alarm events through this driver.
await _alarmProjection.DisposeAsync().ConfigureAwait(false);
foreach (var state in _devices.Values)
Dictionary<string, DeviceState> retired;
AbCipAlarmProjection projection;
int generation;
lock (_lifecycleLock)
{
generation = ++_lifecycleGeneration;
retired = _devices;
_devices = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
_devicesByRoutingKey = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
// Rebuilt wholesale by Initialize; swap, never Clear(): the retained poll engine / discovery may be enumerating.
_tagsByRawPath = new(StringComparer.Ordinal);
_declaredTags = new();
projection = _alarmProjection;
}
// The projection is replaced (not reused) when a config body is adopted; dispose it so its poll loop does
// not outlive it and raise duplicate alarm events through this driver. Idempotent, and reusable afterwards.
await projection.DisposeAsync().ConfigureAwait(false);
foreach (var state in retired.Values)
{
try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { }
}
foreach (var state in _devices.Values)
var loops = retired.Values.Select(s => s.ProbeTask).OfType<Task>().ToArray();
if (loops.Length > 0)
{
var probeTask = state.ProbeTask;
if (probeTask is null) continue;
// Own bounded timeout, independent of the caller token: cancelling init must not abort the join
// and dispose handles under a probe loop that is still running.
try { await probeTask.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); }
catch (TimeoutException) { }
catch (OperationCanceledException) { }
try { await Task.WhenAll(loops).WaitAsync(ProbeJoinTimeout).ConfigureAwait(false); }
catch (TimeoutException)
{
_logger.LogWarning("AbCip driver {DriverInstanceId}: a probe loop did not exit within {Timeout}; disposing anyway",
_driverInstanceId, ProbeJoinTimeout);
}
catch (Exception) { /* a loop never faults by design; nothing to add here */ }
}
foreach (var state in _devices.Values)
foreach (var state in retired.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeHandles();
}
_devices.Clear();
_devicesByRoutingKey.Clear();
// Rebuilt wholesale by Initialize; stale entries would trip its member fan-out collision check.
_tagsByRawPath = new(StringComparer.Ordinal); // swap, never Clear(): the retained poll engine / discovery may be enumerating
_declaredTags = new();
return generation;
}
/// <inheritdoc />
@@ -541,46 +629,17 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
}
/// <inheritdoc />
/// <remarks>
/// A complete teardown: the poll engine (and every subscription), the alarm projection, every probe loop
/// (joined ≤ 5 s) and every runtime handle. Safe to run concurrently with <see cref="DisposeAsync"/> and with
/// an <see cref="InitializeAsync"/> it overtakes (that init then publishes nothing).
/// </remarks>
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
await _alarmProjection.DisposeAsync().ConfigureAwait(false);
await _poll.DisposeAsync().ConfigureAwait(false);
// Phase 1: signal every probe loop to stop.
foreach (var state in _devices.Values)
{
try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { }
}
// Phase 2: wait for each probe loop to observe cancellation and exit. The loop never
// throws on cancellation (it catches OperationCanceledException internally), but guard
// anyway so one slow device can't wedge the whole shutdown.
foreach (var state in _devices.Values)
{
var probeTask = state.ProbeTask;
if (probeTask is null) continue;
try
{
await probeTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).ConfigureAwait(false);
}
catch (TimeoutException) { }
catch (OperationCanceledException) { }
}
// Phase 3: now the loops are gone, dispose the CTS + native handles with no live reader.
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.DisposeHandles();
}
_devices.Clear();
_devicesByRoutingKey.Clear();
_tagsByRawPath = new(StringComparer.Ordinal);
_declaredTags = new();
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
UpdateHealth(h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null));
}
// ---- ISubscribable (polling overlay via shared engine) ----
@@ -875,10 +934,10 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
// exactly as the per-item writes did before.
if (worst is not { } worstState) return;
_health = new DriverHealth(
UpdateHealth(current => new DriverHealth(
worstState,
anySuccess ? now : _health.LastSuccessfulRead,
worstState == DriverState.Healthy ? null : worstError);
anySuccess ? now : current.LastSuccessfulRead,
worstState == DriverState.Healthy ? null : worstError));
}
private async Task<ReadOutcome> ReadSingleAsync(
@@ -1896,10 +1955,13 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery,
internal DeviceState? GetDeviceState(string hostAddress) =>
_devices.TryGetValue(hostAddress, out var s) ? s : null;
/// <summary>Releases all resources used by the driver.</summary>
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
/// <summary>Asynchronously releases all resources used by the driver.</summary>
/// <summary>
/// Asynchronously releases all resources used by the driver — the same complete teardown as
/// <see cref="ShutdownAsync"/>. Idempotent, and safe CONCURRENTLY with a still-running
/// <see cref="ShutdownAsync"/> (the runtime's final stop abandons a shutdown that overruns its budget and
/// disposes anyway): each teardown retires a disjoint device table. There is deliberately no synchronous
/// <c>Dispose()</c> — it could only block on this teardown (sync-over-async).
/// </summary>
/// <returns>A value task representing the asynchronous disposal.</returns>
public async ValueTask DisposeAsync()
{