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

Merge branch 'fix/archreview-crit4-twincat-replay'

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-09 06:07:33 -04:00
parent dfc02af3f0 af318fb442
commit 8f7b0017ef
3 changed files with 413 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs
+229 -39
View File
@@ -440,12 +440,14 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
private readonly ConcurrentDictionary<long, NativeSubscription> _nativeSubs = new();
private long _nextNativeSubId;
private long _nextRegId;
/// <inheritdoc />
// Subscribe via native ADS notifications when TwinCATDriverOptions.UseNativeNotifications is
// true, otherwise fall through to the shared PollGroupEngine. Native path registers one
// ITwinCATNotificationHandle per tag against the target's PLC runtime — the PLC pushes
// changes on its own cycle so we skip the poll loop entirely. Unsub path disposes the handles.
// notification per tag against the target's PLC runtime — the PLC pushes changes on its own
// cycle so we skip the poll loop entirely. Each registration stores its REPLAYABLE INTENT on
// the owning device so EnsureConnectedAsync can re-register it after a reconnect (STAB-2).
public async Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
@@ -454,8 +456,7 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
var id = Interlocked.Increment(ref _nextNativeSubId);
var handle = new NativeSubscriptionHandle(id);
var registrations = new List<ITwinCATNotificationHandle>(fullReferences.Count);
var now = DateTime.UtcNow;
var registrations = new List<NativeRegistration>(fullReferences.Count);
try
{
@@ -464,18 +465,20 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
if (!_resolver.TryResolve(reference, out var def)) continue;
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue;
var client = await EnsureConnectedAsync(device, cancellationToken).ConfigureAwait(false);
var parsed = TwinCATSymbolPath.TryParse(def.SymbolPath);
var symbolName = parsed?.ToAdsSymbolName() ?? def.SymbolPath;
var bitIndex = parsed?.BitIndex;
var reg = await client.AddNotificationAsync(
symbolName, def.DataType, bitIndex, publishingInterval,
_options.NotificationMaxDelayMs,
// The callback closes over the stable (handle, reference) pair, so it survives a
// client swap unchanged — replay re-registers the SAME intent onto the new client.
var reg = new NativeRegistration(
Interlocked.Increment(ref _nextRegId), device, reference, symbolName,
def.DataType, bitIndex, publishingInterval,
(_, value) => OnDataChange?.Invoke(this,
new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))),
cancellationToken).ConfigureAwait(false);
value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))));
await RegisterNotificationAsync(device, reg, cancellationToken).ConfigureAwait(false);
registrations.Add(reg);
}
}
@@ -496,11 +499,38 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
return handle;
}
/// <summary>
/// Registers one native notification and records its replayable intent on the device — both
/// under the device <see cref="DeviceState.ConnectGate"/> so the register can't race a
/// concurrent reconnect-replay (which also runs under that gate). The intent is added to the
/// device registry only AFTER the handle is live, so a replay triggered by this same
/// <c>EnsureConnected</c> can't double-register the not-yet-stored tag.
/// </summary>
private async Task RegisterNotificationAsync(DeviceState device, NativeRegistration reg, CancellationToken ct)
{
await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
try
{
var client = await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride: null).ConfigureAwait(false);
var newHandle = await client.AddNotificationAsync(
reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval,
_options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false);
reg.SwapHandle(newHandle); // initial set — previous handle is null
device.NativeRegistrations[reg.Id] = reg;
}
finally
{
device.ConnectGate.Release();
}
}
/// <inheritdoc />
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
if (handle is NativeSubscriptionHandle native && _nativeSubs.TryRemove(native.Id, out var sub))
{
// NativeRegistration.Dispose removes itself from its device registry (so a later replay
// won't touch it) and disposes its live handle.
foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } }
return Task.CompletedTask;
}
@@ -516,7 +546,64 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
private sealed record NativeSubscription(
NativeSubscriptionHandle Handle,
IReadOnlyList<ITwinCATNotificationHandle> Registrations);
IReadOnlyList<NativeRegistration> Registrations);
/// <summary>
/// One live native-notification registration: its replayable intent plus the current live
/// <see cref="ITwinCATNotificationHandle"/>. The handle is swapped (not re-created) when the
/// device reconnects, so unsubscribe / teardown always dispose the latest one. The handle
/// field is exchanged atomically so a reconnect-replay swap and an unsubscribe dispose can't
/// corrupt each other on the common path.
/// </summary>
internal sealed class NativeRegistration : IDisposable
{
/// <summary>Globally-unique registration id (registry key on the device).</summary>
public long Id { get; }
/// <summary>The device this notification is registered against.</summary>
public DeviceState Device { get; }
/// <summary>Driver-side full reference the pushed value is reported under.</summary>
public string Reference { get; }
/// <summary>ADS symbol name (resolved from the symbol path).</summary>
public string SymbolName { get; }
/// <summary>Declared data type.</summary>
public TwinCATDataType DataType { get; }
/// <summary>Bit index for a BOOL-within-word symbol; null otherwise.</summary>
public int? BitIndex { get; }
/// <summary>Requested notification interval.</summary>
public TimeSpan Interval { get; }
/// <summary>Value-change callback (closes over the owning subscription handle + reference).</summary>
public Action<string, object?> OnChange { get; }
private ITwinCATNotificationHandle? _handle;
/// <summary>Initializes a new instance of the <see cref="NativeRegistration"/> class.</summary>
public NativeRegistration(long id, DeviceState device, string reference, string symbolName,
TwinCATDataType dataType, int? bitIndex, TimeSpan interval, Action<string, object?> onChange)
{
Id = id;
Device = device;
Reference = reference;
SymbolName = symbolName;
DataType = dataType;
BitIndex = bitIndex;
Interval = interval;
OnChange = onChange;
}
/// <summary>Installs a freshly-registered handle and returns the previous one (null on first set).</summary>
/// <param name="newHandle">The newly-registered notification handle.</param>
/// <returns>The previous handle, or null.</returns>
public ITwinCATNotificationHandle? SwapHandle(ITwinCATNotificationHandle newHandle) =>
Interlocked.Exchange(ref _handle, newHandle);
/// <summary>Removes this registration from its device and disposes its live handle.</summary>
public void Dispose()
{
Device.NativeRegistrations.TryRemove(Id, out _);
var h = Interlocked.Exchange(ref _handle, null);
try { h?.Dispose(); } catch { /* best-effort */ }
}
}
// ---- IHostConnectivityProbe ----
@@ -536,12 +623,21 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
var client = await EnsureConnectedAsync(state, ct, _options.Probe.Timeout)
.ConfigureAwait(false);
success = await client.ProbeAsync(ct).ConfigureAwait(false);
// A locally-connected client whose wire probe FAILS is wire-dead but port-open:
// force a recycle so the next tick rebuilds + replays native notifications. Without
// this the IsConnected fast-path would keep handing back the dead client forever
// (the STAB-2 compounding bug — native pushes would never recover).
if (!success)
await RecycleClientAsync(state).ConfigureAwait(false);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; }
catch
{
// Probe failure — EnsureConnectedAsync's connect-failure path already disposed
// + cleared the client, so next tick will reconnect.
// Probe threw — the connect-failure path clears the client only when the CONNECT
// failed; a throw from ProbeAsync on an already-connected client would leave it in
// place, so force a recycle to guarantee the next tick rebuilds + replays.
await RecycleClientAsync(state).ConfigureAwait(false);
}
TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
@@ -608,41 +704,126 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false);
try
{
// Re-check under the gate: another caller may have connected while we waited.
if (device.Client is { IsConnected: true } c) return c;
return await EnsureConnectedUnderGateAsync(device, ct, timeoutOverride).ConfigureAwait(false);
}
finally
{
device.ConnectGate.Release();
}
}
// Discard a stale (created-but-disconnected) client before making a fresh one.
if (device.Client is { IsConnected: false } stale)
{
try { stale.Dispose(); } catch { /* best-effort */ }
device.Client = null;
}
/// <summary>
/// The connect-or-reconnect core, assuming the caller already holds
/// <see cref="DeviceState.ConnectGate"/>. When it installs a NEW client (a genuine
/// reconnect, not a fast-path reuse) it replays every stored native-notification intent
/// onto it — otherwise native ADS pushes would silently stop after a drop (STAB-2). Both
/// the initial subscribe (<see cref="RegisterNotificationAsync"/>) and the probe loop call
/// through here under the same gate, so a replay can never race a register.
/// </summary>
private async Task<ITwinCATClient> EnsureConnectedUnderGateAsync(
DeviceState device, CancellationToken ct, TimeSpan? timeoutOverride)
{
// Re-check under the gate: another caller may have connected while we waited.
if (device.Client is { IsConnected: true } c) return c;
var client = _clientFactory.Create();
client.OnSymbolVersionChanged += HandleSymbolVersionChanged;
// timeoutOverride lets the probe loop use TwinCATProbeOptions.Timeout for probe-
// initiated connects rather than the driver-level _options.Timeout. Reads / writes
// pass null and get the driver default.
var effectiveTimeout = timeoutOverride ?? _options.Timeout;
// Discard a stale (created-but-disconnected) client before making a fresh one.
if (device.Client is { IsConnected: false } stale)
{
stale.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
try { stale.Dispose(); } catch { /* best-effort */ }
device.Client = null;
}
var client = _clientFactory.Create();
client.OnSymbolVersionChanged += HandleSymbolVersionChanged;
// timeoutOverride lets the probe loop use TwinCATProbeOptions.Timeout for probe-
// initiated connects rather than the driver-level _options.Timeout. Reads / writes
// pass null and get the driver default.
var effectiveTimeout = timeoutOverride ?? _options.Timeout;
try
{
await client.ConnectAsync(device.ParsedAddress, effectiveTimeout, ct)
.ConfigureAwait(false);
_logger.LogInformation(
"TwinCAT driver '{DriverInstanceId}' connected to {HostAddress}",
_driverInstanceId, device.Options.HostAddress);
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"TwinCAT driver '{DriverInstanceId}' failed to connect to {HostAddress}",
_driverInstanceId, device.Options.HostAddress);
client.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
client.Dispose();
throw;
}
device.Client = client;
// Re-register every native notification onto the fresh client. Native ADS is push, so
// (unlike the poll fallback) it cannot self-heal by re-reading — without this the
// subscription is silently dead after a reconnect.
await ReplayNativeRegistrationsAsync(device, client, ct).ConfigureAwait(false);
return client;
}
/// <summary>
/// Re-registers every stored <see cref="NativeRegistration"/> for a device onto the given
/// freshly-connected client and swaps in the new live handle (disposing the dead one from
/// the previous client). Runs under <see cref="DeviceState.ConnectGate"/>. A single
/// registration that fails to replay is logged and skipped — one bad symbol must not abort
/// recovery of the rest.
/// </summary>
private async Task ReplayNativeRegistrationsAsync(DeviceState device, ITwinCATClient client, CancellationToken ct)
{
var intents = device.NativeRegistrations.Values.ToArray();
if (intents.Length == 0) return;
var replayed = 0;
foreach (var reg in intents)
{
try
{
await client.ConnectAsync(device.ParsedAddress, effectiveTimeout, ct)
.ConfigureAwait(false);
_logger.LogInformation(
"TwinCAT driver '{DriverInstanceId}' connected to {HostAddress}",
_driverInstanceId, device.Options.HostAddress);
var newHandle = await client.AddNotificationAsync(
reg.SymbolName, reg.DataType, reg.BitIndex, reg.Interval,
_options.NotificationMaxDelayMs, reg.OnChange, ct).ConfigureAwait(false);
var old = reg.SwapHandle(newHandle);
try { old?.Dispose(); } catch { /* dead handle from the disposed client */ }
replayed++;
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"TwinCAT driver '{DriverInstanceId}' failed to connect to {HostAddress}",
_driverInstanceId, device.Options.HostAddress);
client.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
client.Dispose();
throw;
"TwinCAT driver '{DriverInstanceId}' failed to replay native notification " +
"'{Symbol}' on device '{HostAddress}' after reconnect",
_driverInstanceId, reg.SymbolName, device.Options.HostAddress);
}
}
_logger.LogInformation(
"TwinCAT driver '{DriverInstanceId}' re-registered {Replayed}/{Total} native notifications " +
"on device '{HostAddress}' after reconnect",
_driverInstanceId, replayed, intents.Length, device.Options.HostAddress);
}
/// <summary>
/// Disposes + nulls a device's client under <see cref="DeviceState.ConnectGate"/> so the
/// next <see cref="EnsureConnectedAsync"/> rebuilds it (and replays notifications). Called
/// by the probe loop when a wire-level probe fails on a locally-"connected" client — the
/// <see cref="ITwinCATClient.IsConnected"/> fast-path keys on the AMS-port state, not wire
/// liveness, so without this forced recycle a wire-dead-but-port-open client would keep
/// being reused and native pushes would never recover.
/// </summary>
private async Task RecycleClientAsync(DeviceState device)
{
await device.ConnectGate.WaitAsync(CancellationToken.None).ConfigureAwait(false);
try
{
if (device.Client is { } c)
{
c.OnSymbolVersionChanged -= HandleSymbolVersionChanged;
try { c.Dispose(); } catch { /* best-effort */ }
device.Client = null;
}
device.Client = client;
return client;
}
finally
{
@@ -722,6 +903,15 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
/// create-or-dispose for this device.</summary>
public SemaphoreSlim ConnectGate { get; } = new(1, 1);
/// <summary>
/// Live native-notification registrations against this device, keyed by registration id.
/// Holds the <b>replayable intent</b> (symbol / type / bit / interval / callback), not just
/// an opaque handle, so that when <see cref="TwinCATDriver.EnsureConnectedAsync"/> swaps in
/// a fresh client after a drop it can re-register every notification onto it (STAB-2).
/// Populated by <c>SubscribeAsync</c>, drained by <c>UnsubscribeAsync</c> / teardown.
/// </summary>
public ConcurrentDictionary<long, NativeRegistration> NativeRegistrations { get; } = new();
/// <summary>Gets the lock object for synchronizing host state transitions.</summary>
public object ProbeLock { get; } = new();
/// <summary>Gets or sets the current host connectivity state.</summary>