From 48b0173ae95d5a0094cd1fedc79d3088ff244fa8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:17:16 -0400 Subject: [PATCH] fix(abcip): a reader racing a re-init neither leaks a runtime nor degrades Found by the H13 stress test (200 re-inits under a concurrent reader): a read that resolved the device just before the table swap could cold-create a runtime into the retired state after its teardown had disposed it (leaked native handle / lease), and could hit a per-tag SemaphoreSlim the teardown had disposed, which the generic catch reported as a comms fault (Degraded). - DeviceState.Retire marks the state retired before disposing handles; each handle is removed-then-disposed exactly once. A creator re-checks the flag after its TryAdd (fenced, Dekker-paired with Retire) and disposes the orphan. - A failure on a retired device is the teardown's doing: BadCommunicationError for that one call, no evict, no health signal. - Per-tag/RMW gates are no longer disposed (async-only waits hold no handle), matching the ConnectGate rationale. Refs #737 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../AbCipDriver.cs | 105 ++++++++++++++---- .../AbCipLifecycleConcurrencyTests.cs | 31 +++++- 2 files changed, 111 insertions(+), 25 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs index c9c2f669..e4b6f9bb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -561,7 +561,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, { // Never-published states hold no probe loop and no runtime yet; dispose for symmetry. if (builtDevices is not null) - foreach (var state in builtDevices.Values) state.DisposeHandles(); + foreach (var state in builtDevices.Values) state.Retire(); if (ex is OperationCanceledException) { _logger.LogInformation("AbCip driver {DriverInstanceId} initialization abandoned: {Reason}", @@ -631,7 +631,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, state.ProbeCts?.Dispose(); state.ProbeCts = null; state.ProbeTask = null; - state.DisposeHandles(); + state.Retire(); } return generation; } @@ -1155,6 +1155,13 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, WarnRefusalOnce(def.Name, "read", reference, def.DeviceHostAddress, ate.Message); return ReadOutcome.Refused; } + catch (Exception) when (device.Retired) + { + // A re-init / shutdown retired this device under the read: its handles were disposed by the teardown, not + // lost on the wire — no evict (the teardown owns them), no health signal. + results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadCommunicationError, null, now); + return ReadOutcome.None; + } catch (Exception ex) { // Transport exception — evict so the next read creates a fresh handle. An AbTransportException @@ -1271,6 +1278,11 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, WarnRefusalOnce(parent.Name, "whole-UDT read", group.ParentName, parent.DeviceHostAddress, ate.Message); return ReadOutcome.Refused; } + catch (Exception) when (device.Retired) + { + StampGroupStatus(group, results, now, AbCipStatusMapper.BadCommunicationError); + return ReadOutcome.None; + } catch (Exception ex) { EvictRuntime(device, parent.Name); @@ -1492,6 +1504,11 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, results[i] = new WriteResult(AbCipStatusMapper.MapStatus(ate.Status)); WarnRefusalOnce(def.Name, "write", w.FullReference, def.DeviceHostAddress, ate.Message); } + catch (Exception) when (device.Retired) + { + // The device was retired under this write by a re-init / shutdown — not a comms fault. + results[i] = new WriteResult(AbCipStatusMapper.BadCommunicationError); + } catch (Exception ex) { // Transport / wire error — evict so the next write creates a fresh handle. @@ -1554,6 +1571,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, DeviceState device, string parentTagName, CancellationToken ct) { if (device.ParentRuntimes.TryGetValue(parentTagName, out var existing)) return existing; + ThrowIfRetired(device); var runtime = _tagFactory.Create(device.BuildCreateParams(parentTagName, _options.Timeout)); try @@ -1567,10 +1585,25 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } // Two concurrent callers can both miss the cache + both initialize a runtime; only the // first TryAdd wins. Dispose the loser so it doesn't leak a native tag handle. - if (device.ParentRuntimes.TryAdd(parentTagName, runtime)) - return runtime; - runtime.Dispose(); - return device.ParentRuntimes[parentTagName]; + if (!device.ParentRuntimes.TryAdd(parentTagName, runtime)) + { + runtime.Dispose(); + return device.ParentRuntimes[parentTagName]; + } + Interlocked.MemoryBarrier(); // pairs with DeviceState.Retire + if (device.Retired) + { + if (device.ParentRuntimes.TryRemove(parentTagName, out var orphan)) orphan.Dispose(); + ThrowIfRetired(device); + } + return runtime; + } + + /// Throws when a teardown has retired (a reader raced a re-init / shutdown). + private static void ThrowIfRetired(DeviceState device) + { + if (device.Retired) + throw new ObjectDisposedException(nameof(DeviceState), $"AbCip device {device.Options.HostAddress} was retired by a re-initialization or shutdown."); } /// @@ -1607,6 +1640,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Re-check under the gate: a sibling cold tag may have connected (or a concurrent write // may have materialised this very handle) while this caller waited. if (device.Runtimes.TryGetValue(def.Name, out var raced)) return raced; + ThrowIfRetired(device); // Inside an open backoff window a data-path caller fails fast — no runtime // create, no Forward-Open. The malformed-path config error above still surfaces (a permanent @@ -1662,10 +1696,20 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // The gate makes a cold create single-flight, but a concurrent EvictRuntime (a failed read // on this handle) can still land between the re-check and the add; only the first TryAdd // wins. Dispose the loser so it doesn't leak a native tag handle. - if (device.Runtimes.TryAdd(def.Name, runtime)) - return runtime; - runtime.Dispose(); - return device.Runtimes[def.Name]; + if (!device.Runtimes.TryAdd(def.Name, runtime)) + { + runtime.Dispose(); + return device.Runtimes[def.Name]; + } + // A teardown that retired this state while we created would never see this handle: take it back out + // (TryRemove — exactly one of us disposes it) rather than leak it into a dead table. + Interlocked.MemoryBarrier(); // pairs with DeviceState.Retire + if (device.Retired) + { + if (device.Runtimes.TryRemove(def.Name, out var orphan)) orphan.Dispose(); + ThrowIfRetired(device); + } + return runtime; } finally { @@ -2230,19 +2274,38 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, AuthoredType: authoredType, UseConnectedMessaging: Profile.SupportsConnectedMessaging); - /// Disposes all runtime tag handles and clears the caches. + private volatile bool _retired; + + /// + /// True once a teardown retired this state. A reader that resolved the device just before the swap can + /// still be running against it: its failures are the teardown's doing, not the device's (no Degraded), and a + /// runtime it cold-creates afterwards is disposed by the creator instead of leaking into a dead table. + /// + public bool Retired => _retired; + + /// Marks the state retired, then disposes every runtime handle (see ). + public void Retire() + { + _retired = true; + // Dekker pairing with the creator's post-add check: the flag is globally visible before the tables are + // enumerated, and the creator fences between its add and its read of the flag — so at least one of the two + // sees the other's write and the handle is disposed. + Interlocked.MemoryBarrier(); + DisposeHandles(); + } + + /// + /// Disposes every runtime tag handle, each exactly once (removed before it is disposed, so a racing + /// creator's own cleanup cannot dispose it a second time). The per-runtime / RMW gates are deliberately NOT + /// disposed, for the same reason as : an in-flight reader of a retired state may + /// still release one, and a only awaited asynchronously holds no wait handle. + /// public void DisposeHandles() { - foreach (var r in Runtimes.Values) r.Dispose(); - Runtimes.Clear(); - foreach (var r in ParentRuntimes.Values) r.Dispose(); - ParentRuntimes.Clear(); - // Dispose + clear the per-runtime/RMW gates so ReinitializeAsync cycles don't orphan - // their SemaphoreSlim instances (each leaks a wait handle once contended). - foreach (var sem in _runtimeLocks.Values) sem.Dispose(); - _runtimeLocks.Clear(); - foreach (var sem in _rmwLocks.Values) sem.Dispose(); - _rmwLocks.Clear(); + foreach (var key in Runtimes.Keys) + if (Runtimes.TryRemove(key, out var r)) r.Dispose(); + foreach (var key in ParentRuntimes.Keys) + if (ParentRuntimes.TryRemove(key, out var r)) r.Dispose(); } } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs index 00cb921a..eea409d2 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs @@ -123,14 +123,32 @@ public sealed class AbCipLifecycleConcurrencyTests /// /// Readers of the device/routing tables (the retained poll engine, the actor's host-status pull) never - /// see a table mid-mutation while the driver re-initialises underneath them. + /// see a table mid-mutation while the driver re-initialises underneath them, never trip over a per-tag lock a + /// retire disposed, and never leak a runtime they cold-created into a device state a retire already tore down. /// [Fact] - public async Task Readers_survive_repeated_reinitialisation() + public async Task Readers_survive_repeated_reinitialisation_and_leak_no_runtime() { - var (drv, _) = Build(probeReadDelay: TimeSpan.Zero); + var created = new System.Collections.Concurrent.ConcurrentBag(); + var factory = new FakeAbCipTagFactory + { + Customise = p => + { + var t = new FakeAbCipTag(p) { Value = 1 }; + created.Add(t); + return t; + }, + }; + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions(Host)], + RawTags = AbCipRawTags.From(new AbCipTagDefinition("Speed", Host, "Motor1.Speed", AbCipDataType.DInt)), + Probe = new AbCipProbeOptions { Enabled = false }, + }, "abcip-lifecycle", factory); await drv.InitializeAsync("{}", CancellationToken.None); using var stop = new CancellationTokenSource(); + var started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var badHealth = 0; var reader = Task.Run(async () => { var reads = 0; @@ -139,17 +157,22 @@ public sealed class AbCipLifecycleConcurrencyTests _ = drv.GetHostStatuses(); _ = drv.ResolveHost("Speed"); await drv.ReadAsync(["Speed"], CancellationToken.None); + if (drv.GetHealth().State == DriverState.Degraded) Interlocked.Increment(ref badHealth); reads++; + started.TrySetResult(); } return reads; }); + await started.Task.WaitAsync(TimeSpan.FromSeconds(5)); - for (var i = 0; i < 50; i++) + for (var i = 0; i < 200; i++) await drv.InitializeAsync("{}", CancellationToken.None); stop.Cancel(); (await reader).ShouldBeGreaterThan(0); // positive control: the reader actually ran + Volatile.Read(ref badHealth).ShouldBe(0, "a re-init is not a comms fault"); await drv.DisposeAsync(); + created.Where(t => !t.Disposed).ShouldBeEmpty("every runtime a reader created must be disposed by some teardown"); } /// F10: the driver exposes only an async dispose; a sync Dispose() that blocked on the async