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

Merge branch 'fix/archreview-crit3-s7-reconnect'

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-09 06:07:33 -04:00
parent 45455b7677 25c0c6f68f
commit dfc02af3f0
4 changed files with 660 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+203 -37
View File
@@ -34,16 +34,21 @@ public sealed class S7Driver
{
private readonly string _driverInstanceId;
private readonly ILogger<S7Driver> _logger;
private readonly IS7PlcFactory _plcFactory;
/// <summary>Initializes a new instance of the <see cref="S7Driver"/> class.</summary>
/// <param name="options">Driver configuration (the constructor-supplied fallback used when
/// Initialize/Reinitialize receive an empty config body).</param>
/// <param name="driverInstanceId">Unique driver instance identifier.</param>
/// <param name="plcFactory">Optional S7 connection factory; defaults to <see cref="S7PlcFactory"/>
/// (real S7.Net). Tests inject a fake to exercise the reconnect path without a live PLC.</param>
/// <param name="logger">Optional logger; a null logger is used when not supplied.</param>
public S7Driver(S7DriverOptions options, string driverInstanceId, ILogger<S7Driver>? logger = null)
public S7Driver(S7DriverOptions options, string driverInstanceId,
IS7PlcFactory? plcFactory = null, ILogger<S7Driver>? logger = null)
{
_options = options;
_driverInstanceId = driverInstanceId;
_plcFactory = plcFactory ?? new S7PlcFactory();
_logger = logger ?? NullLogger<S7Driver>.Instance;
_resolver = new EquipmentTagRefResolver<S7TagDefinition>(
r => _tagsByName.TryGetValue(r, out var t) ? t : null,
@@ -115,11 +120,30 @@ public sealed class S7Driver
internal SemaphoreSlim Gate => _gate;
/// <summary>
/// Active S7.Net PLC connection. Null until <see cref="InitializeAsync"/> returns; null
/// after <see cref="ShutdownAsync"/>. Read-only outside this class; PR 64's Read/Write
/// will take the <see cref="_gate"/> before touching it.
/// Active S7 PLC connection. Null until <see cref="InitializeAsync"/> returns; null
/// after <see cref="ShutdownAsync"/>. Also (re)opened lazily by
/// <see cref="EnsureConnectedAsync"/> after a transient drop. Read-only outside this class;
/// Read/Write/probe take the <see cref="_gate"/> before touching it.
/// </summary>
internal Plc? Plc { get; private set; }
internal IS7Plc? Plc { get; private set; }
/// <summary>
/// True once <see cref="InitializeAsync"/> has parsed the tag table + opened the first
/// connection, false after <see cref="ShutdownAsync"/>. Distinct from <c>Plc != null</c>
/// because the reconnect path can leave <see cref="Plc"/> transiently null / dead while the
/// driver is still initialized — a data call then repairs the connection rather than
/// throwing "not initialized". Guarded by <see cref="_gate"/>.
/// </summary>
private bool _initialized;
/// <summary>
/// Set when a connection-fatal fault (socket drop / <see cref="ErrorCode.ConnectionError"/>)
/// is observed on a read/write, so the next <see cref="EnsureConnectedAsync"/> disposes the
/// dead handle and re-opens. Kept as a flag (rather than nulling <see cref="Plc"/> inline) so
/// the dead handle is disposed exactly once, in <see cref="EnsureConnectedAsync"/>. Guarded
/// by <see cref="_gate"/>.
/// </summary>
private bool _plcDead;
private DriverHealth _health = new(DriverState.Unknown, null, null);
private bool _disposed;
@@ -153,18 +177,26 @@ public sealed class S7Driver
// the single grep target / future seam should a type ever need re-gating at init.
RejectUnsupportedTagDataTypes();
var plc = new Plc(S7CpuTypeMap.ToS7Net(_options.CpuType), _options.Host, _options.Port, _options.Rack, _options.Slot);
// S7netplus writes timeouts into the underlying TcpClient via Plc.WriteTimeout /
// Plc.ReadTimeout (milliseconds). Set before OpenAsync so the handshake itself
// honours the bound.
plc.WriteTimeout = (int)_options.Timeout.TotalMilliseconds;
plc.ReadTimeout = (int)_options.Timeout.TotalMilliseconds;
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.Timeout);
await plc.OpenAsync(cts.Token).ConfigureAwait(false);
// 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.CpuType, _options.Host, _options.Port, _options.Rack, _options.Slot, _options.Timeout);
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cts.CancelAfter(_options.Timeout);
await plc.OpenAsync(cts.Token).ConfigureAwait(false);
}
catch
{
// 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;
}
Plc = plc;
_plcDead = false;
// Parse every tag's address once at init so config typos fail fast here instead
// of surfacing as BadInternalError on every Read against the bad tag. The parser
@@ -179,6 +211,7 @@ public sealed class S7Driver
_parsedByName[t.Name] = parsed;
}
_initialized = true;
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
_logger.LogInformation("S7Driver connected. Driver={DriverInstanceId} Host={Host} CPU={CpuType} Tags={TagCount}",
_driverInstanceId, _options.Host, _options.CpuType, _options.Tags.Count);
@@ -199,9 +232,11 @@ public sealed class S7Driver
catch (Exception ex)
{
// Clean up a partially-constructed Plc so a retry from the caller doesn't leak
// the TcpClient. S7netplus's Close() is best-effort and idempotent.
try { Plc?.Close(); } catch { }
// the TcpClient. Dispose is best-effort and idempotent.
try { Plc?.Dispose(); } catch { }
Plc = null;
_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);
throw;
@@ -257,7 +292,9 @@ public sealed class S7Driver
foreach (var state in subscriptions)
state.Cts.Dispose();
try { Plc?.Close(); } catch { /* best-effort — tearing down anyway */ }
_initialized = false;
_plcDead = false;
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
Plc = null;
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}
@@ -429,17 +466,34 @@ public sealed class S7Driver
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
// Validate the list before RequirePlc() so a null argument produces an
// Validate the list before the init check so a null argument produces an
// ArgumentNullException (consistent with DiscoverAsync) rather than an
// InvalidOperationException from the not-initialized check.
ArgumentNullException.ThrowIfNull(fullReferences);
var plc = RequirePlc();
RequireInitialized();
var now = DateTime.UtcNow;
var results = new DataValueSnapshot[fullReferences.Count];
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// (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.
IS7Plc plc;
try
{
plc = await EnsureConnectedAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) { throw; }
catch (Exception ex)
{
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);
return results;
}
for (var i = 0; i < fullReferences.Count; i++)
{
var name = fullReferences[i];
@@ -472,12 +526,18 @@ public sealed class S7Driver
catch (PlcException pex)
{
// A genuine device-layer fault (CPU error, hardware fault) — transient
// enough to keep retrying; report BadDeviceFailure and degrade health.
// enough to keep retrying; report BadDeviceFailure and degrade health. A
// connection-fatal PlcException (socket drop / ConnectionError) additionally
// marks the handle dead so the next read reopens it.
MarkConnectionDeadIfFatal(pex);
results[i] = new DataValueSnapshot(null, StatusBadDeviceFailure, null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, pex.Message);
}
catch (Exception ex)
{
// Socket / timeout / transport fault — mark the handle dead when it's a
// connection loss so EnsureConnectedAsync reopens on the next call.
MarkConnectionDeadIfFatal(ex);
results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
@@ -487,7 +547,7 @@ public sealed class S7Driver
return results;
}
private async Task<object> ReadOneAsync(Plc plc, S7TagDefinition tag, CancellationToken ct)
private async Task<object> ReadOneAsync(IS7Plc plc, S7TagDefinition tag, CancellationToken ct)
{
// Authored tags pre-parse their address at init (_parsedByName); an equipment-tag ref
// (resolved transiently by _resolver) has no _parsedByName entry, so parse its address
@@ -533,7 +593,7 @@ public sealed class S7Driver
/// to the pure <see cref="DecodeArrayBlock"/> decode loop. Timer/Counter areas are
/// already rejected at init, so only DB/M/I/Q reach here.
/// </summary>
private async Task<object> ReadArrayAsync(Plc plc, S7TagDefinition tag, S7ParsedAddress addr, CancellationToken ct)
private async Task<object> ReadArrayAsync(IS7Plc plc, S7TagDefinition tag, S7ParsedAddress addr, CancellationToken ct)
{
var count = tag.ArrayCount!.Value;
var elementBytes = ElementByteSize(addr.Size);
@@ -583,7 +643,7 @@ public sealed class S7Driver
/// PLC round-trip), then hands the raw block to the pure <see cref="DecodeScalarBlock"/>
/// decode. Mirrors <see cref="ReadArrayAsync"/>'s shape.
/// </summary>
private async Task<object> ReadScalarBlockAsync(Plc plc, S7TagDefinition tag, S7ParsedAddress addr, CancellationToken ct)
private async Task<object> ReadScalarBlockAsync(IS7Plc plc, S7TagDefinition tag, S7ParsedAddress addr, CancellationToken ct)
{
var width = ScalarByteWidth(tag, addr);
// For DB/M/I/Q, addr.ByteOffset is the start BYTE and `width` the byte count. For
@@ -921,15 +981,31 @@ public sealed class S7Driver
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
{
// Same as ReadAsync — validate before RequirePlc() so a null argument is a
// Same as ReadAsync — validate before the init check so a null argument is a
// typed argument error, not the "not initialized" surface.
ArgumentNullException.ThrowIfNull(writes);
var plc = RequirePlc();
RequireInitialized();
var results = new WriteResult[writes.Count];
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
// (Re)open under the gate; a failed (re)connect degrades every write to
// BadCommunicationError rather than throwing.
IS7Plc plc;
try
{
plc = await EnsureConnectedAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) { throw; }
catch (Exception ex)
{
for (var i = 0; i < writes.Count; i++)
results[i] = new WriteResult(StatusBadCommunicationError);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
return results;
}
for (var i = 0; i < writes.Count; i++)
{
var w = writes[i];
@@ -988,14 +1064,18 @@ public sealed class S7Driver
{
// Genuine device-layer fault — degrade health so a PLC-down-during-writes
// scenario is visible to the operator (previously health was never updated
// on write failure).
// on write failure). A connection-fatal PlcException also marks the handle
// dead so the next call reopens.
MarkConnectionDeadIfFatal(pex);
results[i] = new WriteResult(StatusBadDeviceFailure);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, pex.Message);
}
catch (Exception ex)
{
// Socket/timeout/conversion failure. Map to BadCommunicationError (not
// BadInternalError) for transport faults; degrade health.
// BadInternalError) for transport faults; degrade health, and mark the handle
// dead when it's a connection loss.
MarkConnectionDeadIfFatal(ex);
results[i] = new WriteResult(StatusBadCommunicationError);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
@@ -1005,7 +1085,7 @@ public sealed class S7Driver
return results;
}
private async Task WriteOneAsync(Plc plc, S7TagDefinition tag, object? value, CancellationToken ct)
private async Task WriteOneAsync(IS7Plc plc, S7TagDefinition tag, object? value, CancellationToken ct)
{
// Defence-in-depth guard: authored array tags are rejected at init
// (RejectUnsupportedTagConfigs guard-e), but a transient equipment-tag ref resolved by
@@ -1047,7 +1127,7 @@ public sealed class S7Driver
/// The pure <see cref="EncodeScalarBlock"/> produces the big-endian bytes; this method
/// owns only the network I/O (mirrors <see cref="ReadScalarBlockAsync"/>).
/// </summary>
private async Task WriteScalarBlockAsync(Plc plc, S7TagDefinition tag, S7ParsedAddress addr, object? value, CancellationToken ct)
private async Task WriteScalarBlockAsync(IS7Plc plc, S7TagDefinition tag, S7ParsedAddress addr, object? value, CancellationToken ct)
{
var bytes = EncodeScalarBlock(tag, value);
await plc.WriteBytesAsync(ToS7NetArea(addr.Area), addr.DbNumber, addr.ByteOffset, bytes, ct)
@@ -1085,8 +1165,91 @@ public sealed class S7Driver
_ => throw new InvalidOperationException($"Unknown S7DataType {dataType}"),
};
private Plc RequirePlc() =>
Plc ?? throw new InvalidOperationException("S7Driver not initialized");
private void RequireInitialized()
{
if (!_initialized) throw new InvalidOperationException("S7Driver not initialized");
}
/// <summary>
/// Returns a live S7 connection, lazily disposing a dead handle and re-opening a fresh one
/// when the previous connection dropped (STAB-1 reconnect path). The fast path returns the
/// current <see cref="Plc"/> when it is still connected and not marked dead; otherwise it
/// builds + opens a new connection via <see cref="_plcFactory"/> with the same
/// timeout-linked CTS the initial connect uses.
/// </summary>
/// <remarks>
/// MUST be called while holding <see cref="_gate"/> — it mutates <see cref="Plc"/> /
/// <see cref="_plcDead"/> and reuses the single-connection-per-PLC invariant the gate
/// enforces. The read / write / probe paths all take the gate before calling it.
/// </remarks>
/// <param name="ct">Cancellation token for the (re)connect attempt.</param>
/// <returns>A live <see cref="IS7Plc"/>.</returns>
private async Task<IS7Plc> EnsureConnectedAsync(CancellationToken ct)
{
// Fast path — the current connection is live and not flagged dead.
if (!_plcDead && Plc is { IsConnected: true } live) return live;
// Dispose the stale / dead handle before making a fresh one (exactly once, here).
if (Plc is not null)
{
try { Plc.Dispose(); } catch { /* best-effort */ }
Plc = null;
}
_plcDead = false;
var plc = _plcFactory.Create(
_options.CpuType, _options.Host, _options.Port, _options.Rack, _options.Slot, _options.Timeout);
try
{
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(_options.Timeout);
await plc.OpenAsync(cts.Token).ConfigureAwait(false);
}
catch
{
try { plc.Dispose(); } catch { /* best-effort */ }
throw;
}
Plc = plc;
_logger.LogInformation("S7Driver reconnected. Driver={DriverInstanceId} Host={Host} CPU={CpuType}",
_driverInstanceId, _options.Host, _options.CpuType);
return plc;
}
/// <summary>
/// Flags the current connection dead when <paramref name="ex"/> is a connection-level loss,
/// so the next <see cref="EnsureConnectedAsync"/> reopens. A no-op for data-address / type
/// faults — those don't warrant tearing down a healthy socket.
/// </summary>
/// <param name="ex">The exception observed on a read / write.</param>
private void MarkConnectionDeadIfFatal(Exception ex)
{
if (IsS7ConnectionFatal(ex)) _plcDead = true;
}
/// <summary>
/// True when <paramref name="ex"/> (or any inner exception) is a socket-level / connection
/// loss that a reopen can repair — a <see cref="System.Net.Sockets.SocketException"/> /
/// <see cref="System.IO.IOException"/> / <see cref="ObjectDisposedException"/>, or an S7.Net
/// <see cref="PlcException"/> carrying <see cref="ErrorCode.ConnectionError"/>. A
/// data-address / type error (S7.Net's <see cref="ErrorCode.ReadData"/> /
/// <see cref="ErrorCode.WriteData"/> for a bad address, PUT/GET-denied, etc.) is deliberately
/// NOT treated as fatal — reopening wouldn't help and would churn a healthy connection.
/// </summary>
/// <param name="ex">The exception to classify.</param>
/// <returns><c>true</c> when a reconnect is warranted.</returns>
private static bool IsS7ConnectionFatal(Exception ex)
{
for (Exception? e = ex; e is not null; e = e.InnerException)
{
if (e is System.Net.Sockets.SocketException or System.IO.IOException or ObjectDisposedException)
return true;
if (e is PlcException { ErrorCode: ErrorCode.ConnectionError })
return true;
}
return false;
}
/// <summary>
/// Detects an S7 PUT/GET-disabled / access-protection fault inside an S7.Net
@@ -1363,13 +1526,14 @@ public sealed class S7Driver
using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
probeCts.CancelAfter(_options.Probe.Timeout);
var plc = Plc;
if (plc is null) throw new InvalidOperationException("Plc dropped during probe");
await _gate.WaitAsync(probeCts.Token).ConfigureAwait(false);
try
{
_ = await plc.ReadStatusAsync(probeCts.Token).ConfigureAwait(false);
// EnsureConnectedAsync doubles as the reconnect backstop: if a read/write
// marked the handle dead (or the socket dropped between ticks), the probe
// reopens it here so recovery doesn't wait for the next data call.
var plc = await EnsureConnectedAsync(probeCts.Token).ConfigureAwait(false);
await plc.ReadStatusAsync(probeCts.Token).ConfigureAwait(false);
success = true;
}
finally { _gate.Release(); }
@@ -1460,7 +1624,9 @@ public sealed class S7Driver
try { state.Cts.Dispose(); } catch { }
}
try { Plc?.Close(); } catch { /* best-effort — tearing down anyway */ }
_initialized = false;
_plcDead = false;
try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ }
Plc = null;
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
}