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

fix(twincat): bounded, race-safe lifecycle; drop counter; post-unsubscribe purge (L8)

Browse Source 
- Device and tag tables are build-and-swap: InitializeAsync builds a fresh
  generation and publishes it by one swap under a lifecycle lock, then
  releases the replaced generation; teardown swaps in empty tables. A
  retained poll loop never sees a table mid-Clear (it used to read
  _tagsByRawPath while a re-init only ever added to it), and each device
  set is released by exactly one caller.
- A lifecycle generation makes an InitializeAsync overtaken by a shutdown
  (or a newer init) release what it built instead of installing it - the
  Lane A abandoned-init-overlaps-shutdown case.
- ShutdownAsync joins all probe loops together within 2 s (it awaited each
  unbounded - a probe wedged in the blocking ADS Connect hung the actor's
  teardown) in parallel with the poll-engine dispose, the whole bounded at
  4.5 s; DisposeAsync may run concurrently and returns at once. A connect
  that completes on a released device disposes its client.
- Releasing a generation disposes its native registrations, so a direct
  re-init no longer leaves live notifications bound to dead DeviceStates.
- Connect gates and RMW semaphores are no longer disposed under in-flight
  users (no kernel handle is ever allocated); a fault on a released device
  is not recorded as driver health.
- The notification channel counts DropWrite overflows
  (ITwinCATClient.DroppedNotificationCount); the probe tick warns with the
  delta. A queued value whose registration was disposed is purged in the
  client dispatch loop and dropped driver-side for any client.

