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

fix(modbus): lifecycle gate + non-vacuous reprobe regression test

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 04:07:27 -04:00
parent 140bf0c131
commit 409a58b624
2 changed files with 251 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs
+93 -6
View File
@@ -87,6 +87,25 @@ public sealed class ModbusDriver
private CancellationTokenSource? _probeCts;
private CancellationTokenSource? _reprobeCts;
/// <summary>
/// Guards <see cref="InitializeAsync"/> / <see cref="ReinitializeAsync"/> /
/// <see cref="ShutdownAsync"/> / <see cref="DisposeAsync"/> 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
/// <c>_transport</c> out from under an in-flight connect. Not hypothetical:
/// <c>DriverInstanceActor</c> dispatches initialize on a fire-and-forget <c>Task.Run</c> and
/// its <c>AdoptConfigDuringInit</c> can fire twice back to back. Same shape as
/// <c>MqttDriver._lifecycleGate</c>.
/// </summary>
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
/// <summary>
/// Bounded wait the teardown paths take on <see cref="_lifecycleGate"/>. 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.
/// </summary>
private static readonly TimeSpan LifecycleGateTimeout = TimeSpan.FromSeconds(10);
private DriverHealth _health = new(DriverState.Unknown, null, null);
/// <summary>Occurs when a subscribed tag value changes.</summary>
@@ -251,6 +270,20 @@ public sealed class ModbusDriver
/// <inheritdoc />
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
/// <summary>The gate-held body of <see cref="InitializeAsync"/>.</summary>
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
/// <inheritdoc />
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();
}
}
/// <inheritdoc />
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);
}
}
/// <summary>The gate-held body of <see cref="ShutdownAsync"/>, shared with
/// <see cref="ReinitializeAsync"/>.</summary>
private async Task ShutdownCoreAsync()
{
var lastRead = ReadHealth().LastSuccessfulRead;
await TeardownAsync().ConfigureAwait(false);
@@ -1693,6 +1763,23 @@ public sealed class ModbusDriver
/// <returns>A task that represents the asynchronous operation.</returns>
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);
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLifecycleHygieneTests.cs
+158 -9
View File
@@ -264,12 +264,13 @@ public sealed class ModbusLifecycleHygieneTests
}
/// <summary>
/// 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.
/// <c>_probeCts</c> is re-assigned by every initialize, so a second initialize must cancel the
/// previous one. <c>ProbeLoopAsync</c> 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.
/// </summary>
[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();
}
/// <summary>
/// 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.
/// </summary>
private sealed class ReprobeCountingTransport : IModbusTransport
{
/// <summary>Register address whose reads always fail.</summary>
public ushort FailAddress { get; init; }
/// <summary>Total sends this transport has been asked to perform.</summary>
public int SendCount;
/// <summary>Establishes a connection asynchronously.</summary>
/// <param name="ct">Cancellation token for the operation.</param>
public Task ConnectAsync(CancellationToken ct) => Task.CompletedTask;
/// <summary>Counts the send, failing reads aimed at <see cref="FailAddress"/>.</summary>
/// <param name="unitId">The Modbus unit identifier.</param>
/// <param name="pdu">The Protocol Data Unit to send.</param>
/// <param name="ct">Cancellation token for the operation.</param>
public Task<byte[]> 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<byte[]>(new IOException($"address {addr} is unreadable"));
return Task.FromResult(new byte[] { pdu[0], 2, 0, 0 });
}
/// <summary>Disposes the transport asynchronously.</summary>
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}
/// <summary>Reads a private instance field off the driver.</summary>
private static object GetPrivateField(ModbusDriver drv, string name) =>
typeof(ModbusDriver).GetField(name, BindingFlags.NonPublic | BindingFlags.Instance)!.GetValue(drv)!;
/// <summary>
/// <c>_reprobeCts</c> 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 <c>_transport</c> 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.
/// </summary>
[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();
}
/// <summary>
/// Two initializes racing on the same driver — the shape <c>DriverInstanceActor</c> produces by
/// dispatching initialize on a fire-and-forget <c>Task.Run</c> with <c>AdoptConfigDuringInit</c>
/// 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 <c>_transport</c>
/// mid-connect), leaving the driver holding a disposed transport.
/// </summary>
[Fact]
public async Task Concurrent_InitializeAsync_calls_serialize_and_leave_exactly_one_live_transport()
{
var handed = new ConcurrentQueue<FakeTransport>();
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();
}