dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(abcip): a reader racing a re-init neither leaks a runtime nor degrades

Browse Source 
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
This commit is contained in:
Joseph Doherty
2026-10-22 14:17:16 -04:00
parent 22d4680e7b
commit 48b0173ae9
2 changed files with 111 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs
+84 -21
View File
@@ -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;
}
/// <summary>Throws when a teardown has retired <paramref name="device"/> (a reader raced a re-init / shutdown).</summary>
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.");
}
/// <summary>
@@ -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);
/// <summary>Disposes all runtime tag handles and clears the caches.</summary>
private volatile bool _retired;
/// <summary>
/// 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.
/// </summary>
public bool Retired => _retired;
/// <summary>Marks the state retired, then disposes every runtime handle (see <see cref="DisposeHandles"/>).</summary>
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();
}
/// <summary>
/// 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 <see cref="ConnectGate"/>: an in-flight reader of a retired state may
/// still release one, and a <see cref="SemaphoreSlim"/> only awaited asynchronously holds no wait handle.
/// </summary>
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();
}
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs
+27 -4
View File
@@ -123,14 +123,32 @@ public sealed class AbCipLifecycleConcurrencyTests
/// <summary>
/// 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.
/// </summary>
[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<FakeAbCipTag>();
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");
}
/// <summary>F10: the driver exposes only an async dispose; a sync <c>Dispose()</c> that blocked on the async