Refs #739
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:24:10 -04:00
parent f17ba8e258
commit cbacaee4ee
5 changed files with 561 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs
+12
View File
@@ -56,10 +56,22 @@ internal class FakeTwinCATClient : ITwinCATClient
{
ConnectCount++;
if (ThrowOnConnect) throw Exception ?? new InvalidOperationException();
if (BlockConnect is { } block)
{
// Models the real client's blocking SDK Connect: ignores the token until released.
return block.Task.ContinueWith(_ => { IsConnected = true; }, TaskScheduler.Default);
}
IsConnected = true;
return Task.CompletedTask;
}
/// <summary>When set, <see cref="ConnectAsync"/> does not complete (and ignores cancellation) until this
/// is completed — a probe wedged in the synchronous ADS <c>Connect</c>.</summary>
public TaskCompletionSource? BlockConnect { get; set; }
/// <summary>Gets or sets the simulated notification-queue drop count.</summary>
public long DroppedNotificationCount { get; set; }
/// <summary>Simulates reading a value from a symbol.</summary>
/// <param name="symbolPath">The path to the symbol to read.</param>
/// <param name="type">The data type of the symbol.</param>
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATLifecycleTests.cs
+271
View File
@@ -0,0 +1,271 @@
using System.Collections.Concurrent;
using System.Diagnostics;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests;
/// <summary>
/// #739 L8 + the Lane A runtime contract (final stop = <c>ShutdownAsync</c> then <c>DisposeAsync</c>,
/// bounded 5 s + 250 ms): a probe wedged in the blocking ADS connect cannot hang teardown; dispose may run
/// while a shutdown is still running; an abandoned initialise overlapping the next shutdown never installs
/// its generation; a value queued before unsubscribe is not delivered after it; re-init swaps its tables
/// and retracts the previous generation's notifications; queue drops are reported.
/// </summary>
[Trait("Category", "Unit")]
public sealed class TwinCATLifecycleTests
{
private const string Host = "ads://5.23.91.23.1.1:851";
private static TwinCATDriver NewDriver(FakeTwinCATClientFactory factory, bool probe = true, bool native = true,
TwinCATCapturingLogger? log = null, params TwinCATTagDefinition[] tags) =>
new(new TwinCATDriverOptions
{
Devices = [new TwinCATDeviceOptions(Host)],
RawTags = TwinCATRawTags.Entries(tags.Length > 0
? tags
: [new TwinCATTagDefinition("Speed", Host, "MAIN.Speed", TwinCATDataType.DInt)]),
Probe = probe
? new TwinCATProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(30), Timeout = TimeSpan.FromMilliseconds(100) }
: new TwinCATProbeOptions { Enabled = false },
UseNativeNotifications = native,
}, "drv-1", factory, log);
private static async Task WaitForAsync(Func<bool> condition, TimeSpan timeout)
{
var deadline = DateTime.UtcNow + timeout;
while (!condition() && DateTime.UtcNow < deadline)
await Task.Delay(10);
}
/// <summary>
/// A probe wedged in the (uncancellable) connect no longer hangs ShutdownAsync — the join is bounded —
/// and the client the wedged connect eventually produces is disposed, not installed on a dead device.
/// </summary>
[Fact]
public async Task Shutdown_is_bounded_when_a_probe_is_wedged_in_connect()
{
var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var factory = new FakeTwinCATClientFactory { Customise = () => new FakeTwinCATClient { BlockConnect = wedge } };
var drv = NewDriver(factory);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2));
factory.Clients.Count.ShouldBe(1); // the probe is inside the wedged connect
var sw = Stopwatch.StartNew();
await drv.ShutdownAsync(TestContext.Current.CancellationToken).WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
sw.Elapsed.ShouldBeLessThan(TwinCATDriver.ShutdownBudget + TimeSpan.FromMilliseconds(500));
wedge.SetResult();
await WaitForAsync(() => factory.Clients[0].DisposeCount == 1, TimeSpan.FromSeconds(2));
factory.Clients[0].DisposeCount.ShouldBe(1);
drv.DeviceCount.ShouldBe(0);
}
/// <summary>
/// DisposeAsync while a ShutdownAsync is still running (joining a wedged probe) returns promptly and
/// both complete without throwing; the device is released exactly once.
/// </summary>
[Fact]
public async Task Dispose_concurrent_with_a_running_shutdown_completes_promptly()
{
var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var factory = new FakeTwinCATClientFactory { Customise = () => new FakeTwinCATClient { BlockConnect = wedge } };
var drv = NewDriver(factory);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2));
var shutdown = drv.ShutdownAsync(CancellationToken.None);
var sw = Stopwatch.StartNew();
await drv.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(250));
shutdown.IsCompleted.ShouldBeFalse("the positive control: the first shutdown really was still joining the probe");
wedge.SetResult();
await shutdown.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
drv.Dispose(); // and a late sync Dispose is a no-op
drv.GetHealth().State.ShouldBe(DriverState.Unknown);
}
/// <summary>
/// An InitializeAsync overtaken by a ShutdownAsync before it commits must not install its devices or
/// start probe loops when it resumes. Positive control: a fresh initialise afterwards installs normally.
/// </summary>
[Fact]
public async Task An_initialise_overtaken_by_shutdown_before_commit_installs_nothing()
{
var factory = new FakeTwinCATClientFactory();
var drv = NewDriver(factory);
var paused = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var resume = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
drv.BeforeCommitForTest = async () => { paused.TrySetResult(); await resume.Task; };
var abandoned = drv.InitializeAsync("{}", CancellationToken.None);
await paused.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
resume.SetResult();
await Should.ThrowAsync<OperationCanceledException>(() => abandoned.WaitAsync(TimeSpan.FromSeconds(5)));
drv.DeviceCount.ShouldBe(0);
await Task.Delay(150, TestContext.Current.CancellationToken); // several probe intervals
factory.Clients.ShouldBeEmpty("no probe loop of the overtaken generation may be running");
drv.GetHealth().State.ShouldBe(DriverState.Unknown);
drv.BeforeCommitForTest = null;
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
drv.DeviceCount.ShouldBe(1);
await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2));
factory.Clients.Count.ShouldBe(1);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
}
/// <summary>
/// The Lane A shape: an InitializeAsync the runtime abandoned while it is still releasing the previous
/// generation (joining a probe wedged in connect) overlaps the next ShutdownAsync. Shutdown stays
/// bounded and, once everything settles, no device, probe loop or live client survives.
/// </summary>
[Fact]
public async Task An_abandoned_initialise_overlapping_shutdown_leaves_nothing_running()
{
var wedge = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var first = true;
var factory = new FakeTwinCATClientFactory
{
Customise = () =>
{
var c = new FakeTwinCATClient { BlockConnect = first ? wedge : null };
first = false;
return c;
},
};
var drv = NewDriver(factory);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
await WaitForAsync(() => factory.Clients.Count == 1, TimeSpan.FromSeconds(2));
var abandoned = drv.InitializeAsync("{}", CancellationToken.None); // joins the wedged generation-1 probe
await Task.Delay(50, TestContext.Current.CancellationToken);
var sw = Stopwatch.StartNew();
await drv.ShutdownAsync(TestContext.Current.CancellationToken).WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
sw.Elapsed.ShouldBeLessThan(TwinCATDriver.ShutdownBudget + TimeSpan.FromMilliseconds(500));
wedge.SetResult();
try { await abandoned.WaitAsync(TimeSpan.FromSeconds(5)); } catch (OperationCanceledException) { }
drv.DeviceCount.ShouldBe(0);
await WaitForAsync(() => factory.Clients.All(c => c.DisposeCount > 0 || !c.IsConnected), TimeSpan.FromSeconds(2));
var settled = factory.Clients.Count;
await Task.Delay(150, TestContext.Current.CancellationToken);
factory.Clients.Count.ShouldBe(settled, "no probe loop may still be creating clients");
factory.Clients.ShouldAllBe(c => c.DisposeCount > 0 || !c.IsConnected);
}
/// <summary>
/// A value the client queued before the unsubscribe is dropped if dispatched after it.
/// Positive control: the same callback delivers while subscribed.
/// </summary>
[Fact]
public async Task A_notification_dispatched_after_unsubscribe_is_dropped()
{
var factory = new FakeTwinCATClientFactory();
var drv = NewDriver(factory, probe: false);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
var events = new ConcurrentQueue<DataChangeEventArgs>();
drv.OnDataChange += (_, e) => events.Enqueue(e);
var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
var callback = factory.Clients[0].Notifications.Single().OnChange; // what the client's queue holds
callback("MAIN.Speed", 1);
events.Count.ShouldBe(1);
await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
callback("MAIN.Speed", 2);
events.Count.ShouldBe(1);
}
/// <summary>
/// A direct re-initialise with a changed config swaps the tag table (the removed tag is unknown, the
/// added one resolves) and retracts the previous generation's live notifications.
/// </summary>
[Fact]
public async Task Reinitialise_swaps_tables_and_retracts_the_previous_generations_notifications()
{
var factory = new FakeTwinCATClientFactory();
var drv = NewDriver(factory, probe: false);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
_ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken);
factory.Clients[0].Notifications.Count.ShouldBe(1);
var next = System.Text.Json.JsonSerializer.Serialize(new
{
devices = new[] { new { hostAddress = Host } },
rawTags = new[]
{
new
{
rawPath = "Torque",
tagConfig = TwinCATTagDefinitionFactory.ToTagConfig(
new TwinCATTagDefinition("Torque", Host, "MAIN.Torque", TwinCATDataType.DInt)),
writeIdempotent = false,
deviceName = Host,
},
},
probe = new { enabled = false },
});
await drv.InitializeAsync(next, TestContext.Current.CancellationToken);
factory.Clients[0].Notifications.ShouldBeEmpty();
factory.Clients[0].DisposeCount.ShouldBe(1);
(await drv.ReadAsync(["Speed"], TestContext.Current.CancellationToken)).Single().StatusCode
.ShouldBe(TwinCATStatusMapper.BadNodeIdUnknown);
(await drv.ReadAsync(["Torque"], TestContext.Current.CancellationToken)).Single().StatusCode
.ShouldBe(TwinCATStatusMapper.Good);
}
/// <summary>Reads racing a stream of re-initialises never throw (no table is ever mid-Clear).</summary>
[Fact]
public async Task Reads_racing_reinitialise_never_throw()
{
var factory = new FakeTwinCATClientFactory();
var drv = NewDriver(factory, probe: false);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
using var stop = new CancellationTokenSource(TimeSpan.FromMilliseconds(400));
var good = 0;
var reader = Task.Run(async () =>
{
while (!stop.IsCancellationRequested)
{
var r = await drv.ReadAsync(["Speed"], CancellationToken.None);
if (r[0].StatusCode == TwinCATStatusMapper.Good) Interlocked.Increment(ref good);
}
}, TestContext.Current.CancellationToken);
while (!stop.IsCancellationRequested)
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
await reader; // throws if any read threw
good.ShouldBeGreaterThan(0);
}
/// <summary>Notification-queue drops are reported once per probe tick that sees new drops.</summary>
[Fact]
public async Task Notification_queue_drops_are_reported()
{
var fake = new FakeTwinCATClient();
var log = new TwinCATCapturingLogger();
var drv = NewDriver(new FakeTwinCATClientFactory { Customise = () => fake }, log: log);
await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
await WaitForAsync(() => fake.IsConnected, TimeSpan.FromSeconds(2));
await Task.Delay(100, TestContext.Current.CancellationToken);
log.Warnings.Count(w => w.Contains("dropped")).ShouldBe(0); // no drops, no warning
fake.DroppedNotificationCount = 5;
await WaitForAsync(() => log.Warnings.Any(w => w.Contains("dropped 5")), TimeSpan.FromSeconds(2));
await Task.Delay(100, TestContext.Current.CancellationToken);
log.Warnings.Count(w => w.Contains("dropped")).ShouldBe(1);
fake.DroppedNotificationCount = 8;
await WaitForAsync(() => log.Warnings.Any(w => w.Contains("dropped 3")), TimeSpan.FromSeconds(2));
log.Warnings.Count(w => w.Contains("dropped")).ShouldBe(2);
await drv.ShutdownAsync(TestContext.Current.CancellationToken);
}
}