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

fix(twincat): review round 2 - teardown confined to its generation; write health; re-arm epoch

Browse Source 
- I1: the poll engine belongs to a teardown generation. ShutdownAsync/Dispose
  retire the current engine and install a fresh one in the same swap as the
  device/tag tables, then dispose the retired one (possibly in the background
  after a cancelled or over-budget shutdown). A subscription made after the
  next InitializeAsync can no longer land in an engine whose late dispose tail
  would dispose its CTS without cancelling it (a loop no UnsubscribeAsync could
  find). Poll handles are generation-qualified (TwinCATPollHandle) because each
  engine numbers handles from 1; SubscribeAsync re-homes a subscription that
  raced a retirement.
- M1: a write's status feeds per-device health - Good clears that device's
  comms fault and advances LastSuccessfulRead, a comms status degrades, a
  refusal keeps the state.
- M2: a symbol-version re-arm bumps a per-registration epoch; an in-flight
  retry that fails afterwards no longer re-imposes the backoff.
- M3: a re-arm restarts the refusal backoff at 30 s.
- M4: documented that PLC-not-in-RUN detection needs the probe.
- M5: WSTRING and multi-dimensional array encode-plan tests; the cancelled-
  shutdown test asserts Unknown health and awaits its background teardown.
- M6: an infinity spelled as a string ('Infinity', '-inf') passes like a
  float/double infinity; a finite overflow stays BadOutOfRange.

