diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs index 36cea8d1..5c8a3599 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs @@ -562,19 +562,31 @@ public sealed class ModbusDriver // Running transition before any register round-trip has happened. Started inside the // generation-checked publish so a teardown that bypassed the lifecycle gate can never // run between the publish and the loop start and strand a loop. + // + // #729: each loop's CancellationToken VALUE is snapshotted here, before scheduling, and the + // worker delegate closes over that value — never over the CTS (or the mutable _probeCts / + // _reprobeCts fields). The scheduling token only suppresses a still-queued task; once the + // pool has committed to running the delegate, teardown can complete underneath it (the + // bounded join in StopBackgroundLoopsAsync gives up and disposes the CTS). A delegate that + // then read cts.Token would throw ObjectDisposedException; one that read the field would + // see null or adopt the NEXT initialization's token. A token value stays valid after its + // source is disposed and still reports the cancellation, so the late worker exits cleanly + // on its own generation's cancelled token. if (_options.Probe.Enabled) { var cts = new CancellationTokenSource(); + var probeToken = cts.Token; var probeTransport = transport; _probeCts = cts; - _probeTask = Task.Run(() => ProbeLoopAsync(generation, probeTransport, cts.Token), cts.Token); + _probeTask = Task.Run(() => ProbeLoopAsync(generation, probeTransport, probeToken), probeToken); } if (_options.AutoProhibitReprobeInterval is not null) { var cts = new CancellationTokenSource(); + var reprobeToken = cts.Token; _reprobeCts = cts; - _reprobeTask = Task.Run(() => ReprobeLoopAsync(cts.Token), cts.Token); + _reprobeTask = Task.Run(() => ReprobeLoopAsync(reprobeToken), reprobeToken); } } transport = null; // published — owned by the driver now @@ -592,6 +604,16 @@ public sealed class ModbusDriver } /// Cancels, bounded-joins and disposes the probe and re-probe loops. Idempotent. + /// + /// Limits (#729): the join is bounded by , so this does NOT + /// guarantee a loop has finished when it returns — a loop stuck in a transport call that ignores its + /// token, or a worker the pool had committed to but not yet invoked, is abandoned, not awaited. Such + /// a worker holds its own already-cancelled token value (never the CTS or the fields), so it exits on + /// its first token check without faulting. Transport lifetime is NOT bound to the loops: the caller + /// disposes the transport after this returns, so a probe send that ignores its token and outlives the + /// join completes (or faults) against a disposed transport; the loop then sees its cancelled token and + /// returns without a health transition. + /// private async Task StopBackgroundLoopsAsync() { var probeCts = _probeCts; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBackgroundLoopTokenLifetimeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBackgroundLoopTokenLifetimeTests.cs new file mode 100644 index 00000000..91cee001 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBackgroundLoopTokenLifetimeTests.cs @@ -0,0 +1,282 @@ +using System.Diagnostics.Tracing; +using System.Reflection; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Modbus; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// #729 — the probe worker delegate must not obtain its after it is +/// already running. The scheduling token passed to Task.Run only suppresses a task that is still +/// QUEUED; once the pool has passed the task's cancellation/start check, lifecycle teardown can finish +/// underneath it (the bounded join gives up and the CTS is disposed). A delegate that then reads +/// cts.Token throws ; one that reads the _probeCts +/// field sees null or the next generation's token. +/// +/// +/// The seam is TPL's own TaskStarted event: TplEventSource raises it synchronously on the +/// worker thread AFTER the start check and BEFORE the delegate is invoked, so holding the listener +/// callback parks the probe worker in exactly the window the issue describes — deterministically, no +/// timing stress. The target task is identified by the TaskScheduled event raised on the test's +/// own (AsyncLocal-armed) flow during InitializeAsync; more than one armed schedule fails closed. +/// Queued-task cancellation (shutdown before the pool reaches the start check) is not exercised here: +/// there is no deterministic seam ahead of the start check, and that path is the BCL's own behaviour +/// for the scheduling token, which the fix leaves unchanged. +/// +[Trait("Category", "Unit")] +public sealed class ModbusBackgroundLoopTokenLifetimeTests +{ + private static readonly TimeSpan Bound = TimeSpan.FromSeconds(15); + + private sealed class CountingTransport : IModbusTransport + { + public int SendCount; + public int DisposeCount; + + public Task ConnectAsync(CancellationToken ct) => Task.CompletedTask; + + public Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) + { + Interlocked.Increment(ref SendCount); + return Task.FromResult(new byte[] { pdu[0], 2, 0, 0 }); + } + + public ValueTask DisposeAsync() + { + Interlocked.Increment(ref DisposeCount); + return ValueTask.CompletedTask; + } + } + + /// Parks the one task scheduled on the armed flow inside its TaskStarted callback. + private sealed class TaskStartGate : EventListener + { + private const string TplSourceName = "System.Threading.Tasks.TplEventSource"; + private const EventKeywords TaskTransfer = (EventKeywords)0x1; + private const EventKeywords Tasks = (EventKeywords)0x2; + + public static readonly AsyncLocal Armed = new(); + + private readonly ManualResetEventSlim _release = new(false); + private int _targetTaskId = -1; + private int _armedSchedules; + private volatile bool _closed; + + public TaskCompletionSource Started { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public int ArmedSchedules => Volatile.Read(ref _armedSchedules); + public bool ReleaseTimedOut { get; private set; } + + /// Stops correlating new schedules once the arming operation has returned. + public void Close() => _closed = true; + + public void Release() => _release.Set(); + + protected override void OnEventSourceCreated(EventSource eventSource) + { + if (eventSource.Name == TplSourceName) + EnableEvents(eventSource, EventLevel.Informational, TaskTransfer | Tasks); + } + + protected override void OnEventWritten(EventWrittenEventArgs e) + { + if (e.EventName == "TaskScheduled") + { + if (_closed || !Armed.Value) return; + Interlocked.Increment(ref _armedSchedules); + Interlocked.CompareExchange(ref _targetTaskId, TaskId(e), -1); + } + else if (e.EventName == "TaskStarted") + { + var target = Volatile.Read(ref _targetTaskId); + if (target == -1 || TaskId(e) != target) return; + Started.TrySetResult(); + // Bounded: a test that never releases must not wedge a pool thread forever. + if (!_release.Wait(Bound)) ReleaseTimedOut = true; + } + } + + private static int TaskId(EventWrittenEventArgs e) + { + var i = e.PayloadNames!.IndexOf("TaskID"); + return i < 0 ? -2 : Convert.ToInt32(e.Payload![i]); + } + + public override void Dispose() + { + _release.Set(); + base.Dispose(); + } + } + + private static ModbusDriverOptions ProbeOnlyOptions() => new() + { + Host = "fake", + Probe = new ModbusProbeOptions + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(50), + Timeout = TimeSpan.FromSeconds(1), + }, + AutoProhibitReprobeInterval = null, + RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("A", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16)]), + }; + + private static T? Field(ModbusDriver drv, string name) where T : class => + (T?)typeof(ModbusDriver).GetField(name, BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(drv); + + /// Initializes on an armed flow and returns the parked background-loop task. + private static async Task InitializeAndParkProbeAsync(ModbusDriver drv, TaskStartGate gate, string taskField = "_probeTask") + { + TaskStartGate.Armed.Value = true; + try { await drv.InitializeAsync("{}", CancellationToken.None); } + finally + { + gate.Close(); + TaskStartGate.Armed.Value = false; + } + + gate.ArmedSchedules.ShouldBe(1, "exactly one task (the background loop) must be scheduled by initialize; anything else is ambiguous"); + var probeTask = Field(drv, taskField); + probeTask.ShouldNotBeNull(); + + await gate.Started.Task.WaitAsync(Bound, TestContext.Current.CancellationToken); + probeTask.IsCompleted.ShouldBeFalse("the worker must be parked after its start check, before its delegate runs"); + return probeTask; + } + + private static async Task SettleAsync(Task task) + { + try { await task.WaitAsync(Bound, TestContext.Current.CancellationToken); } + catch (TimeoutException) { throw; } + catch { /* outcome asserted by the caller from task.Status */ } + } + + /// + /// A probe worker already past its start check when ShutdownAsync tears the driver down must still + /// complete cleanly on its own cancelled token once it runs — not fault on a disposed CTS, and not + /// touch the disposed transport. + /// + [Fact] + public async Task Started_probe_worker_released_after_shutdown_exits_cleanly_on_its_own_cancelled_token() + { + var fake = new CountingTransport(); + var drv = new ModbusDriver(ProbeOnlyOptions(), "modbus-729", _ => fake); + using var gate = new TaskStartGate(); + try + { + var probeTask = await InitializeAndParkProbeAsync(drv, gate); + + // Teardown runs to completion while the worker is parked: the bounded join times out and the + // CTS is disposed and the field cleared before the delegate has read anything. + await drv.ShutdownAsync(CancellationToken.None); + Field(drv, "_probeCts").ShouldBeNull(); + fake.DisposeCount.ShouldBe(1); + probeTask.IsCompleted.ShouldBeFalse(); + + gate.Release(); + await SettleAsync(probeTask); + + gate.ReleaseTimedOut.ShouldBeFalse(); + probeTask.Exception?.Flatten().InnerException.ShouldBeNull( + "the late worker must not fault obtaining its token after teardown disposed the CTS"); + probeTask.Status.ShouldBe(TaskStatus.RanToCompletion); + Volatile.Read(ref fake.SendCount).ShouldBe(0, "a worker released after teardown must not probe the disposed transport"); + } + finally + { + gate.Release(); + await drv.DisposeAsync(); + } + } + + /// + /// A probe worker of generation N released after a replacement initialize must run on generation + /// N's (cancelled) token — not adopt generation N+1's live token and keep probing the N transport — + /// while generation N+1's own probe loop stays live. + /// + [Fact] + public async Task Started_probe_worker_released_after_replacement_keeps_its_own_generation_token() + { + var transports = new List(); + var drv = new ModbusDriver(ProbeOnlyOptions(), "modbus-729", _ => + { + var t = new CountingTransport(); + lock (transports) transports.Add(t); + return t; + }); + using var gate = new TaskStartGate(); + try + { + var firstProbe = await InitializeAndParkProbeAsync(drv, gate); + var first = transports[0]; + + await drv.ReinitializeAsync("{}", CancellationToken.None); + transports.Count.ShouldBe(2); + var second = transports[1]; + first.DisposeCount.ShouldBe(1); + var liveCts = Field(drv, "_probeCts"); + liveCts.ShouldNotBeNull(); + firstProbe.IsCompleted.ShouldBeFalse(); + + gate.Release(); + await SettleAsync(firstProbe); + + gate.ReleaseTimedOut.ShouldBeFalse(); + firstProbe.Exception?.Flatten().InnerException.ShouldBeNull(); + firstProbe.Status.ShouldBe(TaskStatus.RanToCompletion, + "the generation-1 worker must exit on its own cancelled token"); + Volatile.Read(ref first.SendCount).ShouldBe(0, "the generation-1 worker must not probe its disposed transport"); + + // Generation 2 is unaffected: its loop is live and its CTS is still the published, uncancelled one. + var deadline = DateTime.UtcNow + Bound; + while (Volatile.Read(ref second.SendCount) == 0 && DateTime.UtcNow < deadline) + await Task.Delay(10, TestContext.Current.CancellationToken); + Volatile.Read(ref second.SendCount).ShouldBeGreaterThan(0); + Field(drv, "_probeCts").ShouldBeSameAs(liveCts); + liveCts!.IsCancellationRequested.ShouldBeFalse(); + } + finally + { + gate.Release(); + await drv.DisposeAsync(); + } + } + + /// The re-probe loop gets the same guarantee as the probe loop. + [Fact] + public async Task Started_reprobe_worker_released_after_shutdown_exits_cleanly_on_its_own_cancelled_token() + { + var fake = new CountingTransport(); + var opts = new ModbusDriverOptions + { + Host = "fake", + Probe = new ModbusProbeOptions { Enabled = false }, + AutoProhibitReprobeInterval = TimeSpan.FromMilliseconds(50), + RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("A", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16)]), + }; + var drv = new ModbusDriver(opts, "modbus-729", _ => fake); + using var gate = new TaskStartGate(); + try + { + var reprobeTask = await InitializeAndParkProbeAsync(drv, gate, "_reprobeTask"); + + await drv.ShutdownAsync(CancellationToken.None); + Field(drv, "_reprobeCts").ShouldBeNull(); + reprobeTask.IsCompleted.ShouldBeFalse(); + + gate.Release(); + await SettleAsync(reprobeTask); + + gate.ReleaseTimedOut.ShouldBeFalse(); + reprobeTask.Exception?.Flatten().InnerException.ShouldBeNull( + "the late re-probe worker must not fault obtaining its token after teardown disposed the CTS"); + reprobeTask.Status.ShouldBe(TaskStatus.RanToCompletion); + } + finally + { + gate.Release(); + await drv.DisposeAsync(); + } + } +}