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))); +}