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

fix(s7): lifecycle gate + generation-stamped health and publishes (R07:H5)

Browse Source 
Overlapping InitializeAsync calls (the actor's fire-and-forget init plus a
retry tick) could leak an opened PLC connection and orphan a probe loop: no
gate guarded the lifecycle, only _gate for I/O.

- _lifecycleGate (the Modbus pattern) serialises Initialize / Reinitialize /
  Shutdown / Dispose; teardown waits a bounded time, then tears down anyway.
- Every initialize and teardown starts a lifecycle generation. Health writes
  (init, read/write passes, poll errors, shutdown) and the init publish carry
  it and are dropped once it has moved on, so an abandoned init never
  publishes over a later shutdown and a shutdown's final write never clobbers
  a newer init. A superseded init disposes what it opened and throws.
- The init publish (options, tag tables, Plc, probe loop) happens under _gate;
  the tag tables are built fresh and swapped by one reference write instead of
  being cleared in place under live poll passes.
- Read/write passes re-check initialization under _gate, so a pass that waited
  out a teardown cannot reopen a connection on a shut-down driver.
- The semaphores are no longer disposed: Dispose can race a running shutdown.

Refs #742
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:10:21 -04:00
parent 87a7414df7
commit ecf1e86464
2 changed files with 467 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+324 -115
View File
@@ -59,7 +59,7 @@ public sealed class S7Driver
_plcFactory = plcFactory ?? new S7PlcFactory();
_logger = logger ?? NullLogger<S7Driver>.Instance;
_resolver = new EquipmentTagRefResolver<S7TagDefinition>(
r => _tagsByRawPath.TryGetValue(r, out var t) ? t : null);
r => TagsByRawPath.TryGetValue(r, out var t) ? t : null);
// The polling overlay runs on the shared PollGroupEngine (structural array diff +
// drain-before-dispose + capped-exponential failure backoff + caller-token-filtered OCE),
// retiring the bespoke S7 poll fork. minInterval 100 ms matches the S7 scan-mailbox floor;
@@ -128,13 +128,6 @@ public sealed class S7Driver
/// <summary>OPC UA StatusCode used for a genuine device fault (CPU error, hardware fault).</summary>
private const uint StatusBadDeviceFailure = 0x808B0000u;
// v3: the RawPath → definition table. Keyed by RawPath (the v3 tag identity, == def.Name),
// case-sensitive ordinal. Built in InitializeAsync from _options.RawTags via the pure factory.
private readonly Dictionary<string, S7TagDefinition> _tagsByRawPath = new(StringComparer.Ordinal);
// Pre-parsed S7 address per tag, keyed by RawPath (== def.Name). Populated alongside
// _tagsByRawPath at init so a config typo fails fast here rather than as a per-read fault.
private readonly Dictionary<string, S7ParsedAddress> _parsedByRawPath = new(StringComparer.Ordinal);
// Resolves a read/write/subscribe fullReference (a RawPath in v3) to a tag definition by a single
// authored-table hit; a miss is a miss (the driver surfaces BadNodeIdUnknown, never a parse attempt).
private readonly EquipmentTagRefResolver<S7TagDefinition> _resolver;
@@ -225,19 +218,50 @@ public sealed class S7Driver
/// <inheritdoc />
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract
// allows it, and a reconnect retry can overlap a running attempt). Stop the previous probe loop and drop the previous connection
// before they are overwritten below. Deliberately NOT ShutdownAsync: that also drops _poll and the
// subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseConnectionAsync().ConfigureAwait(false);
Health = new DriverHealth(DriverState.Initializing, null, null);
// Serialised against every other lifecycle operation: DriverInstanceActor dispatches initialize
// on a fire-and-forget Task.Run and a reconnect retry can start a second attempt while the
// first is still inside OpenAsync. Without the gate attempt A's freshly opened Plc and probe
// loop were overwritten by attempt B and never released (R07:H5) — the Modbus pattern.
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
/// <summary>The lifecycle-gated body of <see cref="InitializeAsync"/>.</summary>
/// <param name="driverConfigJson">The driver config.</param>
/// <param name="cancellationToken">Caller cancellation.</param>
/// <returns>A task that completes when the attempt has published, been superseded, or failed.</returns>
private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// This attempt's generation. Every health write and table/connection publish below is stamped
// with it and dropped once the generation has moved on: an attempt abandoned by its caller can
// still be running when a ShutdownAsync (which only waits a bounded time for the lifecycle
// gate) starts, and it must never publish over it.
var gen = BeginGeneration();
// InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract
// allows it). Stop the previous probe loop and drop the previous connection before they are
// replaced below. Deliberately NOT a full teardown: that also drops _poll and the subscriptions,
// which are owned by the shutdown path (the actor re-subscribes after every successful init).
await ReleaseConnectionAsync(gen).ConfigureAwait(false);
TryWriteHealth(gen, new DriverHealth(DriverState.Initializing, null, null));
// Re-parse the supplied DriverConfig JSON so a config change delivered through the IDriver
// contract is honoured. An empty / placeholder document (e.g. the "{}" some unit tests pass)
// keeps the constructor-supplied options. Local until published: a superseded attempt must
// not leave its options behind for the live connection to reopen against.
var options = _options;
IS7Plc? plc = null;
try
{
// Re-parse the supplied DriverConfig JSON so a config change delivered through the
// IDriver contract is honoured. An empty / placeholder document
// (e.g. the "{}" some unit tests pass) keeps the constructor-supplied options.
if (HasConfigBody(driverConfigJson))
_options = S7DriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson);
options = S7DriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson);
// v3: build the RawPath → definition table from the authored raw tags. Each entry's TagConfig
// blob is mapped by the pure factory, which also applies S7TagShapeRules (the address must
@@ -246,36 +270,32 @@ public sealed class S7Driver
// through ISkippedTagReporter, reads BadNodeIdUnknown), never thrown: these used to be init
// guards that faulted every tag on the PLC and made the actor retry the same init forever.
// The entry's WriteIdempotent flag is threaded onto the def (it lives on the RawTagEntry).
_tagsByRawPath.Clear();
_parsedByRawPath.Clear();
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
_skippedTags.BeginRebuild(_options.RawTags.Count);
foreach (var entry in _options.RawTags)
// Built into fresh dictionaries and published by ONE reference swap — a retained poll pass
// reading the old table never sees it half-cleared.
var tags = new Dictionary<string, S7TagDefinition>(StringComparer.Ordinal);
var parsed = new Dictionary<string, S7ParsedAddress>(StringComparer.Ordinal);
var skipped = new List<(string RawPath, string Reason)>();
foreach (var entry in options.RawTags)
{
if (S7TagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var def, out var shapeRejection)
&& S7AddressParser.TryParse(def.Address, out var parsedAddress))
{
_tagsByRawPath[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent };
_parsedByRawPath[entry.RawPath] = parsedAddress;
tags[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent };
parsed[entry.RawPath] = parsedAddress;
}
else
{
var reason = shapeRejection ?? "TagConfig did not map to a definition";
_skippedTags.Skip(entry.RawPath, reason);
_logger.LogWarning(
"S7 tag skipped: {Reason} Driver={DriverInstanceId} RawPath={RawPath}",
reason, _driverInstanceId, entry.RawPath);
skipped.Add((entry.RawPath, shapeRejection ?? "TagConfig did not map to a definition"));
}
}
_skippedTags.Commit();
// Build the connection through the factory (the same seam EnsureConnectedAsync's reopen
// uses) so the read/write timeouts are baked in identically on connect and reconnect.
var plc = _plcFactory.Create(_options);
plc = _plcFactory.Create(options);
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.Timeout);
cts.CancelAfter(options.Timeout);
await plc.OpenAsync(cts.Token).ConfigureAwait(false);
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
@@ -283,88 +303,157 @@ public sealed class S7Driver
// The linked CancelAfter fired — a connect TIMEOUT, not a caller cancellation.
// Surface it as a TimeoutException so init-time and reconnect-time timeouts report
// the same exception type (for consistency); the outer catch still faults health.
try { plc.Dispose(); } catch { }
throw new TimeoutException(
$"S7 connect to {_options.Host}:{_options.Port} timed out after " +
$"{(int)_options.Timeout.TotalMilliseconds} ms.");
$"S7 connect to {options.Host}:{options.Port} timed out after " +
$"{(int)options.Timeout.TotalMilliseconds} ms.");
}
catch
// Publish everything this attempt built under _gate (no pass can be mid-flight against a
// half-published state) and only while this attempt is still the current generation.
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// Dispose the partially-constructed connection so a failed init doesn't leak it
// before the outer catch nulls Plc (Plc isn't assigned yet, so it can't).
try { plc.Dispose(); } catch { }
throw;
if (!IsCurrentGeneration(gen))
{
// The outer finally disposes the unpublished plc; the catch writes no health (the
// generation moved on). Thrown, not returned: a caller must never read "initialized"
// into an attempt that published nothing.
throw new InvalidOperationException(
"S7 initialize was superseded by a newer lifecycle operation before it published; " +
"its connection was discarded.");
}
_options = options;
Volatile.Write(ref _tables, new TagTables(tags, parsed));
_skippedTags.BeginRebuild(options.RawTags.Count);
foreach (var (rawPath, reason) in skipped)
{
_skippedTags.Skip(rawPath, reason);
_logger.LogWarning(
"S7 tag skipped: {Reason} Driver={DriverInstanceId} RawPath={RawPath}",
reason, _driverInstanceId, rawPath);
}
_skippedTags.Commit();
Plc = plc;
plc = null; // published — no longer this method's to dispose
_plcDead = false;
_connectBackoff.RecordSuccess();
Volatile.Write(ref _negotiatedPduSize, Plc.MaxPduSize);
_initialized = true;
TryWriteHealth(gen, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
// Kick off the probe loop once the connection is up. Initial HostState stays Unknown
// until the first probe tick succeeds — avoids broadcasting a premature Running
// transition before any PDU round-trip has happened. Tracked (not fire-and-forget) so
// a teardown can await it before disposing its CTS; Task.Run gets None so the handle
// is always awaitable — the loop's own token check handles cancellation.
if (options.Probe.Enabled)
{
var probeCts = new CancellationTokenSource();
_probeCts = probeCts;
_probeTask = Task.Run(() => ProbeLoopAsync(probeCts.Token), CancellationToken.None);
}
}
finally
{
_gate.Release();
}
Plc = plc;
_plcDead = false;
_connectBackoff.RecordSuccess();
Volatile.Write(ref _negotiatedPduSize, plc.MaxPduSize);
_initialized = true;
Health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
_logger.LogInformation("S7Driver connected. Driver={DriverInstanceId} Host={Host} CPU={CpuType} PduSize={PduSize} Tags={TagCount}",
_driverInstanceId, _options.Host, _options.CpuType, _negotiatedPduSize, _tagsByRawPath.Count);
// Kick off the probe loop once the connection is up. Initial HostState stays
// Unknown until the first probe tick succeeds — avoids broadcasting a premature
// Running transition before any PDU round-trip has happened.
if (_options.Probe.Enabled)
{
_probeCts = new CancellationTokenSource();
// Track the probe Task (not fire-and-forget) so ShutdownAsync can await it
// before disposing _gate / _probeCts. Pass None to Task.Run so
// the delegate always runs and the handle is always awaitable; the loop's own
// token check handles cancellation.
_probeTask = Task.Run(() => ProbeLoopAsync(_probeCts.Token), CancellationToken.None);
}
_driverInstanceId, options.Host, options.CpuType, _negotiatedPduSize, tags.Count);
}
catch (Exception ex)
{
// Clean up a partially-constructed Plc so a retry from the caller doesn't leak
// the TcpClient. Dispose is best-effort and idempotent.
try { Plc?.Dispose(); } catch { }
Plc = null;
Volatile.Write(ref _negotiatedPduSize, 0);
_initialized = false;
_plcDead = false;
Health = new DriverHealth(DriverState.Faulted, null, ex.Message);
_logger.LogError(ex, "S7Driver connect failed. Driver={DriverInstanceId} Host={Host}", _driverInstanceId, _options.Host);
// A superseded attempt reports nothing: the generation that replaced it owns health.
if (TryWriteHealth(gen, new DriverHealth(DriverState.Faulted, null, ex.Message)))
{
_options = options;
_logger.LogError(ex, "S7Driver connect failed. Driver={DriverInstanceId} Host={Host}", _driverInstanceId, options.Host);
}
throw;
}
finally
{
// A connection this attempt opened but did not publish (failed or superseded) is released
// here, so a retry from the caller never leaks the TcpClient.
if (plc is not null) { try { plc.Dispose(); } catch { /* best-effort */ } }
}
}
/// <inheritdoc />
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// InitializeAsync re-parses driverConfigJson, so a config change delivered here is
// Holds the lifecycle gate across BOTH halves so no other lifecycle operation slips between
// them. InitializeCoreAsync re-parses driverConfigJson, so a config change delivered here is
// applied in place rather than silently discarded.
await ShutdownAsync(cancellationToken).ConfigureAwait(false);
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
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)
{
// Signal cancellation to the probe + poll loops first, collect their Task handles,
// then await all of them with a bounded timeout BEFORE disposing the shared semaphore
// and CTS objects. Without the drain, a loop iteration mid-_gate would call Release()
// on (or WaitAsync against) a disposed semaphore.
// Bounded wait for the lifecycle gate: a wedged initialize (a connect that never returns, or an
// attempt its caller abandoned) must not turn shutdown into a hang. Past the bound we tear down
// anyway — the generation bump makes the still-running attempt discard instead of publish.
var gated = false;
try { gated = await _lifecycleGate.WaitAsync(LifecycleGateTimeout, cancellationToken).ConfigureAwait(false); }
catch (OperationCanceledException) { /* the caller stopped waiting for the gate — still tear down */ }
if (!gated)
_logger.LogWarning(
"S7 lifecycle gate still held at shutdown; tearing down anyway. Driver={DriverInstanceId}",
_driverInstanceId);
try
{
await ShutdownCoreAsync().ConfigureAwait(false);
}
finally
{
if (gated) _lifecycleGate.Release();
}
}
/// <summary>
/// The complete teardown: the poll engine (every subscription), the probe loop, the connection,
/// the sticky write fault and the per-generation memos. It is its own generation, so its final
/// health write can never clobber an initialize that started after it.
/// </summary>
/// <returns>A task that completes once everything is released.</returns>
private async Task ShutdownCoreAsync()
{
var gen = BeginGeneration();
// Drain the polling overlay first — the engine cancels each loop and awaits it before
// disposing its CTS (drain-before-dispose), so no poll iteration touches _gate
// after this returns.
await _poll.DisposeAsync().ConfigureAwait(false);
await ReleaseConnectionAsync().ConfigureAwait(false);
_writeFault = null;
_connectBackoff.RecordSuccess();
Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null);
// disposing its CTS (drain-before-dispose), so no poll iteration touches _gate after this.
try { await _poll.DisposeAsync().ConfigureAwait(false); }
catch (Exception ex) { _logger.LogWarning(ex, "S7 poll engine drain failed. Driver={DriverInstanceId}", _driverInstanceId); }
await ReleaseConnectionAsync(gen).ConfigureAwait(false);
lock (_healthLock)
{
if (_generation != gen) return; // a newer initialize owns the state now
_writeFault = null;
_connectBackoff.RecordSuccess();
Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null);
}
}
/// <summary>
/// Cancels + bounded-joins the probe loop, then disposes its CTS and the PLC connection. The
/// connection part of <see cref="ShutdownAsync"/> (everything but the poll engine and health).
/// connection part of a teardown (everything but the poll engine and health). Releases only
/// while <paramref name="gen"/> is still current — a teardown that ran past the lifecycle gate
/// must not dispose a connection a NEWER initialize already published.
/// </summary>
private async Task ReleaseConnectionAsync()
/// <param name="gen">The generation of the lifecycle operation doing the release.</param>
/// <returns>A task that completes once the prior state is released.</returns>
private async Task ReleaseConnectionAsync(long gen)
{
var probeCts = _probeCts;
var probeTask = _probeTask;
@@ -379,11 +468,6 @@ public sealed class S7Driver
catch { /* loop faults are already surfaced via health; teardown continues */ }
}
// The probe loop has now observed cancellation and released _gate — safe to dispose the CTS.
probeCts?.Dispose();
_probeCts = null;
_probeTask = null;
// Swap the connection out under _gate so a Read/Write/poll pass that already holds the gate (they
// re-read Plc inside it) can never see a disposed Plc mid-swap. Bounded: a wedged holder must not hang
// teardown, in which case we proceed exactly as the un-gated path always did.
@@ -395,6 +479,15 @@ public sealed class S7Driver
_driverInstanceId, DrainTimeout);
try
{
if (!IsCurrentGeneration(gen)) return; // superseded: the newer generation released (or owns) this state
// The probe loop has now observed cancellation and released _gate — safe to dispose the CTS.
if (ReferenceEquals(_probeCts, probeCts))
{
probeCts?.Dispose();
_probeCts = null;
_probeTask = null;
}
_initialized = false;
_plcDead = false;
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
@@ -407,6 +500,101 @@ public sealed class S7Driver
}
}
// ---- lifecycle generation ----
/// <summary>
/// Guards <see cref="InitializeAsync"/> / <see cref="ReinitializeAsync"/> /
/// <see cref="ShutdownAsync"/> / <see cref="DisposeAsync"/> against each other (R07:H5, the
/// Modbus <c>_lifecycleGate</c> pattern). Distinct from <see cref="_gate"/>, which serialises I/O.
/// Never disposed: a Dispose racing a still-running ShutdownAsync must not make it throw.
/// </summary>
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
/// <summary>
/// Bounded wait the teardown paths take on <see cref="_lifecycleGate"/> before tearing down anyway.
/// Internal so a test can shorten it instead of waiting the production 10 s.
/// </summary>
internal TimeSpan LifecycleGateTimeout { get; set; } = TimeSpan.FromSeconds(10);
/// <summary>Serialises generation bumps against generation-stamped health writes.</summary>
private readonly object _healthLock = new();
/// <summary>
/// Lifecycle generation: bumped by every initialize and teardown. Health writes and
/// table/connection publishes carry the generation they were computed under and are dropped
/// once it has moved on. Written under <see cref="_healthLock"/>.
/// </summary>
private long _generation;
/// <summary>Starts a new lifecycle generation.</summary>
/// <returns>The new generation.</returns>
private long BeginGeneration()
{
lock (_healthLock) return ++_generation;
}
/// <summary>The current lifecycle generation.</summary>
private long CurrentGeneration => Interlocked.Read(ref _generation);
/// <summary>True while <paramref name="gen"/> is still the current lifecycle generation.</summary>
/// <param name="gen">The generation to test.</param>
/// <returns><see langword="true"/> when nothing has superseded it.</returns>
private bool IsCurrentGeneration(long gen) => CurrentGeneration == gen;
/// <summary>Writes <paramref name="health"/> only while <paramref name="gen"/> is current.</summary>
/// <param name="gen">The generation the health was computed under.</param>
/// <param name="health">The health to publish.</param>
/// <returns><see langword="true"/> when written.</returns>
private bool TryWriteHealth(long gen, DriverHealth health)
{
lock (_healthLock)
{
if (_generation != gen) return false;
Health = health;
return true;
}
}
/// <summary>
/// Derives and writes a new health from the current one, atomically, only while
/// <paramref name="gen"/> is current — for the passes that preserve <c>LastSuccessfulRead</c>.
/// </summary>
/// <param name="gen">The generation the update was computed under.</param>
/// <param name="update">Maps the current health to the new one; returning null writes nothing.</param>
/// <returns><see langword="true"/> when written.</returns>
private bool TryUpdateHealth(long gen, Func<DriverHealth, DriverHealth?> update)
{
lock (_healthLock)
{
if (_generation != gen) return false;
if (update(Health) is not { } next) return false;
Health = next;
return true;
}
}
/// <summary>The published RawPath → definition and RawPath → parsed-address tables, swapped as one.</summary>
/// <param name="Tags">RawPath → definition.</param>
/// <param name="Parsed">RawPath → pre-parsed address.</param>
private sealed record TagTables(
IReadOnlyDictionary<string, S7TagDefinition> Tags,
IReadOnlyDictionary<string, S7ParsedAddress> Parsed)
{
/// <summary>The empty tables a driver starts with.</summary>
public static readonly TagTables Empty = new(
new Dictionary<string, S7TagDefinition>(StringComparer.Ordinal),
new Dictionary<string, S7ParsedAddress>(StringComparer.Ordinal));
}
/// <summary>The live tag tables; replaced by reference at init, never mutated in place.</summary>
private TagTables _tables = TagTables.Empty;
/// <summary>The live RawPath → definition table.</summary>
private IReadOnlyDictionary<string, S7TagDefinition> TagsByRawPath => Volatile.Read(ref _tables).Tags;
/// <summary>The live RawPath → pre-parsed address table.</summary>
private IReadOnlyDictionary<string, S7ParsedAddress> ParsedByRawPath => Volatile.Read(ref _tables).Parsed;
/// <inheritdoc />
public DriverHealth GetHealth() => Health;
@@ -488,8 +676,15 @@ public sealed class S7Driver
var outcome = new ReadPassOutcome();
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
// The lifecycle generation this pass runs in — its health write is dropped if a teardown or
// re-initialize has moved on by the time it lands. Captured under _gate, where the init
// publish happens, and re-checked against _initialized: a pass that passed RequireInitialized
// and then waited out a teardown must not reopen a connection on a shut-down driver.
var gen = CurrentGeneration;
try
{
RequireInitialized();
// (Re)open the connection under the gate. A failed (re)connect degrades the whole
// batch to BadCommunicationError rather than throwing — the poll loop / next read
// retries, and EnsureConnectedAsync repairs on the following call.
@@ -506,7 +701,7 @@ public sealed class S7Driver
{
for (var i = 0; i < fullReferences.Count; i++)
results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
Health = new DriverHealth(DriverState.Degraded, Health.LastSuccessfulRead, ex.Message);
TryUpdateHealth(gen, h => new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, ex.Message));
return results;
}
@@ -635,10 +830,10 @@ public sealed class S7Driver
state = wf.State;
error = wf.Message;
}
Health = new DriverHealth(
TryUpdateHealth(gen, h => new DriverHealth(
state,
outcome.AnySuccess ? now : Health.LastSuccessfulRead,
error);
outcome.AnySuccess ? now : h.LastSuccessfulRead,
error));
}
_gate.Release();
@@ -1068,7 +1263,7 @@ public sealed class S7Driver
/// <param name="addr">The parsed address, valid only when this returns <c>true</c>.</param>
/// <returns><c>true</c> when the address is known or parses.</returns>
private bool TryGetParsedAddress(S7TagDefinition tag, out S7ParsedAddress addr)
=> _parsedByRawPath.TryGetValue(tag.Name, out addr) || S7AddressParser.TryParse(tag.Address, out addr);
=> ParsedByRawPath.TryGetValue(tag.Name, out addr) || S7AddressParser.TryParse(tag.Address, out addr);
/// <summary>
/// The read-failure classification ladder, shared by the per-tag path and the span path so
@@ -1137,10 +1332,10 @@ public sealed class S7Driver
private async Task<object> ReadOneAsync(IS7Plc plc, S7TagDefinition tag, CancellationToken ct)
{
// Every tag pre-parses its address at init (_parsedByRawPath); the on-demand
// Every tag pre-parses its address at init (ParsedByRawPath); the on-demand
// S7AddressParser.Parse is a defensive fallback for a tag not in that table. Parse throws
// FormatException on a bad address, which the caller's catch maps to BadCommunicationError.
var addr = _parsedByRawPath.TryGetValue(tag.Name, out var parsed)
var addr = ParsedByRawPath.TryGetValue(tag.Name, out var parsed)
? parsed
: S7AddressParser.Parse(tag.Address);
@@ -1691,8 +1886,11 @@ public sealed class S7Driver
var outcome = new WritePassOutcome();
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
var gen = CurrentGeneration; // see ReadAsync
try
{
RequireInitialized();
// (Re)open under the gate; a failed (re)connect degrades every write to
// BadCommunicationError rather than throwing.
IS7Plc plc;
@@ -1733,7 +1931,7 @@ public sealed class S7Driver
// Timer/Counter, but the mapper doesn't run that guard).
// TryParse (not Parse) so a malformed address falls through to the existing
// error path rather than throwing here.
if (_parsedByRawPath.TryGetValue(tag.Name, out var tcParsed)
if (ParsedByRawPath.TryGetValue(tag.Name, out var tcParsed)
? tcParsed.Area is S7Area.Timer or S7Area.Counter
: (S7AddressParser.TryParse(tag.Address, out tcParsed) && tcParsed.Area is S7Area.Timer or S7Area.Counter))
{
@@ -1806,7 +2004,7 @@ public sealed class S7Driver
{
// THE single health write of the batch (a finally, so an OCE mid-batch still flushes the
// evidence gathered so far — a pass that observed nothing writes nothing).
ApplyWritePassHealth(outcome);
ApplyWritePassHealth(gen, outcome);
_gate.Release();
}
return results;
@@ -1818,10 +2016,11 @@ public sealed class S7Driver
/// successful sibling. Maintains <see cref="_writeFault"/>, the sticky Faulted a read pass
/// must not erase. Caller holds <see cref="_gate"/>.
/// </summary>
/// <param name="gen">The lifecycle generation the batch ran in; a superseded batch writes nothing.</param>
/// <param name="outcome">The batch's accumulated outcomes.</param>
private void ApplyWritePassHealth(WritePassOutcome outcome)
private void ApplyWritePassHealth(long gen, WritePassOutcome outcome)
{
if (!outcome.HasObservations) return;
if (!outcome.HasObservations || !IsCurrentGeneration(gen)) return;
var worst = outcome.WorstFailure;
var priorWriteFault = _writeFault;
@@ -1836,9 +2035,10 @@ public sealed class S7Driver
// side) were holding — a clean write is the only evidence the alert has cleared.
// Only a Faulted the write side owns: a read-side Faulted (e.g. access denied on another DB) is the
// read pass's to clear, and Health still carries the write fault's own message iff it is ours.
if (outcome.AnySuccess && _writeFault is null && priorWriteFault is { } owned
&& Health.State == DriverState.Faulted && Health.LastError == owned.Message)
Health = new DriverHealth(DriverState.Healthy, Health.LastSuccessfulRead, null);
if (outcome.AnySuccess && _writeFault is null && priorWriteFault is { } owned)
TryUpdateHealth(gen, h => h.State == DriverState.Faulted && h.LastError == owned.Message
? new DriverHealth(DriverState.Healthy, h.LastSuccessfulRead, null)
: null);
return;
}
@@ -1850,7 +2050,7 @@ public sealed class S7Driver
state = wf.State;
error = wf.Message;
}
Health = new DriverHealth(state, Health.LastSuccessfulRead, error);
TryUpdateHealth(gen, h => new DriverHealth(state, h.LastSuccessfulRead, error));
}
/// <summary>
@@ -1881,10 +2081,10 @@ public sealed class S7Driver
private async Task WriteOneAsync(IS7Plc plc, S7TagDefinition tag, object? value, CancellationToken ct)
{
// Parse the address the same way ReadOneAsync does: every tag pre-parses at init
// (_parsedByRawPath), with an on-demand parse as a defensive fallback. Needed here so the
// (ParsedByRawPath), with an on-demand parse as a defensive fallback. Needed here so the
// wide-type write can byte-address the block (the narrow path below addresses by the raw
// address string instead).
var addr = _parsedByRawPath.TryGetValue(tag.Name, out var parsed)
var addr = ParsedByRawPath.TryGetValue(tag.Name, out var parsed)
? parsed
: S7AddressParser.Parse(tag.Address);
@@ -2243,7 +2443,7 @@ public sealed class S7Driver
{
ArgumentNullException.ThrowIfNull(builder);
var folder = builder.Folder("S7", "S7");
foreach (var t in _tagsByRawPath.Values)
foreach (var t in TagsByRawPath.Values)
{
// A tag carrying a non-null array count (>= 1) surfaces as a 1-D OPC UA array node.
// A null count stays scalar. A count of 1 IS a valid 1-element array: the foundation
@@ -2308,8 +2508,9 @@ public sealed class S7Driver
private void HandlePollError(Exception ex)
{
_logger.LogWarning(ex, "S7 poll reader failed. Driver={DriverInstanceId}", _driverInstanceId);
if (Health.State != DriverState.Faulted)
Health = new DriverHealth(DriverState.Degraded, Health.LastSuccessfulRead, ex.Message);
TryUpdateHealth(CurrentGeneration, h => h.State == DriverState.Faulted
? null
: new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, ex.Message));
}
// ---- IHostConnectivityProbe ----
@@ -2418,7 +2619,10 @@ public sealed class S7Driver
if (_disposed) return;
_disposed = true;
SynchronousTeardown();
_gate.Dispose();
// _gate / _lifecycleGate are deliberately NOT disposed: a Dispose can race a still-running
// ShutdownAsync (the runtime disposes right after a bounded shutdown), which would then throw
// ObjectDisposedException mid-teardown. A SemaphoreSlim whose wait handle was never requested
// holds no unmanaged resource.
}
/// <summary>Asynchronously disposes the driver and releases resources.</summary>
@@ -2427,9 +2631,10 @@ public sealed class S7Driver
{
if (_disposed) return;
_disposed = true;
// Safe while a ShutdownAsync is still running: it waits (bounded) for the lifecycle gate, and
// the teardown is idempotent. The semaphores are not disposed (see Dispose).
try { await ShutdownAsync(CancellationToken.None).ConfigureAwait(false); }
catch { /* disposal is best-effort */ }
_gate.Dispose();
}
/// <summary>
@@ -2439,6 +2644,10 @@ public sealed class S7Driver
/// </summary>
private void SynchronousTeardown()
{
// Its own lifecycle generation, like ShutdownCoreAsync: an initialize still running past this
// point discards what it built instead of publishing it over the torn-down driver.
var gen = BeginGeneration();
// Drain the polling overlay (drain-before-dispose).
try { _poll.DisposeAsync().AsTask().Wait(DrainTimeout); }
catch { /* timeouts/loop faults are tolerated — teardown continues */ }
@@ -2463,6 +2672,6 @@ public sealed class S7Driver
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
Plc = null;
Volatile.Write(ref _negotiatedPduSize, 0); // a stale post-teardown reading would read as "online"
Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null);
TryUpdateHealth(gen, h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null));
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7LifecycleGateTests.cs
+143
View File
@@ -0,0 +1,143 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.S7.Tests;
/// <summary>
/// R07:H5 — overlapping lifecycle calls. <c>DriverInstanceActor</c> dispatches initialize on a
/// fire-and-forget task and a reconnect retry can start a second attempt while the first is still in
/// <c>OpenAsync</c>; an abandoned initialize can also still be running when the next shutdown starts.
/// Every wait here is a condition wait on the fake's own signals, never a sleep.
/// </summary>
[Trait("Category", "Unit")]
public sealed class S7LifecycleGateTests
{
private static readonly TimeSpan Bound = TimeSpan.FromSeconds(10);
private static S7DriverOptions Options(bool probe = false) => new()
{
Host = "192.0.2.1",
Timeout = TimeSpan.FromSeconds(30), // the fake's gated open is released by the test, never timed out
Probe = new S7ProbeOptions { Enabled = probe, Interval = TimeSpan.FromSeconds(30) },
RawTags = S7RawTags.Entries(new S7TagDefinition("W0", "DB1.DBW0", S7DataType.UInt16, Writable: false)),
};
/// <summary>A factory whose FIRST open blocks until released; later opens complete immediately.</summary>
private sealed class FirstOpenBlocks
{
public readonly ScriptedS7PlcFactory Factory = new();
public readonly TaskCompletionSource FirstOpenEntered = new(TaskCreationOptions.RunContinuationsAsynchronously);
public readonly TaskCompletionSource ReleaseFirstOpen = new(TaskCreationOptions.RunContinuationsAsynchronously);
public FirstOpenBlocks()
{
Factory.Open = async (index, ct) =>
{
if (index != 0) return;
FirstOpenEntered.TrySetResult();
await ReleaseFirstOpen.Task.WaitAsync(ct).ConfigureAwait(false);
};
}
}
/// <summary>
/// A second initialize waits for the first instead of running beside it, then releases the first
/// attempt's connection: exactly one live connection remains. Before the gate, attempt B opened
/// its own connection while A was still in <c>OpenAsync</c>, and A's published connection (and
/// probe loop) was overwritten and never released.
/// </summary>
[Fact]
public async Task Overlapping_initializes_are_serialised_and_leave_one_live_connection()
{
var ct = TestContext.Current.CancellationToken;
var f = new FirstOpenBlocks();
using var drv = new S7Driver(Options(), "s7-overlap", f.Factory);
var a = drv.InitializeAsync("{}", ct);
await f.FirstOpenEntered.Task.WaitAsync(Bound, ct);
var b = drv.InitializeAsync("{}", ct);
// Positive control first: B is genuinely pending (not finished, not faulted)...
b.IsCompleted.ShouldBeFalse();
// ...and while A holds the gate B has not created a connection of its own. B can only reach
// its open after A completes, so this is not a race the test can lose by being slow.
f.Factory.Created.Count.ShouldBe(1, "a second initialize must not open beside the first");
f.ReleaseFirstOpen.SetResult();
await a.WaitAsync(Bound, ct);
await b.WaitAsync(Bound, ct);
f.Factory.Created.Count.ShouldBe(2);
f.Factory.Created[0].Disposed.ShouldBeTrue("B releases A's published connection before replacing it");
f.Factory.Created[1].Disposed.ShouldBeFalse();
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
(await drv.ReadAsync(["W0"], ct))[0].StatusCode.ShouldBe(0u);
}
/// <summary>
/// An initialize still running when a shutdown gives up waiting for it must not publish its
/// connection, its health or its probe loop over the shut-down driver; it fails and disposes what
/// it opened. Before the fix the late attempt published Healthy + a live connection after the
/// shutdown returned.
/// </summary>
[Fact]
public async Task An_initialize_superseded_by_shutdown_publishes_nothing()
{
var ct = TestContext.Current.CancellationToken;
var f = new FirstOpenBlocks();
using var drv = new S7Driver(Options(probe: true), "s7-superseded", f.Factory)
{
LifecycleGateTimeout = TimeSpan.FromSeconds(1),
};
var init = drv.InitializeAsync("{}", ct);
await f.FirstOpenEntered.Task.WaitAsync(Bound, ct);
// The shutdown cannot get the lifecycle gate (the attempt holds it), gives up after its bound
// and tears down anyway.
await drv.ShutdownAsync(ct).WaitAsync(Bound, ct);
drv.GetHealth().State.ShouldBe(DriverState.Unknown);
// The abandoned attempt now completes its open...
f.ReleaseFirstOpen.SetResult();
var ex = await Should.ThrowAsync<InvalidOperationException>(() => init.WaitAsync(Bound, ct));
ex.Message.ShouldContain("superseded");
// ...and published nothing: no connection kept, health untouched, the driver still shut down.
f.Factory.Created.ShouldHaveSingleItem().Disposed.ShouldBeTrue();
drv.GetHealth().State.ShouldBe(DriverState.Unknown, "a superseded attempt must not write Healthy or Faulted");
await Should.ThrowAsync<InvalidOperationException>(() => drv.ReadAsync(["W0"], ct));
// Positive control: the driver is not wedged — a fresh initialize publishes normally.
await drv.InitializeAsync("{}", ct).WaitAsync(Bound, ct);
drv.GetHealth().State.ShouldBe(DriverState.Healthy);
(await drv.ReadAsync(["W0"], ct))[0].StatusCode.ShouldBe(0u);
}
/// <summary>
/// Dispose racing a shutdown that is still running is safe: neither throws, and the driver ends
/// torn down. The semaphores are no longer disposed, so the running shutdown cannot hit an
/// ObjectDisposedException.
/// </summary>
[Fact]
public async Task Dispose_while_a_shutdown_is_still_running_is_safe()
{
var ct = TestContext.Current.CancellationToken;
var f = new FirstOpenBlocks();
var drv = new S7Driver(Options(), "s7-dispose-race", f.Factory) { LifecycleGateTimeout = TimeSpan.FromSeconds(1) };
var init = drv.InitializeAsync("{}", ct);
await f.FirstOpenEntered.Task.WaitAsync(Bound, ct);
var shutdown = drv.ShutdownAsync(ct); // waits on the gate the blocked init holds
var dispose = drv.DisposeAsync().AsTask();
await shutdown.WaitAsync(Bound, ct);
await dispose.WaitAsync(Bound, ct);
f.ReleaseFirstOpen.SetResult();
await Should.ThrowAsync<InvalidOperationException>(() => init.WaitAsync(Bound, ct));
f.Factory.Created.ShouldHaveSingleItem().Disposed.ShouldBeTrue();
drv.GetHealth().State.ShouldBe(DriverState.Unknown);
}
}