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

fix(focas): build-and-swap the device table; make the lifecycle overlap-safe (R07:H13)

Browse Source 
_devices was a plain Dictionary cleared and refilled in place by InitializeAsync /
ShutdownAsync while the retained poll engine, the fixed-tree and alarm loops and the
actor's health tick enumerated it ("Collection was modified"). It is now built whole and
published by reference swap, and a release swaps it out to empty before tearing it down.

Lane A's runtime overlaps lifecycle calls, so the driver now tolerates both shapes:
- a lifecycle generation is bumped by every Initialize/Shutdown; an initialise that finds
  itself superseded (abandoned by the runtime, then a shutdown ran) publishes nothing,
  starts no loops, writes no health and throws OperationCanceledException;
- a shutdown overlapping another (dispose beside a still-running shutdown) waits for the
  first's teardown via a release tail instead of returning on an empty table;
- released devices are Retired: EnsureConnectedAsync refuses to reconnect them, and a
  connect that completes after retirement disposes what it opened. Clients are disposed
  under the device ConnectGate (bounded).

Refs #740
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:03:08 -04:00
parent 8a4cf8b823
commit a32894e93b
3 changed files with 424 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs
+223 -111
View File
@@ -43,7 +43,40 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
private readonly Func<string, FocasDriverBinding>? _rebind;
private readonly PollGroupEngine _poll;
private readonly ILogger<FocasDriver> _logger;
private readonly Dictionary<string, DeviceState> _devices = new(StringComparer.OrdinalIgnoreCase);
/// <summary>
/// Host address → device state. Built whole in <see cref="InitializeAsync"/> and SWAPPED in (and swapped
/// out whole by a release), never cleared or refilled in place: the retained poll engine, the fixed-tree
/// and alarm loops and the actor's health tick all read it concurrently with a re-initialise, and a
/// <see cref="Dictionary{TKey,TValue}"/> mutated under an enumerator throws "Collection was modified".
/// Readers take one <c>Volatile.Read</c> snapshot and use it for the whole operation.
/// </summary>
private IReadOnlyDictionary<string, DeviceState> _devices = EmptyDevices;
private static readonly IReadOnlyDictionary<string, DeviceState> EmptyDevices =
new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
/// <summary>The live device table (one consistent snapshot).</summary>
private IReadOnlyDictionary<string, DeviceState> Devices => Volatile.Read(ref _devices);
/// <summary>
/// Guards the publish/retire of the device table against an overlapping lifecycle call. Held only for
/// synchronous swaps — never across an await, and nothing a loop runs ever takes it.
/// </summary>
private readonly object _lifecycleLock = new();
/// <summary>
/// Bumped by every <see cref="InitializeAsync"/> and <see cref="ShutdownAsync"/>. An initialise that finds
/// the counter moved on by the time it would publish has been superseded (the runtime abandoned it and
/// tore the driver down) and must start nothing.
/// </summary>
private int _lifecycleGeneration;
/// <summary>
/// Completes when every device table released so far has finished its teardown. A shutdown that overlaps
/// another (Lane A's dispose beside a still-running shutdown) awaits it, so it never returns while the
/// first one is still joining loops and disposing clients.
/// </summary>
private Task _releaseTail = Task.CompletedTask;
// v3: RawPath → definition, case-sensitive ordinal. Built in InitializeAsync from _options.RawTags
// via the pure FocasTagDefinitionFactory mapper (the reference is always a RawPath).
// Built whole in InitializeAsync and SWAPPED in: the retained poll engine keeps calling
@@ -196,9 +229,14 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
// release the previous device states' loops + clients BEFORE they are overwritten (a live driver can be
// re-initialised without a teardown first). Not ShutdownAsync: that also drops _poll
// and subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init).
//
// The generation is taken FIRST: an initialise the runtime abandons (it timed out and moved on to a
// shutdown) is superseded by that shutdown's bump, and checks for it before publishing anything.
var generation = Interlocked.Increment(ref _lifecycleGeneration);
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
ResetAlarmHealth();
Volatile.Write(ref _health, new DriverHealth(DriverState.Initializing, null, null));
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Initializing, null, null));
var devices = new Dictionary<string, DeviceState>(StringComparer.OrdinalIgnoreCase);
try
{
// adopt a changed config. Options and the backend client factory are re-derived by ONE
@@ -222,7 +260,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
var addr = FocasHostAddress.TryParse(device.HostAddress)
?? throw new InvalidOperationException(
$"FOCAS device has invalid HostAddress '{device.HostAddress}' — expected 'focas://{{ip}}[:{{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
// blob is mapped by the pure factory; the entry's WriteIdempotent flag + DeviceName are
@@ -256,7 +294,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
throw new InvalidOperationException(
$"FOCAS tag '{entry.RawPath}' references device '{entry.DeviceName}' " +
$"which is not in the Devices list. Check for a typo (e.g. wrong port or hostname).");
if (!_devices.TryGetValue(deviceHost, out var device))
if (!devices.TryGetValue(deviceHost, out var device))
throw new InvalidOperationException(
$"FOCAS tag '{entry.RawPath}' references device '{entry.DeviceName}' " +
$"which is not in the Devices list. Check for a typo (e.g. wrong port or hostname).");
@@ -309,75 +347,135 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
// above; reusing the parsed record avoids per-tick allocs on subscription pollers.
parsedAddresses[entry.RawPath] = parsed;
}
// Publish the skip tally only once the whole table is built. An unknown-device throw above
// leaves the previous snapshot standing rather than a truncated one that would read as
// "we now skip fewer tags".
_skippedTags.Commit();
Volatile.Write(ref _parsedAddressesByTagName, parsedAddresses);
Volatile.Write(ref _tagsByRawPath, tagsByRawPath);
Volatile.Write(ref _alarmTags, alarmTags);
if (_options.Probe.Enabled)
// Publish everything at once, and only if no later lifecycle call has superseded this one. The check
// and the publish share the lock with the shutdown's swap-out, so either the shutdown sees (and
// releases) this table, or this initialise sees the shutdown and publishes nothing.
lock (_lifecycleLock)
{
foreach (var state in _devices.Values)
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
}
}
if (generation != Volatile.Read(ref _lifecycleGeneration))
throw new OperationCanceledException(
$"FOCAS initialise of {_driverInstanceId} was superseded by a later shutdown/initialise; nothing was started.");
if (_options.HandleRecycle.Enabled)
{
foreach (var state in _devices.Values)
{
state.RecycleCts = new CancellationTokenSource();
var ct = state.RecycleCts.Token;
state.RecycleTask = Task.Run(() => RecycleLoopAsync(state, ct), ct);
}
}
// See _fixedTreeActive: an authored FIXED: tag arms the loop on its own, because the
// config flag defaults off and the operator who committed those tags has already said
// what they want.
_fixedTreeActive = _options.FixedTree.Enabled
|| _parsedAddressesByTagName.Values.Any(a => a.Kind == FocasAreaKind.Fixed);
if (_fixedTreeActive)
{
foreach (var state in _devices.Values)
{
state.FixedTreeCts = new CancellationTokenSource();
var ct = state.FixedTreeCts.Token;
state.FixedTreeTask = Task.Run(() => FixedTreeLoopAsync(state, ct), ct);
}
// Publish the skip tally only once the whole table is built. An unknown-device throw above
// leaves the previous snapshot standing rather than a truncated one that would read as
// "we now skip fewer tags".
_skippedTags.Commit();
Volatile.Write(ref _parsedAddressesByTagName, parsedAddresses);
Volatile.Write(ref _tagsByRawPath, tagsByRawPath);
Volatile.Write(ref _alarmTags, alarmTags);
// See _fixedTreeActive: an authored FIXED: tag arms the loop on its own, because the
// config flag defaults off and the operator who committed those tags has already said
// what they want.
_fixedTreeActive = _options.FixedTree.Enabled
|| parsedAddresses.Values.Any(a => a.Kind == FocasAreaKind.Fixed);
Volatile.Write(ref _devices, devices);
StartDeviceLoops(devices.Values);
}
// Restart the alarm poll a re-initialise paused, against the tables just rebuilt. A
// driver with no subscription yet starts nothing — the actor subscribes once it is Connected.
await _alarmProjection.ResumeAsync().ConfigureAwait(false);
// A shutdown that landed while the resume ran owns the projection now: put it back the way it left it.
if (generation != Volatile.Read(ref _lifecycleGeneration))
{
await _alarmProjection.PauseAsync().ConfigureAwait(false);
throw new OperationCanceledException(
$"FOCAS initialise of {_driverInstanceId} was superseded while resuming the alarm poll.");
}
Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
}
catch (OperationCanceledException) when (generation != Volatile.Read(ref _lifecycleGeneration))
{
// Superseded: whatever this attempt built but never published is released here (a published table
// is the superseding call's to release); health belongs to that call too.
if (!ReferenceEquals(Devices, devices)) await ReleaseTableAsync(devices.Values).ConfigureAwait(false);
throw;
}
catch (Exception ex)
{
Volatile.Write(ref _health, new DriverHealth(DriverState.Faulted, null, ex.Message));
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Faulted, null, ex.Message));
throw;
}
}
/// <summary>Writes <paramref name="health"/> unless a later lifecycle call has taken over.</summary>
private void WriteHealthIfCurrent(int generation, DriverHealth health)
{
lock (_lifecycleLock)
{
if (generation == Volatile.Read(ref _lifecycleGeneration)) Volatile.Write(ref _health, health);
}
}
/// <summary>Starts the probe / recycle / fixed-tree loops for a freshly built table. Caller holds the lifecycle lock.</summary>
private void StartDeviceLoops(IEnumerable<DeviceState> states)
{
foreach (var state in states)
{
if (_options.Probe.Enabled)
{
state.ProbeCts = new CancellationTokenSource();
var ct = state.ProbeCts.Token;
state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct);
}
if (_options.HandleRecycle.Enabled)
{
state.RecycleCts = new CancellationTokenSource();
var ct = state.RecycleCts.Token;
state.RecycleTask = Task.Run(() => RecycleLoopAsync(state, ct), ct);
}
if (_fixedTreeActive)
{
state.FixedTreeCts = new CancellationTokenSource();
var ct = state.FixedTreeCts.Token;
state.FixedTreeTask = Task.Run(() => FixedTreeLoopAsync(state, ct), ct);
}
}
}
/// <summary>
/// Cancels, bounded-joins and disposes every existing device state (probe / recycle / fixed-tree
/// loops + client) and the alarm projection, then clears the table. Device part of
/// <see cref="ShutdownAsync"/> only (no poll / subscription teardown).
/// Swaps the live device table out (to empty) and releases it: cancels, bounded-joins and disposes every
/// device state's loops and client, after pausing the alarm projection. Device part of
/// <see cref="ShutdownAsync"/> only (no poll / subscription teardown). Also waits for any release an
/// overlapping lifecycle call started earlier, so it never returns while that one is still running.
/// </summary>
private async Task ReleaseDeviceStatesAsync()
{
// Pause, not dispose: the subscription and what the server was told survive the rebuild.
await _alarmProjection.PauseAsync().ConfigureAwait(false);
if (_devices.Count == 0) return;
var loops = new List<Task>();
foreach (var state in _devices.Values)
IReadOnlyDictionary<string, DeviceState> old;
Task prior;
var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
lock (_lifecycleLock)
{
old = _devices;
Volatile.Write(ref _devices, EmptyDevices);
prior = _releaseTail;
_releaseTail = done.Task;
}
try
{
if (old.Count > 0) await ReleaseTableAsync(old.Values).ConfigureAwait(false);
await prior.ConfigureAwait(false);
}
finally { done.TrySetResult(); }
}
/// <summary>
/// Retires and tears down one swapped-out table: retire every device first (so a connect still in flight
/// disposes what it opens), cancel every loop, join them (bounded), then dispose the CTSs and — under each
/// device's connect gate — its client.
/// </summary>
private async Task ReleaseTableAsync(IEnumerable<DeviceState> table)
{
var states = table.ToList();
var loops = new List<Task>();
foreach (var state in states)
{
state.Retire();
// Cancel-then-dispose can race in tight shutdown loops; swallowing is intentional
// but we log the cause so a noisy shutdown leaves a Debug trace.
try { state.ProbeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling probe CTS for {Host} failed", state.Options.HostAddress); }
try { state.RecycleCts?.Cancel(); }
@@ -388,15 +486,18 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
if (state.RecycleTask is { } rt) loops.Add(rt);
if (state.FixedTreeTask is { } ft) loops.Add(ft);
}
// Cancel every loop first, JOIN them (bounded), and only then dispose the CTSs and the
// client. Fire-and-forget loops used to outlive shutdown: one parked in EnsureConnectedAsync could
// finish its connect AFTER DisposeClient and leave a live socket on a shut-down driver.
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("FOCAS re-initialise: {Count} background loop(s) did not stop within {Timeout}; proceeding",
_logger.LogWarning("FOCAS teardown: {Count} background loop(s) did not stop within {Timeout}; proceeding",
loops.Count, LoopJoinTimeout);
}
foreach (var state in _devices.Values)
foreach (var state in states)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
@@ -407,9 +508,22 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
state.FixedTreeCts?.Dispose();
state.FixedTreeCts = null;
state.FixedTreeTask = null;
state.DisposeClient();
await DisposeClientGatedAsync(state).ConfigureAwait(false);
}
_devices.Clear();
}
/// <summary>
/// Disposes a device's client under its <see cref="DeviceState.ConnectGate"/>, so it can neither null a
/// client a concurrent <see cref="EnsureConnectedAsync"/> just installed nor race that connect. The wait
/// is bounded: a connect wedged past <see cref="LoopJoinTimeout"/> may not wedge teardown too (the device
/// is retired, so that connect disposes its own client when it finally returns).
/// </summary>
private async Task DisposeClientGatedAsync(DeviceState state)
{
var entered = await state.ConnectGate.WaitAsync(LoopJoinTimeout).ConfigureAwait(false);
try { state.DisposeClient(); }
catch (Exception ex) { _logger.LogDebug(ex, "FOCAS client dispose failed for {Host}", state.Options.HostAddress); }
finally { if (entered) state.ConnectGate.Release(); }
}
/// <inheritdoc />
@@ -423,58 +537,28 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
internal static readonly TimeSpan LoopJoinTimeout = TimeSpan.FromSeconds(5);
/// <inheritdoc />
// Safe to overlap with itself (Lane A disposes beside a still-running shutdown — the second call waits for
// the first's teardown via _releaseTail) and with an abandoned InitializeAsync (the generation bump below
// supersedes it, so it publishes nothing).
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
var generation = Interlocked.Increment(ref _lifecycleGeneration);
await _poll.DisposeAsync().ConfigureAwait(false);
// Pause, not dispose — see _alarmProjection. A later InitializeAsync resumes it.
await _alarmProjection.PauseAsync().ConfigureAwait(false);
// cancel every loop first, JOIN them (bounded), and only then dispose the CTSs and the
// client. Fire-and-forget loops used to outlive shutdown: one parked in EnsureConnectedAsync could
// finish its connect AFTER DisposeClient and leave a live socket on a shut-down driver.
var loops = new List<Task>();
foreach (var state in _devices.Values)
// Pauses (not disposes) the alarm projection — see _alarmProjection; a later InitializeAsync resumes it.
await ReleaseDeviceStatesAsync().ConfigureAwait(false);
lock (_lifecycleLock)
{
// Cancel-then-dispose can race in tight shutdown loops; swallowing is intentional
// but we now log the cause so a noisy shutdown leaves a Debug trace.
try { state.ProbeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling probe CTS for {Host} failed", state.Options.HostAddress); }
try { state.RecycleCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling recycle CTS for {Host} failed", state.Options.HostAddress); }
try { state.FixedTreeCts?.Cancel(); }
catch (Exception ex) { _logger.LogDebug(ex, "Cancelling fixed-tree CTS for {Host} failed", state.Options.HostAddress); }
if (state.ProbeTask is { } pt) loops.Add(pt);
if (state.RecycleTask is { } rt) loops.Add(rt);
if (state.FixedTreeTask is { } ft) loops.Add(ft);
// A later initialise owns the tables (and health) now — leave them alone.
if (generation != Volatile.Read(ref _lifecycleGeneration)) return;
Volatile.Write(ref _tagsByRawPath, new Dictionary<string, FocasTagDefinition>(StringComparer.Ordinal));
Volatile.Write(ref _alarmTags, new Dictionary<string, FocasAlarmTag>(StringComparer.Ordinal));
Volatile.Write(ref _parsedAddressesByTagName,
new ConcurrentDictionary<string, FocasAddress>(StringComparer.OrdinalIgnoreCase));
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
_fixedTreeActive = false;
ResetAlarmHealth();
Volatile.Write(ref _health, new DriverHealth(DriverState.Unknown, Volatile.Read(ref _health).LastSuccessfulRead, null));
}
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("FOCAS shutdown: {Count} background loop(s) did not stop within {Timeout}; proceeding with teardown",
loops.Count, LoopJoinTimeout);
}
foreach (var state in _devices.Values)
{
state.ProbeCts?.Dispose();
state.ProbeCts = null;
state.ProbeTask = null;
state.RecycleCts?.Dispose();
state.RecycleCts = null;
state.RecycleTask = null;
state.FixedTreeCts?.Dispose();
state.FixedTreeCts = null;
state.FixedTreeTask = null;
state.DisposeClient();
}
_devices.Clear();
Volatile.Write(ref _tagsByRawPath, new Dictionary<string, FocasTagDefinition>(StringComparer.Ordinal));
Volatile.Write(ref _alarmTags, new Dictionary<string, FocasAlarmTag>(StringComparer.Ordinal));
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
_parsedAddressesByTagName.Clear();
_fixedTreeActive = false;
ResetAlarmHealth();
Volatile.Write(ref _health, new DriverHealth(DriverState.Unknown, Volatile.Read(ref _health).LastSuccessfulRead, null));
}
/// <inheritdoc />
@@ -499,12 +583,14 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
}
/// <summary>Gets the number of configured devices.</summary>
internal int DeviceCount => _devices.Count;
internal int DeviceCount => Devices.Count;
/// <summary>Test seam — the live device table snapshot (swapped whole, never mutated in place).</summary>
internal IReadOnlyDictionary<string, DeviceState> DeviceTable => Devices;
/// <summary>Gets the state of a device by host address.</summary>
/// <param name="hostAddress">The host address of the device.</param>
/// <returns>The device state if found; otherwise null.</returns>
internal DeviceState? GetDeviceState(string hostAddress) =>
_devices.TryGetValue(hostAddress, out var s) ? s : null;
Devices.TryGetValue(hostAddress, out var s) ? s : null;
/// <summary>Test seam — the resolved options the factory built this driver from.</summary>
internal FocasDriverOptions Options => _options;
@@ -592,7 +678,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now);
continue;
}
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device))
if (!Devices.TryGetValue(def.DeviceHostAddress, out var device))
{
results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now);
continue;
@@ -680,7 +766,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
var reference = references[i];
if (_fixedTreeActive && TryReadFixedTree(reference, now) is not null) continue;
if (!_resolver.TryResolve(reference, out var def)) continue;
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue;
if (!Devices.TryGetValue(def.DeviceHostAddress, out var device)) continue;
var parsed = ResolveParsedAddress(def);
if (parsed.Kind != FocasAreaKind.Pmc || parsed.PmcLetter is null) continue;
(byDevice ??= []).TryAdd(device, []);
@@ -802,7 +888,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
results[i] = new WriteResult(FocasStatusMapper.BadNotWritable);
continue;
}
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device))
if (!Devices.TryGetValue(def.DeviceHostAddress, out var device))
{
results[i] = new WriteResult(FocasStatusMapper.BadNodeIdUnknown);
continue;
@@ -890,7 +976,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
// Fixed-tree T1 — Identity + Axes subtrees, populated once per session
// from cnc_sysinfo + cnc_rdaxisname at init time and kept in DeviceState.
if (_fixedTreeActive
&& _devices.TryGetValue(device.HostAddress, out var state)
&& Devices.TryGetValue(device.HostAddress, out var state)
&& state.FixedTreeCache is { } cache)
{
var identity = deviceFolder.Folder("Identity", "Identity");
@@ -1142,7 +1228,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// <inheritdoc />
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses() =>
[.. _devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))];
[.. Devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))];
private async Task ProbeLoopAsync(DeviceState state, CancellationToken ct)
{
@@ -1510,7 +1596,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// </summary>
private DataValueSnapshot? TryReadFixedTree(string reference, DateTime now)
{
foreach (var state in _devices.Values)
foreach (var state in Devices.Values)
{
if (!reference.StartsWith(state.Options.HostAddress + "/", StringComparison.OrdinalIgnoreCase)) continue;
if (TryReadFixedTreeOnDevice(state, reference[(state.Options.HostAddress.Length + 1)..], now) is { } hit)
@@ -1776,7 +1862,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// <returns>The device's active alarms.</returns>
internal async Task<IReadOnlyList<FocasActiveAlarm>> ReadDeviceAlarmsAsync(string hostAddress, CancellationToken ct)
{
if (!_devices.TryGetValue(hostAddress, out var state))
if (!Devices.TryGetValue(hostAddress, out var state))
throw new InvalidOperationException($"FOCAS device '{hostAddress}' is not configured.");
var client = await EnsureConnectedAsync(state, ct).ConfigureAwait(false);
return await client.ReadAlarmsAsync(ct).ConfigureAwait(false);
@@ -1822,6 +1908,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
try
{
// A released device (re-initialise / shutdown) is never reconnected: whoever still holds it read
// the table before the swap, and a client opened now would be orphaned on a state nobody tears down.
if (device.IsRetired)
throw new ObjectDisposedException(nameof(DeviceState),
$"FOCAS device {device.Options.HostAddress} was released by a re-initialise or shutdown.");
// Another caller may have connected while this one waited.
if (device.Client is { IsConnected: true } connected) return connected;
@@ -1864,6 +1956,15 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
throw;
}
device.Backoff.RecordSuccess(); // recovery resets the window immediately
// Retired WHILE this connect ran (the release waits on this gate only up to its bound): the connect
// is ours to undo, or it leaves a live session on a shut-down driver.
if (device.IsRetired)
{
client.Dispose();
if (ReferenceEquals(device.Client, client)) device.Client = null;
throw new ObjectDisposedException(nameof(DeviceState),
$"FOCAS device {device.Options.HostAddress} was released while connecting.");
}
return client;
}
finally { device.ConnectGate.Release(); }
@@ -2020,6 +2121,17 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
/// </summary>
public ConcurrentDictionary<int, double> LastSpindleMotorSpeeds { get; } = new();
private int _retired;
/// <summary>
/// True once a re-initialise / shutdown has released this device. A retired device never connects
/// again, and a connect that was already running when it retired disposes what it opened.
/// </summary>
public bool IsRetired => Volatile.Read(ref _retired) != 0;
/// <summary>Marks the device released. Idempotent.</summary>
public void Retire() => Volatile.Write(ref _retired, 1);
/// <summary>Disposes the FOCAS client instance.</summary>
public void DisposeClient()
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FakeFocasClient.cs
+1 -1
View File
@@ -259,7 +259,7 @@ internal sealed class FakeFocasClientFactory : IFocasClientFactory
public IFocasClient Create()
{
var c = Customise?.Invoke() ?? new FakeFocasClient();
Clients.Add(c);
lock (Clients) Clients.Add(c);
return c;
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasDeviceTableLifecycleTests.cs
+200
View File
@@ -0,0 +1,200 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
/// <summary>
/// #740 / R07:H13 — the device table is read by the retained poll engine, the fixed-tree loop, the alarm
/// loop and the actor's health tick, so a re-initialise must build a fresh table and swap it in rather than
/// clearing and refilling the live one. Also pins the lifecycle overlaps the runtime produces since Lane A:
/// a dispose running beside a still-running shutdown, and an abandoned initialise overlapping the next
/// attempt's shutdown.
/// </summary>
[Trait("Category", "Unit")]
public sealed class FocasDeviceTableLifecycleTests
{
private const string Host = "focas://10.0.0.5:8193";
private static FocasDriverOptions Options(bool probe = false, bool fixedTree = false) => new()
{
Devices = [new FocasDeviceOptions(Host)],
RawTags = [FocasRawTags.Entry(new FocasTagDefinition("raw/R", Host, "R100", FocasDataType.Byte))],
Probe = new FocasProbeOptions { Enabled = probe, Interval = TimeSpan.FromMilliseconds(20) },
FixedTree = new FocasFixedTreeOptions { Enabled = fixedTree, PollInterval = TimeSpan.FromMilliseconds(10) },
};
/// <summary>A re-initialise publishes a NEW table; a reader holding the old one never sees it emptied.</summary>
[Fact]
public async Task Reinitialize_swaps_in_a_new_table_and_never_mutates_the_one_a_reader_holds()
{
var drv = new FocasDriver(Options(), "focas-swap", new FakeFocasClientFactory());
await drv.InitializeAsync("{}", CancellationToken.None);
var before = drv.DeviceTable;
before.Count.ShouldBe(1);
await drv.InitializeAsync("{}", CancellationToken.None);
drv.DeviceTable.ShouldNotBeSameAs(before);
drv.DeviceTable.Count.ShouldBe(1);
before.Count.ShouldBe(1, "the table a concurrent reader enumerates must not be cleared under it");
await drv.ShutdownAsync(CancellationToken.None);
before.Count.ShouldBe(1);
drv.DeviceTable.Count.ShouldBe(0);
}
/// <summary>Hammering every table reader during repeated re-initialises throws nothing — and reads still succeed.</summary>
[Fact]
public async Task Readers_survive_repeated_reinitialise()
{
var factory = new FakeFocasClientFactory();
var drv = new FocasDriver(Options(fixedTree: true), "focas-hammer", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
using var stop = new CancellationTokenSource();
var good = 0;
var errors = new List<Exception>();
var reader = Task.Run(async () =>
{
while (!stop.IsCancellationRequested)
{
try
{
_ = drv.GetHostStatuses();
var snaps = await drv.ReadAsync(["raw/R", $"{Host}/Identity/SeriesNumber"], CancellationToken.None);
if (snaps[0].StatusCode == FocasStatusMapper.Good) Interlocked.Increment(ref good);
}
catch (Exception ex) { lock (errors) errors.Add(ex); }
}
});
for (var i = 0; i < 100; i++)
await drv.InitializeAsync("{}", CancellationToken.None);
await Task.Delay(100);
stop.Cancel();
await reader;
await drv.ShutdownAsync(CancellationToken.None);
errors.ShouldBeEmpty();
good.ShouldBeGreaterThan(0, "positive control: the reader really did read through the table");
}
/// <summary>A client whose connect parks until released and ignores cancellation.</summary>
private sealed class ParkedConnectClient(TaskCompletionSource entered, Task release) : FakeFocasClient
{
public override async Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct)
{
entered.TrySetResult();
await release.ConfigureAwait(false);
await base.ConnectAsync(address, timeout, CancellationToken.None).ConfigureAwait(false);
}
}
/// <summary>
/// A data-path read parked in connect when shutdown runs must not leave a live client on the released
/// device: the device is retired, so the connect that completes afterwards disposes what it opened.
/// </summary>
[Fact]
public async Task Connect_completing_after_shutdown_does_not_leave_a_live_client()
{
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var factory = new FakeFocasClientFactory { Customise = () => new ParkedConnectClient(entered, release.Task) };
var drv = new FocasDriver(Options(), "focas-retire", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
var read = drv.ReadAsync(["raw/R"], CancellationToken.None);
await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
var shutdown = drv.ShutdownAsync(CancellationToken.None);
await Task.Delay(50);
release.SetResult();
await shutdown.WaitAsync(TimeSpan.FromSeconds(10));
var snaps = await read.WaitAsync(TimeSpan.FromSeconds(5));
factory.Clients.Count.ShouldBe(1);
factory.Clients[0].IsConnected.ShouldBeFalse("a connect finishing on a retired device must dispose its client");
factory.Clients[0].DisposeCount.ShouldBeGreaterThanOrEqualTo(1);
snaps[0].StatusCode.ShouldBe(FocasStatusMapper.BadCommunicationError);
}
/// <summary>
/// An initialise abandoned mid-way (the runtime timed it out and moved on to a shutdown) must not
/// resurrect the driver when it finally completes: no device table, no loops, no connection.
/// </summary>
[Fact]
public async Task Abandoned_initialise_finishing_after_shutdown_starts_nothing()
{
var factory = new FakeFocasClientFactory();
var options = Options(probe: true);
using var gate = new ManualResetEventSlim(false);
var inRebind = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var drv = new FocasDriver(options, "focas-abandoned", factory, rebind: _ =>
{
inRebind.TrySetResult();
gate.Wait(TimeSpan.FromSeconds(10));
return new FocasDriverBinding(options, factory);
});
var init = Task.Run(() => drv.InitializeAsync("{\"devices\":[]}", CancellationToken.None));
await inRebind.Task.WaitAsync(TimeSpan.FromSeconds(5));
await drv.ShutdownAsync(CancellationToken.None);
gate.Set();
await Should.ThrowAsync<OperationCanceledException>(() => init.WaitAsync(TimeSpan.FromSeconds(10)));
await Task.Delay(200); // several probe intervals — a started loop would have connected by now
drv.DeviceCount.ShouldBe(0);
factory.Clients.ShouldBeEmpty();
drv.GetHealth().State.ShouldNotBe(DriverState.Healthy);
// Positive control: the same driver, initialised without being superseded, does start its probe.
await drv.InitializeAsync("{\"devices\":[]}", CancellationToken.None);
await WaitUntil(() => factory.Clients.Count > 0);
drv.DeviceCount.ShouldBe(1);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
await drv.DisposeAsync();
}
/// <summary>A loop wedged in a read that ignores cancellation for a while.</summary>
private sealed class SlowProbeClient : FakeFocasClient
{
public override async Task<bool> ProbeAsync(CancellationToken ct)
{
await Task.Delay(400, CancellationToken.None).ConfigureAwait(false);
return true;
}
}
/// <summary>
/// Lane A disposes while its own shutdown may still be running. The dispose must not return before the
/// teardown it overlaps has released the client — it waits for it rather than finding an empty table.
/// </summary>
[Fact]
public async Task Dispose_concurrent_with_shutdown_waits_for_the_teardown()
{
var factory = new FakeFocasClientFactory { Customise = () => new SlowProbeClient() };
var drv = new FocasDriver(Options(probe: true), "focas-dispose", factory);
await drv.InitializeAsync("{}", CancellationToken.None);
await WaitUntil(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected);
var shutdown = drv.ShutdownAsync(CancellationToken.None);
var dispose = drv.DisposeAsync().AsTask();
await dispose.WaitAsync(TimeSpan.FromSeconds(10));
factory.Clients.ShouldAllBe(c => !c.IsConnected);
factory.Clients.ShouldAllBe(c => c.DisposeCount >= 1);
await shutdown.WaitAsync(TimeSpan.FromSeconds(10));
drv.DeviceCount.ShouldBe(0);
}
private static async Task WaitUntil(Func<bool> condition)
{
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
while (!condition())
{
if (DateTime.UtcNow > deadline) throw new TimeoutException("condition not reached");
await Task.Delay(10);
}
}
}