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 9996bc1d..713eb27f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs @@ -87,6 +87,25 @@ public sealed class ModbusDriver private CancellationTokenSource? _probeCts; private CancellationTokenSource? _reprobeCts; + /// + /// Guards / / + /// / against each other. The three + /// fields above are published by one lifecycle operation and released by the next, so two + /// concurrent operations would dispose each other's freshly-built transport or null + /// _transport out from under an in-flight connect. Not hypothetical: + /// DriverInstanceActor dispatches initialize on a fire-and-forget Task.Run and + /// its AdoptConfigDuringInit can fire twice back to back. Same shape as + /// MqttDriver._lifecycleGate. + /// + private readonly SemaphoreSlim _lifecycleGate = new(1, 1); + + /// + /// Bounded wait the teardown paths take on . A wedged initialize + /// (a connect that never returns) must not turn shutdown or dispose into a hang — past the + /// timeout they tear down anyway, which is the lesser evil of the two. + /// + private static readonly TimeSpan LifecycleGateTimeout = TimeSpan.FromSeconds(10); + private DriverHealth _health = new(DriverState.Unknown, null, null); /// Occurs when a subscribed tag value changes. @@ -251,6 +270,20 @@ public sealed class ModbusDriver /// public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + { + await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); + } + finally + { + _lifecycleGate.Release(); + } + } + + /// The gate-held body of . + private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken) { WriteHealth(new DriverHealth(DriverState.Initializing, null, null)); try @@ -261,10 +294,11 @@ public sealed class ModbusDriver // background loops that no later dispose could reach. Reachable without any teardown of // ours: DriverInstanceActor re-runs InitializeAsync with no ShutdownAsync when an ApplyDelta // lands while the driver is in Connecting/Reconnecting (AdoptConfigDuringInit). - // Deliberately NOT the shared TeardownAsync(): that also disposes _poll, the PollGroupEngine - // constructed once in the ctor and never rebuilt, which would leave the re-initialized - // driver unable to serve subscriptions. A first initialize sees three nulls, so this is a - // no-op on the initial call. + // Deliberately NOT the shared TeardownAsync(): that also disposes _poll, which tears down + // every ACTIVE SUBSCRIPTION — and nothing re-subscribes them afterwards, because a + // mid-life re-initialize is not a shutdown the OPC UA layer knows to re-drive. A caller + // that genuinely wants the subscriptions gone calls ShutdownAsync. A first initialize + // sees three nulls, so this block is a no-op on the initial call. try { _probeCts?.Cancel(); } catch { } _probeCts?.Dispose(); _probeCts = null; @@ -333,12 +367,48 @@ public sealed class ModbusDriver /// public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { - await ShutdownAsync(cancellationToken); - await InitializeAsync(driverConfigJson, cancellationToken); + // Holds the gate across BOTH halves and calls the ungated cores: taking it once here is what + // makes the shutdown-then-initialize pair atomic. Routing through the public (gated) methods + // instead would both self-deadlock and leave a window between them for a concurrent + // initialize to slip into. + await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await ShutdownCoreAsync().ConfigureAwait(false); + await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); + } + finally + { + _lifecycleGate.Release(); + } } /// public async Task ShutdownAsync(CancellationToken cancellationToken) + { + if (await _lifecycleGate.WaitAsync(LifecycleGateTimeout, cancellationToken).ConfigureAwait(false)) + { + try + { + await ShutdownCoreAsync().ConfigureAwait(false); + } + finally + { + _lifecycleGate.Release(); + } + } + else + { + _logger.LogWarning( + "Modbus lifecycle gate still held at shutdown; tearing the transport down anyway. Driver={DriverInstanceId}", + _driverInstanceId); + await ShutdownCoreAsync().ConfigureAwait(false); + } + } + + /// The gate-held body of , shared with + /// . + private async Task ShutdownCoreAsync() { var lastRead = ReadHealth().LastSuccessfulRead; await TeardownAsync().ConfigureAwait(false); @@ -1693,6 +1763,23 @@ public sealed class ModbusDriver /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() { + if (await _lifecycleGate.WaitAsync(LifecycleGateTimeout).ConfigureAwait(false)) + { + try + { + await TeardownAsync().ConfigureAwait(false); + } + finally + { + _lifecycleGate.Release(); + } + + return; + } + + _logger.LogWarning( + "Modbus lifecycle gate still held at dispose; tearing the transport down anyway. Driver={DriverInstanceId}", + _driverInstanceId); await TeardownAsync().ConfigureAwait(false); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLifecycleHygieneTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLifecycleHygieneTests.cs index b67adce0..f356156e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLifecycleHygieneTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLifecycleHygieneTests.cs @@ -264,12 +264,13 @@ public sealed class ModbusLifecycleHygieneTests } /// - /// The probe / re-probe CancellationTokenSources are re-assigned by every initialize, so a - /// second initialize must cancel the previous pair or their background loops run forever - /// against a transport that is already gone. + /// _probeCts is re-assigned by every initialize, so a second initialize must cancel the + /// previous one. ProbeLoopAsync CAPTURES its transport reference at loop entry, so a + /// leaked probe loop keeps hammering the transport the first initialize published — which is + /// exactly what this measures. Re-probe is off here so every send is attributable to the probe. /// [Fact] - public async Task A_second_InitializeAsync_stops_the_first_probe_and_reprobe_loops() + public async Task A_second_InitializeAsync_stops_the_first_probe_loop() { var first = new FakeTransport(); var second = new FakeTransport(); @@ -280,10 +281,10 @@ public sealed class ModbusLifecycleHygieneTests Probe = new ModbusProbeOptions { Enabled = true, - Interval = TimeSpan.FromMilliseconds(50), + Interval = TimeSpan.FromMilliseconds(25), Timeout = TimeSpan.FromSeconds(1), }, - AutoProhibitReprobeInterval = TimeSpan.FromMilliseconds(50), + AutoProhibitReprobeInterval = null, // re-probe off: isolate the probe loop RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("A", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16)]), }; var drv = new ModbusDriver(opts, "modbus-1", _ => Interlocked.Increment(ref handed) == 1 ? first : second); @@ -291,16 +292,164 @@ public sealed class ModbusLifecycleHygieneTests await drv.InitializeAsync("{}", CancellationToken.None); await Task.Delay(200); Interlocked.CompareExchange(ref first.SendCount, 0, 0).ShouldBeGreaterThan(0, - "the first initialize's probe loop should have issued at least one send"); + "precondition: the first initialize's probe loop must actually be sending"); await drv.InitializeAsync("{}", CancellationToken.None); - // The first transport's loops must stop; tolerate one in-flight send straddling the cancel. + // Tolerate one in-flight send straddling the cancel; a leaked loop produces ~16 in 400ms. var atReinit = Interlocked.CompareExchange(ref first.SendCount, 0, 0); await Task.Delay(400); var atRest = Interlocked.CompareExchange(ref first.SendCount, 0, 0); (atRest - atReinit).ShouldBeLessThanOrEqualTo(1, - "the previous initialize's probe/re-probe loops must be cancelled by the next initialize"); + "the previous initialize's probe loop must be cancelled by the next initialize"); + + await drv.DisposeAsync(); + } + + /// + /// Transport for the re-probe test: counts every send and fails one address permanently, so a + /// seeded prohibition on that address is never cleared and the re-probe loop keeps re-probing + /// it tick after tick. A transport that answered normally would let the very first re-probe + /// CLEAR the prohibition, emptying the candidate set and making the test vacuous. + /// + private sealed class ReprobeCountingTransport : IModbusTransport + { + /// Register address whose reads always fail. + public ushort FailAddress { get; init; } + /// Total sends this transport has been asked to perform. + public int SendCount; + + /// Establishes a connection asynchronously. + /// Cancellation token for the operation. + public Task ConnectAsync(CancellationToken ct) => Task.CompletedTask; + + /// Counts the send, failing reads aimed at . + /// The Modbus unit identifier. + /// The Protocol Data Unit to send. + /// Cancellation token for the operation. + public Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) + { + Interlocked.Increment(ref SendCount); + var addr = (ushort)((pdu[1] << 8) | pdu[2]); + if (addr == FailAddress) + return Task.FromException(new IOException($"address {addr} is unreadable")); + return Task.FromResult(new byte[] { pdu[0], 2, 0, 0 }); + } + + /// Disposes the transport asynchronously. + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + /// Reads a private instance field off the driver. + private static object GetPrivateField(ModbusDriver drv, string name) => + typeof(ModbusDriver).GetField(name, BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(drv)!; + + /// + /// _reprobeCts is re-assigned by every initialize, so a second initialize must cancel the + /// previous one. Two things make this measurable where the probe test's approach would not: + /// the re-probe loop only sends when the prohibition set is NON-EMPTY (so one is seeded, on an + /// address the transport fails permanently), and it re-reads the _transport FIELD every + /// pass rather than capturing it — so a leaked loop hits the transport the SECOND initialize + /// published. The second initialize is therefore given re-probe-less options: any send at all on + /// the second transport is the leaked loop from the first. + /// + [Fact] + public async Task A_second_InitializeAsync_stops_the_first_reprobe_loop() + { + const ushort prohibited = 100; + var first = new ReprobeCountingTransport { FailAddress = prohibited }; + var second = new ReprobeCountingTransport { FailAddress = prohibited }; + var handed = 0; + var opts = new ModbusDriverOptions + { + Host = "fake", + Probe = new ModbusProbeOptions { Enabled = false }, // probe off: isolate the re-probe loop + AutoProhibitReprobeInterval = TimeSpan.FromMilliseconds(25), + RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("A", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16)]), + }; + var drv = new ModbusDriver(opts, "modbus-1", _ => Interlocked.Increment(ref handed) == 1 ? first : second); + + // Seed a single-register prohibition BEFORE the first initialize, so no loop is running while + // the (unsynchronised) dictionary is mutated. Start == End keeps it off the bisection path, so + // each tick is exactly one StraightReprobeAsync send. + var prohibitions = (System.Collections.IDictionary)GetPrivateField(drv, "_autoProhibited"); + var stateType = typeof(ModbusDriver).GetNestedType("ProhibitionState", BindingFlags.NonPublic)!; + prohibitions.Add( + ((byte)1, ModbusRegion.HoldingRegisters, prohibited, prohibited), + Activator.CreateInstance(stateType, nonPublic: true)!); + + await drv.InitializeAsync("{}", CancellationToken.None); + await Task.Delay(200); + Interlocked.CompareExchange(ref first.SendCount, 0, 0).ShouldBeGreaterThan(0, + "precondition: the seeded prohibition must make the re-probe loop actually send"); + + // The second initialize must NOT start a re-probe loop of its own, or a leaked first loop and a + // healthy second one would be indistinguishable by send count. "{}" keeps these options. + typeof(ModbusDriver).GetField("_options", BindingFlags.NonPublic | BindingFlags.Instance)! + .SetValue(drv, new ModbusDriverOptions + { + Host = "fake", + Probe = new ModbusProbeOptions { Enabled = false }, + AutoProhibitReprobeInterval = null, + RawTags = opts.RawTags, + }); + + await drv.InitializeAsync("{}", CancellationToken.None); + + // Non-vacuity guard: the candidate set the leaked loop would iterate must still be non-empty, + // so a zero send count is attributable to the cancel and not to an emptied prohibition set. + prohibitions.Count.ShouldBe(1, "a re-initialize must not clear the prohibition set"); + + var atReinit = Interlocked.CompareExchange(ref second.SendCount, 0, 0); + await Task.Delay(400); + var atRest = Interlocked.CompareExchange(ref second.SendCount, 0, 0); + (atRest - atReinit).ShouldBeLessThanOrEqualTo(1, + "the previous initialize's re-probe loop must be cancelled by the next initialize"); + + await drv.DisposeAsync(); + } + + /// + /// Two initializes racing on the same driver — the shape DriverInstanceActor produces by + /// dispatching initialize on a fire-and-forget Task.Run with AdoptConfigDuringInit + /// firing twice — must serialize on the lifecycle gate. Ungated, the teardown at the top of one + /// initialize disposes the transport the other just published (or nulls _transport + /// mid-connect), leaving the driver holding a disposed transport. + /// + [Fact] + public async Task Concurrent_InitializeAsync_calls_serialize_and_leave_exactly_one_live_transport() + { + var handed = new ConcurrentQueue(); + var opts = new ModbusDriverOptions + { + Host = "fake", + Probe = new ModbusProbeOptions { Enabled = false }, + RawTags = ModbusRawTags.Entries([new ModbusTagDefinition("A", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16)]), + }; + var drv = new ModbusDriver(opts, "modbus-1", _ => + { + var t = new FakeTransport(); + handed.Enqueue(t); + return t; + }); + + await Task.WhenAll( + Task.Run(() => drv.InitializeAsync("{}", CancellationToken.None)), + Task.Run(() => drv.InitializeAsync("{}", CancellationToken.None))); + + var built = handed.ToArray(); + built.Length.ShouldBe(2); + // Serialized: the earlier initialize's transport is disposed by the later one, and exactly one + // survives — never two live transports, and never a disposed transport left installed. + built.Count(t => t.DisposeCount == 0).ShouldBe(1, "exactly one transport must be left live"); + built.Count(t => t.DisposeCount == 1).ShouldBe(1, "the superseded transport must be disposed once"); + + // The survivor is the installed one and the driver still works through it. + var live = built.Single(t => t.DisposeCount == 0); + live.HoldingRegisters[0] = 9; + var r = await drv.ReadAsync(["A"], CancellationToken.None); + r[0].StatusCode.ShouldBe(0u); + Convert.ToInt32(r[0].Value).ShouldBe(9); await drv.DisposeAsync(); }