Refs #739
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:00:23 -04:00
parent 991d277c03
commit a8d313ed5a
5 changed files with 417 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs
+3
View File
@@ -1098,10 +1098,13 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
return d;
}
// An explicit infinity — a float/double infinity, or a string that SPELLS one ("Infinity", "-inf", "∞")
// — is the caller's own value and passes; a finite number that overflows ("1e400") does not.
private static bool IsExplicitInfinity(object? value) => value switch
{
float f => float.IsInfinity(f),
double d => double.IsInfinity(d),
string s => s.Contains("inf", StringComparison.OrdinalIgnoreCase) || s.Contains('∞'),
_ => false,
};
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs
+123 -20
View File
@@ -27,7 +27,28 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
private readonly string _driverInstanceId;
private readonly ITwinCATClientFactory _clientFactory;
private readonly ILogger<TwinCATDriver> _logger;
private readonly PollGroupEngine _poll;
// The poll engine belongs to a teardown GENERATION (#739 fix round 2): ShutdownAsync / Dispose retire the
// current engine and install a fresh one in the same swap, then dispose the retired one — possibly in the
// background after a cancelled or over-budget shutdown. PollGroupEngine.DisposeAsync clears whatever its
// own table holds when it finishes, so a subscription made after a later InitializeAsync must never land
// in a retiring engine, or the old teardown would dispose its CTS without cancelling it (a poll loop no
// UnsubscribeAsync can find, running forever).
private volatile PollGeneration _poll;
private long _nextPollGeneration;
/// <summary>One generation's poll engine plus its serial (the serial is part of every handle it issues).</summary>
private sealed record PollGeneration(long Serial, PollGroupEngine Engine);
/// <summary>
/// A poll subscription handle, qualified by the engine generation that issued it. Each engine numbers its
/// handles from 1, so without the generation a retired engine's late callback would carry a handle EQUAL
/// to a new subscription's and be taken for it.
/// </summary>
private sealed record TwinCATPollHandle(long Generation, ISubscriptionHandle Inner) : ISubscriptionHandle
{
/// <inheritdoc />
public string DiagnosticId => $"twincat-poll-g{Generation}-{Inner.DiagnosticId}";
}
// 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
@@ -120,18 +141,27 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
_logger = logger ?? NullLogger<TwinCATDriver>.Instance;
_resolver = new EquipmentTagRefResolver<TwinCATTagDefinition>(
r => _tagsByRawPath.TryGetValue(r, out var t) ? t : null);
_poll = new PollGroupEngine(
_poll = CreatePollEngine();
}
/// <summary>Builds one generation's poll engine; its callbacks publish generation-qualified handles.</summary>
private PollGeneration CreatePollEngine()
{
var driverInstanceId = _driverInstanceId;
var serial = Interlocked.Increment(ref _nextPollGeneration);
return new PollGeneration(serial, new PollGroupEngine(
reader: ReadAsync,
onChange: (handle, tagRef, snapshot) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
OnDataChange?.Invoke(this, new DataChangeEventArgs(new TwinCATPollHandle(serial, handle), tagRef, snapshot)),
onError: HandlePollError,
backoffCap: PollBackoffCap,
onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items)),
onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this,
new DataChangeBatchEventArgs(new TwinCATPollHandle(serial, handle), items)),
// The engine owns no Meter (Core.Abstractions takes no metrics dependency), so this
// closure IS the production consumer of its overrun signal.
onOverrun: _ => OtOpcUaTelemetry.PollOverruns.Add(1,
new KeyValuePair<string, object?>(OtOpcUaTelemetry.PollOverrunDriverTypeTag, DriverTypeNames.TwinCAT),
new KeyValuePair<string, object?>(OtOpcUaTelemetry.PollOverrunDriverInstanceTag, driverInstanceId)));
new KeyValuePair<string, object?>(OtOpcUaTelemetry.PollOverrunDriverInstanceTag, driverInstanceId))));
}
/// <summary>Upper bound on the poll-loop failure backoff — adopts the S7-proven 30 s cap fleet-wide.</summary>
@@ -327,8 +357,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
}
}
/// <summary>Swaps out every table under the lifecycle lock and returns the device set the caller now owns.</summary>
private ICollection<DeviceState> TakeAllForTeardown()
/// <summary>
/// Swaps out every table AND the poll engine under the lifecycle lock and returns what the caller now
/// owns. Everything returned belongs to the retired generation only; nothing a later
/// <see cref="InitializeAsync"/> / <see cref="SubscribeAsync"/> creates is reachable from it.
/// </summary>
private (ICollection<DeviceState> devices, PollGroupEngine poll) TakeAllForTeardown()
{
lock (_lifecycleLock)
{
@@ -336,10 +370,16 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
var taken = _devices;
_devices = NewDeviceTable();
_tagsByRawPath = EmptyTags;
return taken.Values;
var poll = _poll;
_poll = CreatePollEngine();
return (taken.Values, poll.Engine);
}
}
/// <summary>The most recent teardown's background work (poll-engine dispose + device release). Test
/// visibility only: a cancelled shutdown leaves it running.</summary>
internal Task? LastTeardownForTest { get; private set; }
/// <inheritdoc />
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
@@ -362,11 +402,13 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } }
_nativeSubs.Clear();
var taken = TakeAllForTeardown();
var (taken, retiredPoll) = TakeAllForTeardown();
// 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));
// poll loop stuck in a read on it); the whole teardown is bounded. Both halves own ONLY the retired
// generation, so finishing in the background after a cancelled wait is safe.
var teardown = Task.WhenAll(retiredPoll.DisposeAsync().AsTask(), ReleaseDevicesAsync(taken));
LastTeardownForTest = teardown;
var cancelled = false;
try { await teardown.WaitAsync(ShutdownBudget, cancellationToken).ConfigureAwait(false); }
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@@ -515,6 +557,19 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
return new DataValueSnapshot(value, status, now, now);
}
/// <summary>
/// Applies the health contract to a write the client returned a status for: Good is the device
/// answering (clears that device's comms fault, advances <c>LastSuccessfulRead</c>); a comms-class
/// status is a comms fault; anything else is a refusal — the device answered, the state is unchanged.
/// </summary>
private void NoteWriteOutcome(DeviceState device, uint status, string reference)
{
var now = DateTime.UtcNow;
if (status == TwinCATStatusMapper.Good) RecordDeviceAnswered(now, device, force: true);
else if (IsCommsStatus(status)) MarkCommsFault($"ADS status {status:X8} writing {reference}", device);
else RecordDeviceAnsweredKeepState(now);
}
/// <summary>
/// True for a status that means the request did not get a device answer (router / transport /
/// timeout). Every other non-Good status is the device refusing this one tag (symbol not found, type
@@ -675,6 +730,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
{
gate.Release();
}
NoteWriteOutcome(device, results[i].StatusCode, w.FullReference);
continue;
}
@@ -685,12 +741,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
// array — the inverse of the array read.
results[i] = new WriteResult(await client.WriteArrayAsync(
symbolName, def.DataType, elements, cancellationToken).ConfigureAwait(false));
NoteWriteOutcome(device, results[i].StatusCode, w.FullReference);
continue;
}
var status = await client.WriteValueAsync(
symbolName, def.DataType, parsed?.BitIndex, w.Value, cancellationToken).ConfigureAwait(false);
results[i] = new WriteResult(status);
NoteWriteOutcome(device, status, w.FullReference);
}
catch (OperationCanceledException) { throw; }
catch (NotSupportedException)
@@ -826,7 +884,17 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
if (!_options.UseNativeNotifications)
return _poll.Subscribe(fullReferences, publishingInterval);
{
// A teardown racing this call may retire the engine we read: re-home the subscription on the
// current one so it never lives in an engine that is being disposed.
while (true)
{
var generation = _poll;
var inner = generation.Engine.Subscribe(fullReferences, publishingInterval);
if (ReferenceEquals(generation, _poll)) return new TwinCATPollHandle(generation.Serial, inner);
generation.Engine.Unsubscribe(inner);
}
}
var id = Interlocked.Increment(ref _nextNativeSubId);
var handle = new NativeSubscriptionHandle(id);
@@ -987,6 +1055,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// ADS state is not <c>Run</c> (coordinator ruling: a stopped / config-mode PLC is Degraded with a
/// reason even though it answers), or an uncleared comms fault seen on it.
/// </summary>
/// <remarks>
/// The ADS state comes from the probe's ReadState, so PLC-not-in-RUN detection needs the probe
/// (<c>Probe.Enabled</c>, the default); with the probe off the state is unknown and never degrades.
/// </remarks>
private static string? DeviceDegradedReason(DeviceState d)
{
if (d.ConsecutiveProbeFailures >= ProbeFailuresBeforeDegraded)
@@ -1042,9 +1114,20 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// or retried by the reconnect replay; a comms-class failure stays on the every-tick retry.
/// </summary>
/// <returns>True when the published status changes (the caller republishes the tag).</returns>
private static bool NoteDeadFailure(NativeRegistration reg, uint status, bool refused, DateTime now)
/// <param name="reg">The registration.</param>
/// <param name="status">The status the failure maps to.</param>
/// <param name="refused">True for a refusal (backs off), false for a comms-class failure.</param>
/// <param name="now">The failure time.</param>
/// <param name="epochAtAttempt">The registration's <see cref="NativeRegistration.RearmEpoch"/> when the attempt
/// started; when a re-arm happened while it was in flight the backoff is NOT re-imposed (the re-arm wins).</param>
private static bool NoteDeadFailure(NativeRegistration reg, uint status, bool refused, DateTime now,
int? epochAtAttempt = null)
{
if (refused)
if (epochAtAttempt is int epoch && epoch != reg.RearmEpoch)
{
// A symbol-version change re-armed this registration mid-attempt: keep it re-armed.
}
else if (refused)
{
reg.RefusalBackoff = reg.RefusalBackoff == TimeSpan.Zero
? RefusedRetryInitial
@@ -1141,7 +1224,10 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
}
// Async teardown: the engine drains the in-flight poll without blocking the caller's
// thread — this runs inside a DriverInstanceActor mailbox turn.
await _poll.UnsubscribeAsync(handle).ConfigureAwait(false);
// A handle of a retired generation has nothing left to stop: its engine was disposed with it.
var current = _poll;
if (handle is TwinCATPollHandle poll && poll.Generation == current.Serial)
await current.Engine.UnsubscribeAsync(poll.Inner).ConfigureAwait(false);
}
private sealed record NativeSubscriptionHandle(long Id) : ISubscriptionHandle
@@ -1255,9 +1341,25 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// <summary>Earliest time the probe-tick retry may re-attempt this dead registration.</summary>
public DateTime NextRetryUtc { get; set; }
/// <summary>Current refusal backoff (zero after a comms-class failure or a success).</summary>
/// <summary>Current refusal backoff (zero after a comms-class failure, a success or a re-arm).</summary>
public TimeSpan RefusalBackoff { get; set; }
private int _rearmEpoch;
/// <summary>Bumped by every symbol-version re-arm; a retry that started before it must not re-impose a backoff.</summary>
public int RearmEpoch => Volatile.Read(ref _rearmEpoch);
/// <summary>
/// Re-arms the retry after a PLC program change: retry on the next tick and restart the backoff
/// from <see cref="RefusedRetryInitial"/> (the refusal it backed off from may no longer apply).
/// </summary>
public void Rearm()
{
Interlocked.Increment(ref _rearmEpoch);
RefusalBackoff = TimeSpan.Zero;
NextRetryUtc = DateTime.MinValue;
}
private volatile bool _disposed;
/// <summary>Gets a value indicating whether this registration has been disposed (its callbacks are dropped).</summary>
@@ -1793,6 +1895,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
if (reg.HasLiveHandle) continue;
// A refused registration waits out its backoff (a symbol-version change re-arms it).
if (now < reg.NextRetryUtc) continue;
var epoch = reg.RearmEpoch;
try
{
var newHandle = await client.AddNotificationAsync(
@@ -1816,7 +1919,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
// or the reverse); the same failure again publishes nothing.
var refusal = ex as TwinCATNotificationRefusedException;
var status = refusal?.StatusCode ?? TwinCATStatusMapper.BadCommunicationError;
if (NoteDeadFailure(reg, status, refusal is not null, now))
if (NoteDeadFailure(reg, status, refusal is not null, now, epoch))
try { reg.OnRefused(status); } catch { /* never abort the rest */ }
_logger.LogDebug(ex,
"TwinCAT driver '{DriverInstanceId}' probe-tick retry of dead notification " +
@@ -1880,7 +1983,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
foreach (var d in _devices.Values)
if (ReferenceEquals(d.Client, sender))
foreach (var reg in d.NativeRegistrations.Values)
reg.NextRetryUtc = DateTime.MinValue;
reg.Rearm();
OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs(
"TwinCAT symbol-version-changed (DeviceSymbolVersionInvalid 0x0711) — PLC program re-downloaded",
ScopeHint: "TwinCAT"));
@@ -1902,12 +2005,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } }
_nativeSubs.Clear();
var taken = TakeAllForTeardown();
var (taken, retiredPoll) = TakeAllForTeardown();
var probes = BeginRelease(taken);
// 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();
var poll = retiredPoll.DisposeAsync().AsTask();
try { Task.WhenAll([poll, .. probes]).Wait(ProbeJoinTimeout); } catch { /* cancel races are expected */ }
EndRelease(taken);