diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs index 1781bedf..a779b09b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs @@ -721,15 +721,24 @@ internal sealed class AdsTwinCATClient : ITwinCATClient value = DecodeString(reg.Type, raw, reg.StringsUtf8); else if (TwinCATTemporalCodec.IsTemporal(reg.Type) && value is uint rawTemporal) value = TwinCATTemporalCodec.Decode(reg.Type, rawTemporal); - _notificationQueue.Writer.TryWrite(new PendingNotification(reg, value)); + if (!_notificationQueue.Writer.TryWrite(new PendingNotification(reg, value))) + Interlocked.Increment(ref _droppedNotifications); // queue full (DropWrite): count, never block } + private long _droppedNotifications; + + /// + public long DroppedNotificationCount => Interlocked.Read(ref _droppedNotifications); + private async Task DispatchLoopAsync(CancellationToken ct) { try { await foreach (var pending in _notificationQueue.Reader.ReadAllAsync(ct).ConfigureAwait(false)) { + // A value queued before its registration was deleted is purged here rather than delivered + // after the unsubscribe. + if (pending.Registration.IsDeleted) continue; try { pending.Registration.OnChange(pending.Registration.SymbolPath, pending.Value); } catch { /* consumer-side errors stay on this managed task, not the ADS thread */ } } @@ -980,9 +989,15 @@ internal sealed class AdsTwinCATClient : ITwinCATClient /// public void Dispose() { + IsDeleted = true; // Fire-and-forget AMS call — caller has already committed to the tear-down. _ = owner.DeleteNotificationAsync(handle, CancellationToken.None); } + + /// Gets a value indicating whether the registration was disposed; queued values for it are dropped. + public bool IsDeleted { get => Volatile.Read(ref _deleted); private set => Volatile.Write(ref _deleted, value); } + + private bool _deleted; } private static Type MapToClrType(TwinCATDataType type) => type switch diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/ITwinCATClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/ITwinCATClient.cs index b44f3038..cb552f0c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/ITwinCATClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/ITwinCATClient.cs @@ -102,6 +102,12 @@ public interface ITwinCATClient : IDisposable /// The overall ADS code of the most recent such fallback (0 = a malformed response). uint LastSumFallbackCode => 0; + /// + /// Native notifications this client dropped because its bounded dispatch queue was full (the router + /// thread never blocks on a slow consumer). The driver warns when it grows. Default 0. + /// + long DroppedNotificationCount => 0; + /// /// Write a symbolic value. Returns the mapped OPC UA status for the operation /// (0 = Good, non-zero = error mapped via ). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs index 30d9f488..8542bcad 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs @@ -22,30 +22,60 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery public SkippedTagReport? GetSkippedTags() => _skippedTags.Snapshot(); // Mutable so ReinitializeAsync can apply a new config generation. The constructor seeds - // it; InitializeAsync re-parses driverConfigJson over the top of it. - private TwinCATDriverOptions _options; + // it; InitializeAsync re-parses driverConfigJson over the top of it (committed under _lifecycleLock). + private volatile TwinCATDriverOptions _options; private readonly string _driverInstanceId; private readonly ITwinCATClientFactory _clientFactory; private readonly ILogger _logger; private readonly PollGroupEngine _poll; - // ConcurrentDictionary so ShutdownAsync (Clear) and ReadAsync/WriteAsync/SubscribeAsync - // (TryGetValue) don't race — plain Dictionary is not safe for concurrent read+write. - private readonly ConcurrentDictionary _devices = - new(StringComparer.OrdinalIgnoreCase); + + // Device + tag tables are BUILD-AND-SWAP (#739 L8): InitializeAsync builds fresh tables and publishes + // them by reference under _lifecycleLock; ShutdownAsync / Dispose swap in empty ones and release what + // they took. A retained poll loop, the probe tick or an in-flight write therefore always sees one whole + // generation — never a table mid-Clear — and two overlapping teardowns can never release the same + // device twice (each device set is taken by exactly one swap). + private volatile ConcurrentDictionary _devices = NewDeviceTable(); // v3: the authored RawPath → definition table. Keyed case-sensitive ordinal (the RawPath is // the exact wire reference). Built in InitializeAsync from _options.RawTags via the pure mapper. - private readonly ConcurrentDictionary _tagsByRawPath = - new(StringComparer.Ordinal); + private volatile IReadOnlyDictionary _tagsByRawPath = EmptyTags; + + private static readonly IReadOnlyDictionary EmptyTags = + new Dictionary(StringComparer.Ordinal); + + private static ConcurrentDictionary NewDeviceTable() => new(StringComparer.OrdinalIgnoreCase); + + // Serialises the table swaps against each other. _lifecycleGeneration is bumped by every + // InitializeAsync and every teardown, so an InitializeAsync that was overtaken while it awaited + // (an abandoned init overlapping the runtime's next ShutdownAsync, or a newer init) sees a + // different generation at its commit and releases what it built instead of installing it. + private readonly object _lifecycleLock = new(); + private long _lifecycleGeneration; + + /// Test seam: awaited after the new generation is built and before it is committed, so a test can + /// interleave a teardown deterministically. Never set in production. + internal Func? BeforeCommitForTest { get; set; } + + /// Upper bound on joining released devices' probe loops (one wait for all of them). + internal static readonly TimeSpan ProbeJoinTimeout = TimeSpan.FromSeconds(2); + + /// + /// Upper bound on the whole of . The runtime gives a final stop 5 s, so + /// the driver stops waiting a little earlier; anything still running past it (a poll loop wedged in + /// a read) is cancelled already and finishes in the background. + /// + internal static readonly TimeSpan ShutdownBudget = TimeSpan.FromSeconds(4.5); + + // References that already logged their bad-read Warning in the current bad episode (#739 P10). + // Removed when the tag reads Good; cleared on (re-)initialise and shutdown. + private readonly ConcurrentDictionary _warnedRefs = new(StringComparer.Ordinal); // Per-parent-word RMW gate for BOOL-within-word writes: a single-bit write is a // read-modify-write of the parent word (TwinCAT's symbol table doesn't expose "Word.N"), // so concurrent bit-writers to the same word must serialise or they lose each other's // updates. Keyed by device + parent symbol. (Cannot guard against the PLC program writing - // the word between our read and write — inherent to RMW.) - // References that already logged their bad-read Warning in the current bad episode (#739 P10). - // Removed when the tag reads Good; cleared on (re-)initialise and shutdown. - private readonly ConcurrentDictionary _warnedRefs = new(StringComparer.Ordinal); - + // the word between our read and write — inherent to RMW.) Teardown only CLEARS it: a + // SemaphoreSlim owns no kernel handle unless AvailableWaitHandle is read, and disposing one an + // in-flight write still holds turns its Release into an ObjectDisposedException. private readonly ConcurrentDictionary _bitRmwLocks = new(StringComparer.OrdinalIgnoreCase); @@ -118,29 +148,40 @@ public sealed class TwinCATDriver : 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 + clients before - // the overwrite below. Deliberately NOT ShutdownAsync: that also drops _poll and the subscriptions. - await ReleaseDeviceStatesAsync().ConfigureAwait(false); + // allows it, and a reconnect retry can overlap a running attempt). The new generation is BUILT + // first and published by one swap; the previous device states (probe loops, clients, native + // registrations) are released right after the swap — never orphaned, and a retained poll loop + // never sees an empty table in between. Deliberately NOT ShutdownAsync: that also drops _poll and + // the subscriptions. + long generation; + lock (_lifecycleLock) + { + generation = ++_lifecycleGeneration; + _health = new DriverHealth(DriverState.Initializing, null, null); + } _warnedRefs.Clear(); - _health = new DriverHealth(DriverState.Initializing, null, null); + + var devices = NewDeviceTable(); + var committed = false; try { // Apply the supplied config generation. A blank or content-free document keeps the // constructor-seeded options — that path covers callers that have already // materialised options up front (the factory passes both, in agreement). + var options = _options; if (!string.IsNullOrWhiteSpace(driverConfigJson)) { var parsed = TwinCATDriverFactoryExtensions.ParseOptions(driverConfigJson, _driverInstanceId); if (parsed.Devices.Count > 0 || parsed.RawTags.Count > 0) - _options = parsed; + options = parsed; } - foreach (var device in _options.Devices) + foreach (var device in options.Devices) { var addr = TwinCATAmsAddress.TryParse(device.HostAddress) ?? throw new InvalidOperationException( $"TwinCAT device has invalid HostAddress '{device.HostAddress}' — expected 'ads://{{netId}}:{{port}}'."); - _devices[device.HostAddress] = new DeviceState(addr, device); + devices[device.HostAddress] = new DeviceState(addr, device); } // Build the RawPath → definition table from the authored raw tags. Each entry's TagConfig @@ -149,14 +190,15 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery // DeviceName is resolved to the target's live host (def.DeviceHostAddress) via ResolveDeviceHost // so the existing device routing (_devices keyed by HostAddress) is unchanged. A mapper miss is // skipped (logged), never thrown — a bad tag must not fail the whole driver init. - _skippedTags.BeginRebuild(_options.RawTags.Count); - foreach (var entry in _options.RawTags) + var tags = new Dictionary(StringComparer.Ordinal); + _skippedTags.BeginRebuild(options.RawTags.Count); + foreach (var entry in options.RawTags) { if (TwinCATTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var def)) - _tagsByRawPath[entry.RawPath] = def with + tags[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent, - DeviceHostAddress = ResolveDeviceHost(entry.DeviceName), + DeviceHostAddress = ResolveDeviceHost(options, entry.DeviceName), }; else { @@ -168,47 +210,113 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery } _skippedTags.Commit(); - if (_options.Probe.Enabled) + if (BeforeCommitForTest is { } pause) await pause().ConfigureAwait(false); + + ConcurrentDictionary previous; + lock (_lifecycleLock) { - foreach (var state in _devices.Values) + if (generation != _lifecycleGeneration) + throw new OperationCanceledException( + "TwinCAT InitializeAsync was superseded by a shutdown or a newer initialise while it ran."); + + _options = options; + _tagsByRawPath = tags; + previous = _devices; + _devices = devices; + committed = true; + // Probe loops start INSIDE the commit so a teardown that swaps this generation out right + // after always finds (and cancels) them. + if (options.Probe.Enabled) { - state.ProbeCts = new CancellationTokenSource(); - var ct = state.ProbeCts.Token; - // Store the task so ShutdownAsync can await it. - state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct); + foreach (var state in devices.Values) + { + state.ProbeCts = new CancellationTokenSource(); + var ct = state.ProbeCts.Token; + // Store the task so the teardown can join it. + state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct); + } + } + _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); + } + + // The replaced generation is now unreachable from the tables: release it (bounded). + await ReleaseDevicesAsync(previous.Values).ConfigureAwait(false); + } + catch (Exception ex) when (!committed) + { + // Nothing of this generation was published: release what was built. A superseded init leaves + // health to whoever overtook it. + await ReleaseDevicesAsync(devices.Values).ConfigureAwait(false); + if (ex is not OperationCanceledException) + { + lock (_lifecycleLock) + { + if (generation == _lifecycleGeneration) + _health = new DriverHealth(DriverState.Faulted, null, ex.Message); } } - _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); - } - catch (Exception ex) - { - _health = new DriverHealth(DriverState.Faulted, null, ex.Message); throw; } } /// - /// Cancels, awaits and disposes every existing device state's probe loop, client and gate, then - /// clears the device table. Device part of only. + /// Releases a set of device states that have already been swapped out of : + /// marks them released (so a probe wedged in the blocking ADS connect disposes the client it + /// eventually gets), disposes their native registrations (a re-initialise must not leave live ADS + /// notifications bound to a dead generation), cancels every probe loop and joins them all together + /// within , then disposes their clients. The connect gates are NOT + /// disposed — an in-flight read/write may still hold one. /// - private async Task ReleaseDeviceStatesAsync() + /// The released device states (each owned by exactly one caller). + private async Task ReleaseDevicesAsync(ICollection states) { - if (_devices.IsEmpty) return; - foreach (var state in _devices.Values) + if (states.Count == 0) return; + var probes = BeginRelease(states); + if (probes.Count > 0) { - try { state.ProbeCts?.Cancel(); } catch { } - if (state.ProbeTask is Task pt) - { - try { await pt.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } - catch { /* cancelled, faulted or wedged — teardown continues */ } - } - state.ProbeCts?.Dispose(); + try { await Task.WhenAll(probes).WaitAsync(ProbeJoinTimeout).ConfigureAwait(false); } + catch { /* cancelled, faulted or wedged in the synchronous ADS connect — teardown continues */ } + } + EndRelease(states); + } + + /// The non-blocking first half of a release: mark, retract registrations, cancel probes. + private static List BeginRelease(ICollection states) + { + var probes = new List(states.Count); + foreach (var state in states) + { + state.Released = true; + foreach (var reg in state.NativeRegistrations.Values) { try { reg.Dispose(); } catch { } } + try { state.ProbeCts?.Cancel(); } catch { /* already disposed */ } + if (state.ProbeTask is Task pt) probes.Add(pt); + } + return probes; + } + + /// The second half of a release, after the bounded probe join: dispose CTSs and clients. + private static void EndRelease(ICollection states) + { + foreach (var state in states) + { + try { state.ProbeCts?.Dispose(); } catch { } state.ProbeCts = null; state.ProbeTask = null; - state.DisposeClient(); - state.DisposeGate(); + try { state.DisposeClient(); } catch { /* best-effort */ } + } + } + + /// Swaps out every table under the lifecycle lock and returns the device set the caller now owns. + private ICollection TakeAllForTeardown() + { + lock (_lifecycleLock) + { + _lifecycleGeneration++; + var taken = _devices; + _devices = NewDeviceTable(); + _tagsByRawPath = EmptyTags; + return taken.Values; } - _devices.Clear(); } /// @@ -219,6 +327,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery } /// + /// + /// Safe to run concurrently with itself, with / and with + /// an abandoned : the tables are taken by one atomic swap, so each device + /// is released exactly once and an overtaken init releases its own generation. Bounded by + /// . + /// public async Task ShutdownAsync(CancellationToken cancellationToken) { // Native subs first — disposing the handles is cheap + lets the client close its @@ -227,30 +341,22 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } } _nativeSubs.Clear(); - await _poll.DisposeAsync().ConfigureAwait(false); + var taken = TakeAllForTeardown(); - // Cancel every probe loop and await its task before disposing the client + gate so the - // loop can never touch a disposed object. - foreach (var state in _devices.Values) + // The poll loops and the device release wind down in parallel (releasing a client can unblock a + // poll loop stuck in a read on it); the whole teardown is bounded. + var teardown = Task.WhenAll(_poll.DisposeAsync().AsTask(), ReleaseDevicesAsync(taken)); + try { await teardown.WaitAsync(ShutdownBudget).ConfigureAwait(false); } + catch (TimeoutException) { - try { state.ProbeCts?.Cancel(); } catch { } - if (state.ProbeTask is Task pt) - { - try { await pt.ConfigureAwait(false); } - catch (OperationCanceledException) { /* expected — probe loop exits on cancel */ } - catch { /* other probe errors are not fatal to shutdown */ } - } - state.ProbeCts?.Dispose(); - state.ProbeCts = null; - state.DisposeClient(); - state.DisposeGate(); + _logger.LogWarning( + "TwinCAT driver '{DriverInstanceId}' shutdown exceeded {Budget}; the remaining teardown continues in the background", + _driverInstanceId, ShutdownBudget); } - _devices.Clear(); - _tagsByRawPath.Clear(); + catch { /* best-effort teardown */ } + + _warnedRefs.Clear(); _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry - // Dispose + clear the per-parent-word RMW gates so ReinitializeAsync cycles don't orphan - // their SemaphoreSlim instances (each leaks a wait handle once contended). - foreach (var sem in _bitRmwLocks.Values) sem.Dispose(); _bitRmwLocks.Clear(); _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); } @@ -340,7 +446,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery catch (OperationCanceledException) { throw; } catch (Exception ex) { - MarkCommsFault(ex.Message); + MarkCommsFault(ex.Message, device); return new DataValueSnapshot(null, TwinCATStatusMapper.BadCommunicationError, null, now); } } @@ -463,7 +569,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery { foreach (var e in chunk) results[e.Index] = new DataValueSnapshot(null, TwinCATStatusMapper.BadCommunicationError, null, now); - MarkCommsFault(ex.Message); + MarkCommsFault(ex.Message, device); // Once per device: the health message is overwritten by the next good read in the same poll, so // without this a sum-read path failing on every poll left no trace (hw2 live gate, 2026-10-08). if (connected && !device.SumChunkFailureWarned) @@ -575,7 +681,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery catch (Exception ex) { results[i] = new WriteResult(TwinCATStatusMapper.BadCommunicationError); - MarkCommsFault(ex.Message); + MarkCommsFault(ex.Message, device); } } @@ -771,7 +877,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery // The device could not be connected (offline, router down, backoff window open). // Every tag of this device goes Bad + dead for the retry; other devices carry on. (unreachable ??= []).Add(device); - MarkCommsFault(unavailable.InnerException?.Message ?? unavailable.Message); + MarkCommsFault(unavailable.InnerException?.Message ?? unavailable.Message, device); RaiseDataChange(handle, reference, new DataValueSnapshot( null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow)); } @@ -783,7 +889,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery "TwinCAT driver '{DriverInstanceId}' native-notification registration failed for " + "'{Reference}'; it will be retried on the next healthy probe tick", _driverInstanceId, reference); - MarkCommsFault(ex.Message); + MarkCommsFault(ex.Message, device); RaiseDataChange(handle, reference, new DataValueSnapshot( null, TwinCATStatusMapper.BadCommunicationError, null, DateTime.UtcNow)); } @@ -801,6 +907,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery return handle; } + /// + /// The device state was released (re-initialise / teardown) while a connect on it ran. Not an + /// : the caller's token was not cancelled, so the data path maps + /// it like any failed request (Bad for this call) and ignores it. + /// + private sealed class DeviceReleasedException(string host) + : Exception($"TwinCAT device {host} was released during connect."); + /// The connect step of a native registration failed — the whole device is unreachable. private sealed class DeviceUnavailableException(Exception inner) : Exception("TwinCAT device unreachable: " + inner.Message, inner); @@ -809,8 +923,13 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery /// Records a comms fault: preserving LastSuccessfulRead. /// Never downgrades . /// - private void MarkCommsFault(string message) + /// The fault description. + /// The device the fault was observed on; a fault on a device already swapped out of + /// the table (a read/write racing a re-initialise or teardown) belongs to a dead generation and is + /// not recorded. + private void MarkCommsFault(string message, DeviceState? device = null) { + if (device?.Released == true) return; var current = _health; if (current.State != DriverState.Faulted) _health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, message); @@ -1006,8 +1125,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery BitIndex = bitIndex; ArrayLength = arrayLength; Interval = interval; - OnChange = onChange; - OnUnavailableStatus = onUnavailable; + // A notification already queued in the client's dispatch channel can be delivered after this + // registration is disposed (unsubscribe / teardown): drop it here, for every client. + OnChange = (symbol, value) => { if (!IsDisposed) onChange(symbol, value); }; + OnUnavailableStatus = status => { if (!IsDisposed) onUnavailable(status); }; } /// Installs a freshly-registered handle and returns the previous one (null on first set). @@ -1025,9 +1146,15 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery public ITwinCATNotificationHandle? MarkHandleDead() => Interlocked.Exchange(ref _handle, null); + private volatile bool _disposed; + + /// Gets a value indicating whether this registration has been disposed (its callbacks are dropped). + public bool IsDisposed => _disposed; + /// Removes this registration from its device and disposes its live handle. public void Dispose() { + _disposed = true; Device.NativeRegistrations.TryRemove(Id, out _); var h = Interlocked.Exchange(ref _handle, null); try { h?.Dispose(); } catch { /* best-effort */ } @@ -1085,6 +1212,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery { state.ConsecutiveProbeFailures = 0; RecordDeviceAnswered(DateTime.UtcNow, force: true); + if (probed is not null) ReportNotificationDrops(state, probed); } else if (++state.ConsecutiveProbeFailures >= ProbeFailuresBeforeDegraded) { @@ -1107,6 +1235,24 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery } } + /// + /// Logs a Warning when the client's native-notification queue dropped values since the last probe + /// tick (the queue is bounded and drops on overflow — silently, before #739 L8). At most one + /// Warning per probe interval, carrying the delta and the total. + /// + private void ReportNotificationDrops(DeviceState state, ITwinCATClient client) + { + var total = client.DroppedNotificationCount; + var delta = total - state.ReportedNotificationDrops; + if (delta <= 0) return; + state.ReportedNotificationDrops = total; + _logger.LogWarning( + "TwinCAT driver '{DriverInstanceId}' device '{HostAddress}' dropped {Dropped} native notification(s) " + + "since the last probe ({Total} on this connection): the notification queue was full — the consumer " + + "is not keeping up with the PLC's change rate", + _driverInstanceId, state.Options.HostAddress, delta, total); + } + // ---- IDriverDiagnostics ---- // Well-known system symbols, exported by TC3 (and most TC2) PLC runtimes under TwinCAT_SystemInfoVarList. @@ -1296,11 +1442,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery /// CLI, pass the host string as the DeviceName). An empty DeviceName routes to the sole device /// on a single-device instance; an unmatched name leaves the def unrouted (→ BadNodeIdUnknown). /// + /// The options generation being initialised. /// The owning device's name (or host) from the raw-tag entry. /// The resolved device host address, or empty when unresolved. - private string ResolveDeviceHost(string deviceName) + private static string ResolveDeviceHost(TwinCATDriverOptions options, string deviceName) { - var devices = _options.Devices; + var devices = options.Devices; if (string.IsNullOrEmpty(deviceName)) return devices.Count == 1 ? devices[0].HostAddress : ""; foreach (var d in devices) @@ -1372,6 +1519,8 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery { // Re-check under the gate: another caller may have connected while we waited. if (device.Client is { IsConnected: true } c) return c; + // A released device (swapped out by a re-initialise / teardown) never connects again. + if (device.Released) throw new DeviceReleasedException(device.Options.HostAddress); // Inside an open backoff window a data-path caller fails fast — no client // create, no wire connect. The probe loop bypasses this (its interval is the recovery @@ -1412,7 +1561,16 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery device.Backoff.RecordFailure(DateTime.UtcNow); // open the backoff window throw; } + if (device.Released) + { + // This device was torn down while the (blocking) connect ran: never install a client on a + // released generation, where nothing would ever dispose it. + client.OnSymbolVersionChanged -= HandleSymbolVersionChanged; + try { client.Dispose(); } catch { /* best-effort */ } + throw new DeviceReleasedException(device.Options.HostAddress); + } device.Client = client; + device.ReportedNotificationDrops = 0; device.UnavailableDiagSymbols.Clear(); // a rebuilt client may face a PLC that now exports them device.Backoff.RecordSuccess(); // recovery resets the window immediately @@ -1608,29 +1766,17 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } } _nativeSubs.Clear(); - // PollGroupEngine.DisposeAsync awaits loop tasks; we drive that synchronously here - // (bounded wait — same 5s ceiling DisposeAsync uses internally) using Wait() on the - // returned ValueTask so no sync-context capture happens. - try { _poll.DisposeAsync().AsTask().Wait(TimeSpan.FromSeconds(5)); } catch { } + var taken = TakeAllForTeardown(); + var probes = BeginRelease(taken); - foreach (var state in _devices.Values) - { - try { state.ProbeCts?.Cancel(); } catch { } - if (state.ProbeTask is Task pt) - { - try { pt.Wait(TimeSpan.FromSeconds(2)); } catch { /* probe-cancel races are expected */ } - } - state.ProbeCts?.Dispose(); - state.ProbeCts = null; - state.DisposeClient(); - state.DisposeGate(); - } - _devices.Clear(); - _tagsByRawPath.Clear(); + // PollGroupEngine.DisposeAsync awaits loop tasks; drive it synchronously with a bounded Wait() on + // the returned task (no sync-context capture), in parallel with the probe join. + var poll = _poll.DisposeAsync().AsTask(); + try { Task.WhenAll([poll, .. probes]).Wait(ProbeJoinTimeout); } catch { /* cancel races are expected */ } + EndRelease(taken); + + _warnedRefs.Clear(); _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry - // Dispose + clear the per-parent-word RMW gates (mirrors ShutdownAsync) so the - // SemaphoreSlim instances aren't orphaned on disposal. - foreach (var sem in _bitRmwLocks.Values) sem.Dispose(); _bitRmwLocks.Clear(); _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); } @@ -1711,7 +1857,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery UnavailableDiagSymbols.Clear(); } - /// Disposes the connection gate semaphore. - public void DisposeGate() => ConnectGate.Dispose(); + /// + /// Set once this device state has been swapped out of the driver's table. A connect that + /// completes afterwards (a probe wedged in the blocking ADS Connect past the bounded join) + /// disposes its client instead of installing it on a dead generation. + /// + public volatile bool Released; + + /// Total notification drops last reported for this device's current client. + public long ReportedNotificationDrops; } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs index 39f795be..909fc31f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs @@ -56,10 +56,22 @@ internal class FakeTwinCATClient : ITwinCATClient { ConnectCount++; if (ThrowOnConnect) throw Exception ?? new InvalidOperationException(); + if (BlockConnect is { } block) + { + // Models the real client's blocking SDK Connect: ignores the token until released. + return block.Task.ContinueWith(_ => { IsConnected = true; }, TaskScheduler.Default); + } IsConnected = true; return Task.CompletedTask; } + /// When set, does not complete (and ignores cancellation) until this + /// is completed — a probe wedged in the synchronous ADS Connect. + public TaskCompletionSource? BlockConnect { get; set; } + + /// Gets or sets the simulated notification-queue drop count. + public long DroppedNotificationCount { get; set; } + /// Simulates reading a value from a symbol. /// The path to the symbol to read. /// The data type of the symbol. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATLifecycleTests.cs new file mode 100644 index 00000000..c419289b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATLifecycleTests.cs @@ -0,0 +1,271 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT; + +namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests; + +/// +/// #739 L8 + the Lane A runtime contract (final stop = ShutdownAsync then DisposeAsync, +/// bounded 5 s + 250 ms): a probe wedged in the blocking ADS connect cannot hang teardown; dispose may run +/// while a shutdown is still running; an abandoned initialise overlapping the next shutdown never installs +/// its generation; a value queued before unsubscribe is not delivered after it; re-init swaps its tables +/// and retracts the previous generation's notifications; queue drops are reported. +/// +[Trait("Category", "Unit")] +public sealed class TwinCATLifecycleTests +{ + private const string Host = "ads://5.23.91.23.1.1:851"; + + private static TwinCATDriver NewDriver(FakeTwinCATClientFactory factory, bool probe = true, bool native = true, + TwinCATCapturingLogger? log = null, params TwinCATTagDefinition[] tags) => + new(new TwinCATDriverOptions + { + Devices = [new TwinCATDeviceOptions(Host)], + RawTags = TwinCATRawTags.Entries(tags.Length > 0 + ? tags + : [new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt)]), + Probe = probe + ? new TwinCATProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(30), Timeout = TimeSpan.FromMilliseconds(100) } + : new TwinCATProbeOptions { Enabled = false }, + UseNativeNotifications = native, + }, "drv-1", factory, log); + + private static async Task WaitForAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition() && DateTime.UtcNow < deadline) + await Task.Delay(10); + } + + /// + /// A probe wedged in the (uncancellable) connect no longer hangs ShutdownAsync — the join is bounded — + /// and the client the wedged connect eventually produces is disposed, not installed on a dead device. + /// + [Fact] + public async Task Shutdown_is_bounded_when_a_probe_is_wedged_in_connect() + { + var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var factory = new FakeTwinCATClientFactory { Customise = () => new FakeTwinCATClient { BlockConnect = wedge } }; + var drv = NewDriver(factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2)); + factory.Clients.Count.ShouldBe(1); // the probe is inside the wedged connect + + var sw = Stopwatch.StartNew(); + await drv.ShutdownAsync(TestContext.Current.CancellationToken).WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + sw.Elapsed.ShouldBeLessThan(TwinCATDriver.ShutdownBudget + TimeSpan.FromMilliseconds(500)); + + wedge.SetResult(); + await WaitForAsync(() => factory.Clients[0].DisposeCount == 1, TimeSpan.FromSeconds(2)); + factory.Clients[0].DisposeCount.ShouldBe(1); + drv.DeviceCount.ShouldBe(0); + } + + /// + /// DisposeAsync while a ShutdownAsync is still running (joining a wedged probe) returns promptly and + /// both complete without throwing; the device is released exactly once. + /// + [Fact] + public async Task Dispose_concurrent_with_a_running_shutdown_completes_promptly() + { + var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var factory = new FakeTwinCATClientFactory { Customise = () => new FakeTwinCATClient { BlockConnect = wedge } }; + var drv = NewDriver(factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2)); + + var shutdown = drv.ShutdownAsync(CancellationToken.None); + var sw = Stopwatch.StartNew(); + await drv.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(250)); + shutdown.IsCompleted.ShouldBeFalse("the positive control: the first shutdown really was still joining the probe"); + + wedge.SetResult(); + await shutdown.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + drv.Dispose(); // and a late sync Dispose is a no-op + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + } + + /// + /// An InitializeAsync overtaken by a ShutdownAsync before it commits must not install its devices or + /// start probe loops when it resumes. Positive control: a fresh initialise afterwards installs normally. + /// + [Fact] + public async Task An_initialise_overtaken_by_shutdown_before_commit_installs_nothing() + { + var factory = new FakeTwinCATClientFactory(); + var drv = NewDriver(factory); + var paused = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var resume = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + drv.BeforeCommitForTest = async () => { paused.TrySetResult(); await resume.Task; }; + + var abandoned = drv.InitializeAsync("{}", CancellationToken.None); + await paused.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + resume.SetResult(); + + await Should.ThrowAsync(() => abandoned.WaitAsync(TimeSpan.FromSeconds(5))); + drv.DeviceCount.ShouldBe(0); + await Task.Delay(150, TestContext.Current.CancellationToken); // several probe intervals + factory.Clients.ShouldBeEmpty("no probe loop of the overtaken generation may be running"); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + + drv.BeforeCommitForTest = null; + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + drv.DeviceCount.ShouldBe(1); + await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2)); + factory.Clients.Count.ShouldBe(1); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + } + + /// + /// The Lane A shape: an InitializeAsync the runtime abandoned while it is still releasing the previous + /// generation (joining a probe wedged in connect) overlaps the next ShutdownAsync. Shutdown stays + /// bounded and, once everything settles, no device, probe loop or live client survives. + /// + [Fact] + public async Task An_abandoned_initialise_overlapping_shutdown_leaves_nothing_running() + { + var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var first = true; + var factory = new FakeTwinCATClientFactory + { + Customise = () => + { + var c = new FakeTwinCATClient { BlockConnect = first ? wedge : null }; + first = false; + return c; + }, + }; + var drv = NewDriver(factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2)); + + var abandoned = drv.InitializeAsync("{}", CancellationToken.None); // joins the wedged generation-1 probe + await Task.Delay(50, TestContext.Current.CancellationToken); + var sw = Stopwatch.StartNew(); + await drv.ShutdownAsync(TestContext.Current.CancellationToken).WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken); + sw.Elapsed.ShouldBeLessThan(TwinCATDriver.ShutdownBudget + TimeSpan.FromMilliseconds(500)); + + wedge.SetResult(); + try { await abandoned.WaitAsync(TimeSpan.FromSeconds(5)); } catch (OperationCanceledException) { } + + drv.DeviceCount.ShouldBe(0); + await WaitForAsync(() => factory.Clients.All(c => c.DisposeCount > 0 || !c.IsConnected), TimeSpan.FromSeconds(2)); + var settled = factory.Clients.Count; + await Task.Delay(150, TestContext.Current.CancellationToken); + factory.Clients.Count.ShouldBe(settled, "no probe loop may still be creating clients"); + factory.Clients.ShouldAllBe(c => c.DisposeCount > 0 || !c.IsConnected); + } + + /// + /// A value the client queued before the unsubscribe is dropped if dispatched after it. + /// Positive control: the same callback delivers while subscribed. + /// + [Fact] + public async Task A_notification_dispatched_after_unsubscribe_is_dropped() + { + var factory = new FakeTwinCATClientFactory(); + var drv = NewDriver(factory, probe: false); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + var events = new ConcurrentQueue(); + drv.OnDataChange += (_, e) => events.Enqueue(e); + var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + var callback = factory.Clients[0].Notifications.Single().OnChange; // what the client's queue holds + + callback("MAIN.Speed", 1); + events.Count.ShouldBe(1); + + await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken); + callback("MAIN.Speed", 2); + events.Count.ShouldBe(1); + } + + /// + /// A direct re-initialise with a changed config swaps the tag table (the removed tag is unknown, the + /// added one resolves) and retracts the previous generation's live notifications. + /// + [Fact] + public async Task Reinitialise_swaps_tables_and_retracts_the_previous_generations_notifications() + { + var factory = new FakeTwinCATClientFactory(); + var drv = NewDriver(factory, probe: false); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + _ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); + factory.Clients[0].Notifications.Count.ShouldBe(1); + + var next = System.Text.Json.JsonSerializer.Serialize(new + { + devices = new[] { new { hostAddress = Host } }, + rawTags = new[] + { + new + { + rawPath = "Torque", + tagConfig = TwinCATTagDefinitionFactory.ToTagConfig( + new TwinCATTagDefinition("Torque", Host, "MAIN.Torque", TwinCATDataType.DInt)), + writeIdempotent = false, + deviceName = Host, + }, + }, + probe = new { enabled = false }, + }); + await drv.InitializeAsync(next, TestContext.Current.CancellationToken); + + factory.Clients[0].Notifications.ShouldBeEmpty(); + factory.Clients[0].DisposeCount.ShouldBe(1); + (await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken)).Single().StatusCode + .ShouldBe(TwinCATStatusMapper.BadNodeIdUnknown); + (await drv.ReadAsync(["Torque"], TestContext.Current.CancellationToken)).Single().StatusCode + .ShouldBe(TwinCATStatusMapper.Good); + } + + /// Reads racing a stream of re-initialises never throw (no table is ever mid-Clear). + [Fact] + public async Task Reads_racing_reinitialise_never_throw() + { + var factory = new FakeTwinCATClientFactory(); + var drv = NewDriver(factory, probe: false); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + using var stop = new CancellationTokenSource(TimeSpan.FromMilliseconds(400)); + var good = 0; + var reader = Task.Run(async () => + { + while (!stop.IsCancellationRequested) + { + var r = await drv.ReadAsync(["Speed"], CancellationToken.None); + if (r[0].StatusCode == TwinCATStatusMapper.Good) Interlocked.Increment(ref good); + } + }, TestContext.Current.CancellationToken); + while (!stop.IsCancellationRequested) + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await reader; // throws if any read threw + good.ShouldBeGreaterThan(0); + } + + /// Notification-queue drops are reported once per probe tick that sees new drops. + [Fact] + public async Task Notification_queue_drops_are_reported() + { + var fake = new FakeTwinCATClient(); + var log = new TwinCATCapturingLogger(); + var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, log: log); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await WaitForAsync(() => fake.IsConnected, TimeSpan.FromSeconds(2)); + await Task.Delay(100, TestContext.Current.CancellationToken); + log.Warnings.Count(w => w.Contains("dropped")).ShouldBe(0); // no drops, no warning + + fake.DroppedNotificationCount = 5; + await WaitForAsync(() => log.Warnings.Any(w => w.Contains("dropped 5")), TimeSpan.FromSeconds(2)); + await Task.Delay(100, TestContext.Current.CancellationToken); + log.Warnings.Count(w => w.Contains("dropped")).ShouldBe(1); + + fake.DroppedNotificationCount = 8; + await WaitForAsync(() => log.Warnings.Any(w => w.Contains("dropped 3")), TimeSpan.FromSeconds(2)); + log.Warnings.Count(w => w.Contains("dropped")).ShouldBe(2); + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + } +}