From 92171a01ec4ca2428c0ce6b2801c791ab1764a7e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:04:51 -0400 Subject: [PATCH] fix(ablegacy): build-and-swap device/tag tables; teardown safe vs Dispose and abandoned init 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 --- .../AbLegacyDriver.cs | 362 +++++++++++------- .../AbLegacyBitRmwTests.cs | 2 +- .../AbLegacyLifecycleConcurrencyTests.cs | 255 ++++++++++++ .../AbLegacyTestLogger.cs | 23 ++ 4 files changed, 504 insertions(+), 138 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyLifecycleConcurrencyTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyTestLogger.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs index 236720f9..66db900d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs @@ -23,16 +23,30 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover /// Mutable so can adopt a re-parsed config on reinitialize. /// Only ever written on the init path, before any reader/session uses it. - private AbLegacyDriverOptions _options; + private volatile AbLegacyDriverOptions _options; private readonly string _driverInstanceId; private readonly IAbLegacyTagFactory _tagFactory; private readonly ILogger _logger; private readonly PollGroupEngine _poll; - private readonly Dictionary _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 _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 _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 _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); + } } /// @@ -145,53 +162,62 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover /// 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(StringComparer.OrdinalIgnoreCase); + var tags = new Dictionary(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; } } - /// - /// Cancels, bounded-joins and disposes every existing device state, then clears the table. - /// Device-state part of only (no poll / subscription teardown). - /// - private async Task ReleaseDeviceStatesAsync() + /// Writes only while is still the current lifecycle + /// epoch, so a superseded initialise never overwrites the health of the teardown that superseded it. + 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; + } + } + + /// The current lifecycle epoch; a read/write captures it so a late result cannot rewrite health after a teardown. + private long CurrentEpoch => Volatile.Read(ref _lifecycleEpoch); + + /// + /// 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. + /// + /// True for a teardown: supersedes any in-flight initialise. + /// The device states that were published. + private List RetireTables(bool bumpEpoch) + { + lock (_lifecycleLock) + { + if (bumpEpoch) + { + _lifecycleEpoch++; + _tagsByRawPath = new Dictionary(StringComparer.Ordinal); + } + var retired = _devices.Values.ToList(); + _devices = new Dictionary(StringComparer.OrdinalIgnoreCase); + return retired; + } + } + + /// + /// Cancels, bounded-joins and retires the given (already unpublished) device states. + /// + /// The states to release. + /// "shutdown" / "re-initialise", for the join-timeout warning. + private async Task ReleaseDeviceStatesAsync(List 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().ToList(); + var loops = states.Select(d => d.ProbeTask).OfType().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(); } /// @@ -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 ConfigureAwait(false) throughout, so blocking here cannot deadlock. /// - private void JoinProbeLoopsSync() + /// The retired states whose loops to join. + private static void JoinProbeLoopsSync(List states) { - var loops = _devices.Values.Select(d => d.ProbeTask).OfType().ToArray(); + var loops = states.Select(d => d.ProbeTask).OfType().ToArray(); if (loops.Length == 0) return; try { Task.WaitAll(loops, LoopJoinTimeout); } catch (AggregateException) { /* loops observe cancellation; a fault is not actionable at teardown */ } } - /// + /// + /// 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 / + /// (each teardown retires a disjoint set of states) and with an in-flight + /// , which it supersedes: that attempt will not publish. + /// + /// Unused; teardown is bounded by . + /// A task that completes when teardown is done. 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().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); } /// @@ -369,6 +425,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover IReadOnlyList 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 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 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 /// HostAddress as the connection key. The HostAddress fallback is deliberate: the Client CLI /// passes the host as the DeviceName. /// + /// The options being initialised (their device list is the routing table). /// The owning device name delivered on the . /// The matched device's HostAddress; empty when unroutable. - 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 /// 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); } /// @@ -1244,7 +1297,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover /// Per-key creation locks for and . /// 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 . + /// that would be leaked on . /// private readonly System.Collections.Concurrent.ConcurrentDictionary _creationLocks = new(StringComparer.OrdinalIgnoreCase); @@ -1326,25 +1379,60 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover /// Tags already warned about for a malformed address (once-per-tag log). public System.Collections.Concurrent.ConcurrentDictionary MalformedAddressLogged { get; } = new(); + private volatile bool _retired; + + /// True once this state has been unpublished and released; nothing may be cached on it any more. + public bool IsRetired => _retired; + /// - /// Disposes all cached tag runtimes. + /// Caches a freshly initialised runtime under . A concurrent creator that won the + /// race keeps its runtime (the loser is disposed, never overwritten-and-leaked). A runtime cached AFTER + /// 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. /// - public void DisposeRuntimes() + /// The runtime table ( or ). + /// The runtime key. + /// The initialised runtime. + public void Cache(System.Collections.Concurrent.ConcurrentDictionary 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(key, runtime))) + { + try { runtime.Dispose(); } catch { /* best-effort */ } + } + } + + /// + /// Retires the state: marks it, disposes the probe CTS and every cached runtime. Idempotent. The + /// per-key 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 s. A SemaphoreSlim whose + /// AvailableWaitHandle was never touched holds no unmanaged resource; the gates go with the state. + /// + 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(); } } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyBitRmwTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyBitRmwTests.cs index 3229fb53..6d7c1a1b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyBitRmwTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyBitRmwTests.cs @@ -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). diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyLifecycleConcurrencyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyLifecycleConcurrencyTests.cs new file mode 100644 index 00000000..58f4f95d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyLifecycleConcurrencyTests.cs @@ -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; + +/// +/// Review 07 H13 + the Lane A runtime contract: the device and tag tables are build-and-swapped (never cleared +/// in place under a reader), Dispose may run concurrently with a still-running ShutdownAsync, and +/// an abandoned InitializeAsync that overlaps the next ShutdownAsync must not resurrect state. +/// +[Trait("Category", "Unit")] +public sealed class AbLegacyLifecycleConcurrencyTests +{ + private const string Host = "ab://10.0.0.5/1,0"; + + /// A probe tag whose read parks until released, ignoring cancellation (a wedged device call). + 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 condition, string what) + { + var deadline = DateTime.UtcNow.AddSeconds(10); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException(what); + await Task.Delay(10); + } + } + + /// + /// 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. + /// + [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(); + 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); + } + + /// A re-init publishes a NEW device state; the old one is retired, not mutated in place. + [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); + } + + /// + /// 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). + /// + [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); + } + } + + /// + /// 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. + /// + [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); + } + + /// + /// 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). + /// + [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(); + 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(() => 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); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyTestLogger.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyTestLogger.cs new file mode 100644 index 00000000..fd946ffb --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyTestLogger.cs @@ -0,0 +1,23 @@ +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests; + +/// Thread-safe capturing logger shared by the remediation tests (poll loops log from pool threads). +internal sealed class AbLegacyTestLogger : ILogger +{ + /// Every entry logged, in order. + public System.Collections.Concurrent.ConcurrentQueue<(LogLevel Level, string Message)> Entries { get; } = new(); + + /// The Warning-level messages logged. + public IReadOnlyList Warnings => [.. Entries.Where(e => e.Level == LogLevel.Warning).Select(e => e.Message)]; + + /// + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + /// + public bool IsEnabled(LogLevel logLevel) => true; + + /// + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) => Entries.Enqueue((logLevel, formatter(state, exception))); +}