Joseph Doherty
1143 lines
57 KiB
C#
1143 lines
57 KiB
C#
using System.Collections.Concurrent;
|
||
using Microsoft.Extensions.Logging;
|
||
using Microsoft.Extensions.Logging.Abstractions;
|
||
using S7.Net;
|
||
using S7NetDataType = global::S7.Net.DataType;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.S7;
|
||
|
||
/// <summary>
|
||
/// Siemens S7 native driver — speaks S7comm over ISO-on-TCP (port 102) via the S7netplus
|
||
/// library. First implementation of <see cref="IDriver"/> for an in-process .NET Standard
|
||
/// PLC protocol that is NOT Modbus, validating that the v2 driver-capability interfaces
|
||
/// generalize beyond Modbus + Galaxy.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// PR 62 ships the scaffold: <see cref="IDriver"/> only (Initialize / Reinitialize /
|
||
/// Shutdown / GetHealth). <see cref="ITagDiscovery"/>, <see cref="IReadable"/>,
|
||
/// <see cref="IWritable"/>, <see cref="ISubscribable"/>, <see cref="IHostConnectivityProbe"/>
|
||
/// land in PRs 63-65 once the address parser (PR 63) is in place.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Single-connection policy</b>: S7netplus documented pattern is one
|
||
/// <c>Plc</c> instance per PLC, serialized with a <see cref="SemaphoreSlim"/>.
|
||
/// Parallelising reads against a single S7 CPU doesn't help — the CPU scans the
|
||
/// communication mailbox at most once per cycle (2-10 ms) and queues concurrent
|
||
/// requests wire-side anyway. Multiple client-side connections just waste the CPU's
|
||
/// 8-64 connection-resource budget.
|
||
/// </para>
|
||
/// </remarks>
|
||
public sealed class S7Driver
|
||
: IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IDisposable, IAsyncDisposable
|
||
{
|
||
private readonly string _driverInstanceId;
|
||
private readonly ILogger<S7Driver> _logger;
|
||
|
||
/// <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="logger">Optional logger; a null logger is used when not supplied.</param>
|
||
public S7Driver(S7DriverOptions options, string driverInstanceId, ILogger<S7Driver>? logger = null)
|
||
{
|
||
_options = options;
|
||
_driverInstanceId = driverInstanceId;
|
||
_logger = logger ?? NullLogger<S7Driver>.Instance;
|
||
_resolver = new EquipmentTagRefResolver<S7TagDefinition>(
|
||
r => _tagsByName.TryGetValue(r, out var t) ? t : null,
|
||
r => S7EquipmentTagParser.TryParse(r, out var d) ? d : null);
|
||
}
|
||
// ---- ISubscribable + IHostConnectivityProbe state ----
|
||
|
||
private readonly ConcurrentDictionary<long, SubscriptionState> _subscriptions = new();
|
||
private long _nextSubscriptionId;
|
||
private readonly object _probeLock = new();
|
||
private HostState _hostState = HostState.Unknown;
|
||
private DateTime _hostStateChangedUtc = DateTime.UtcNow;
|
||
private CancellationTokenSource? _probeCts;
|
||
|
||
/// <summary>
|
||
/// Handle to the in-flight probe loop. Tracked (rather than fire-and-forget) so
|
||
/// <see cref="ShutdownAsync"/> can await it after cancelling — otherwise a probe
|
||
/// iteration still inside the <see cref="_gate"/> would race a disposed semaphore.
|
||
/// See code-review finding Driver.S7-006.
|
||
/// </summary>
|
||
private Task? _probeTask;
|
||
|
||
/// <summary>
|
||
/// Bounded grace window <see cref="ShutdownAsync"/> waits for the probe + poll loops to
|
||
/// observe cancellation and exit before it disposes the shared semaphore / CTS objects.
|
||
/// </summary>
|
||
private static readonly TimeSpan DrainTimeout = TimeSpan.FromSeconds(5);
|
||
|
||
/// <summary>Occurs when a subscribed tag value or status code changes.</summary>
|
||
public event EventHandler<DataChangeEventArgs>? OnDataChange;
|
||
/// <summary>Occurs when host connectivity status changes.</summary>
|
||
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
|
||
|
||
/// <summary>OPC UA StatusCode used when the tag name isn't in the driver's tag map.</summary>
|
||
private const uint StatusBadNodeIdUnknown = 0x80340000u;
|
||
/// <summary>OPC UA StatusCode used when the tag's data type isn't implemented yet.</summary>
|
||
private const uint StatusBadNotSupported = 0x803D0000u;
|
||
/// <summary>OPC UA StatusCode used when the tag is declared read-only.</summary>
|
||
private const uint StatusBadNotWritable = 0x803B0000u;
|
||
/// <summary>OPC UA StatusCode used when write fails validation (e.g. out-of-range value).</summary>
|
||
private const uint StatusBadInternalError = 0x80020000u;
|
||
/// <summary>OPC UA StatusCode used for socket / timeout / protocol-layer faults.</summary>
|
||
private const uint StatusBadCommunicationError = 0x80050000u;
|
||
/// <summary>OPC UA StatusCode used for a genuine device fault (CPU error, hardware fault).</summary>
|
||
private const uint StatusBadDeviceFailure = 0x808B0000u;
|
||
|
||
private readonly Dictionary<string, S7TagDefinition> _tagsByName = new(StringComparer.OrdinalIgnoreCase);
|
||
private readonly Dictionary<string, S7ParsedAddress> _parsedByName = new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
// Resolves a read/write/subscribe fullReference to a tag definition, bridging the two
|
||
// authoring models: an authored tag-table entry (by name) OR an equipment tag whose
|
||
// reference is its raw TagConfig JSON (parsed once via S7EquipmentTagParser, cached).
|
||
private readonly EquipmentTagRefResolver<S7TagDefinition> _resolver;
|
||
|
||
/// <summary>
|
||
/// Active driver configuration. Seeded from the constructor argument, then replaced by
|
||
/// whatever <see cref="InitializeAsync"/> / <see cref="ReinitializeAsync"/> parse out of
|
||
/// the supplied <c>driverConfigJson</c> — see code-review finding Driver.S7-011. The
|
||
/// constructor value is the fallback used when the caller passes an empty / placeholder
|
||
/// JSON document (e.g. the <c>"{}"</c> some unit tests pass).
|
||
/// </summary>
|
||
private S7DriverOptions _options;
|
||
private readonly SemaphoreSlim _gate = new(1, 1);
|
||
|
||
/// <summary>
|
||
/// Per-connection gate. Internal so PRs 63-65 (read/write/subscribe) can serialize on
|
||
/// the same semaphore without exposing it publicly. Single-connection-per-PLC is a
|
||
/// hard requirement of S7netplus — see class remarks.
|
||
/// </summary>
|
||
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.
|
||
/// </summary>
|
||
internal Plc? Plc { get; private set; }
|
||
|
||
private DriverHealth _health = new(DriverState.Unknown, null, null);
|
||
private bool _disposed;
|
||
|
||
/// <summary>Gets the unique driver instance identifier.</summary>
|
||
public string DriverInstanceId => _driverInstanceId;
|
||
/// <summary>Gets the driver type name.</summary>
|
||
public string DriverType => "S7";
|
||
|
||
/// <summary>Initializes the driver with the provided configuration.</summary>
|
||
/// <param name="driverConfigJson">JSON configuration string.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||
{
|
||
_health = new DriverHealth(DriverState.Initializing, null, null);
|
||
try
|
||
{
|
||
// Re-parse the supplied DriverConfig JSON so a config change delivered through the
|
||
// IDriver contract is honoured (Driver.S7-011). An empty / placeholder document
|
||
// (e.g. the "{}" some unit tests pass) keeps the constructor-supplied options.
|
||
if (HasConfigBody(driverConfigJson))
|
||
_options = S7DriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson);
|
||
|
||
// Timer (T{n}) / Counter (C{n}) addresses parse cleanly but the read path has no
|
||
// S7DataType for them and no decode case — reject them here so a config typo
|
||
// fails fast at init instead of throwing a misleading type-mismatch on every
|
||
// read (Driver.S7-001). Drop this guard when Timer/Counter reads are wired through.
|
||
RejectUnsupportedTagAddresses();
|
||
|
||
// S7DataType values that ReadOneAsync / WriteOneAsync currently throw
|
||
// NotSupportedException for (Int64, UInt64, Float64, String, DateTime) must also
|
||
// be rejected at init — without this guard a site can configure e.g. a Float64
|
||
// tag, see the node appear in the address space via DiscoverAsync, and get
|
||
// BadNotSupported on every access. Half-implemented types must not leak into the
|
||
// configurable surface (Driver.S7-013). Drop entries from the set as each data
|
||
// type is wired through.
|
||
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);
|
||
|
||
Plc = plc;
|
||
|
||
// 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
|
||
// also rejects bit-offset > 7, DB 0, unknown area letters, etc.
|
||
_tagsByName.Clear();
|
||
_parsedByName.Clear();
|
||
_resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses
|
||
foreach (var t in _options.Tags)
|
||
{
|
||
var parsed = S7AddressParser.Parse(t.Address); // throws FormatException
|
||
_tagsByName[t.Name] = t;
|
||
_parsedByName[t.Name] = parsed;
|
||
}
|
||
|
||
_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);
|
||
|
||
// 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 (Driver.S7-006). 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);
|
||
}
|
||
}
|
||
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 { }
|
||
Plc = null;
|
||
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
|
||
_logger.LogError(ex, "S7Driver connect failed. Driver={DriverInstanceId} Host={Host}", _driverInstanceId, _options.Host);
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>Reinitializes the driver with a new configuration.</summary>
|
||
/// <param name="driverConfigJson">JSON configuration string.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||
{
|
||
// InitializeAsync re-parses driverConfigJson, so a config change delivered here is
|
||
// applied in place rather than silently discarded (Driver.S7-011).
|
||
await ShutdownAsync(cancellationToken).ConfigureAwait(false);
|
||
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
|
||
}
|
||
|
||
/// <summary>Shuts down the driver and releases resources.</summary>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
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 — see code-review finding Driver.S7-006.
|
||
var drain = new List<Task>();
|
||
|
||
var probeCts = _probeCts;
|
||
var probeTask = _probeTask;
|
||
try { probeCts?.Cancel(); } catch { }
|
||
if (probeTask is not null) drain.Add(probeTask);
|
||
|
||
var subscriptions = _subscriptions.Values.ToArray();
|
||
_subscriptions.Clear();
|
||
foreach (var state in subscriptions)
|
||
{
|
||
try { state.Cts.Cancel(); } catch { }
|
||
drain.Add(state.PollTask);
|
||
}
|
||
|
||
if (drain.Count > 0)
|
||
{
|
||
try
|
||
{
|
||
await Task.WhenAll(drain).WaitAsync(DrainTimeout, CancellationToken.None)
|
||
.ConfigureAwait(false);
|
||
}
|
||
catch (TimeoutException) { /* a wedged loop — proceed; better than leaking the teardown */ }
|
||
catch { /* loop faults are already surfaced via health; teardown continues */ }
|
||
}
|
||
|
||
// Loops have now observed cancellation and released _gate — safe to dispose the CTSs.
|
||
probeCts?.Dispose();
|
||
_probeCts = null;
|
||
_probeTask = null;
|
||
foreach (var state in subscriptions)
|
||
state.Cts.Dispose();
|
||
|
||
try { Plc?.Close(); } catch { /* best-effort — tearing down anyway */ }
|
||
Plc = null;
|
||
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
|
||
}
|
||
|
||
/// <summary>Gets the current driver health.</summary>
|
||
/// <returns>The current health state.</returns>
|
||
public DriverHealth GetHealth() => _health;
|
||
|
||
/// <summary>Flushes optional caches to free memory.</summary>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||
|
||
/// <summary>
|
||
/// True when <paramref name="driverConfigJson"/> carries a real config body. The
|
||
/// bootstrapper always passes a populated document; some unit tests pass <c>"{}"</c> or
|
||
/// an empty string to exercise lifecycle shape without a config — those keep the
|
||
/// constructor-supplied <see cref="_options"/>.
|
||
/// </summary>
|
||
private static bool HasConfigBody(string? driverConfigJson)
|
||
{
|
||
if (string.IsNullOrWhiteSpace(driverConfigJson)) return false;
|
||
var trimmed = driverConfigJson.Trim();
|
||
return trimmed is not "{}" and not "[]";
|
||
}
|
||
|
||
/// <summary>
|
||
/// Rejects tag addresses the read path cannot serve. Timer (<c>T{n}</c>) and Counter
|
||
/// (<c>C{n}</c>) addresses parse cleanly via <see cref="S7AddressParser"/> but
|
||
/// <see cref="ReadOneAsync"/> has no decode case for them and <see cref="S7DataType"/>
|
||
/// has no Timer/Counter member — left unguarded they fail fast init's promise and throw
|
||
/// a misleading type-mismatch on every read instead (code-review finding Driver.S7-001).
|
||
/// </summary>
|
||
private void RejectUnsupportedTagAddresses()
|
||
{
|
||
foreach (var t in _options.Tags)
|
||
{
|
||
if (S7AddressParser.TryParse(t.Address, out var parsed)
|
||
&& parsed.Area is S7Area.Timer or S7Area.Counter)
|
||
{
|
||
throw new NotSupportedException(
|
||
$"S7 tag '{t.Name}' uses a {parsed.Area} address ('{t.Address}'); " +
|
||
"Timer/Counter tags are not yet supported by the S7 driver. " +
|
||
"Remove the tag or use a DB/M/I/Q address until Timer/Counter reads are wired through.");
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Rejects tags configured with an <see cref="S7DataType"/> that
|
||
/// <see cref="ReinterpretRawValue"/> / <see cref="BoxValueForWrite"/> still throw
|
||
/// <see cref="NotSupportedException"/> for. Without this guard those tags create live
|
||
/// OPC UA nodes via <see cref="DiscoverAsync"/> but every Read/Write returns
|
||
/// <c>BadNotSupported</c> — code-review finding Driver.S7-013. Drop entries from
|
||
/// <see cref="UnimplementedDataTypes"/> as each type is wired through.
|
||
/// </summary>
|
||
private void RejectUnsupportedTagDataTypes()
|
||
{
|
||
foreach (var t in _options.Tags)
|
||
{
|
||
if (UnimplementedDataTypes.Contains(t.DataType))
|
||
{
|
||
throw new NotSupportedException(
|
||
$"S7 tag '{t.Name}' uses data type '{t.DataType}' which is not yet " +
|
||
"supported by the S7 driver — Read/Write would return BadNotSupported. " +
|
||
"Remove the tag or use Bool/Byte/Int16/UInt16/Int32/UInt32/Float32 until " +
|
||
$"{t.DataType} is wired through.");
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// S7DataType members that the read/write helpers throw NotSupportedException for.
|
||
/// Kept here (rather than reflecting over <see cref="ReinterpretRawValue"/>) so
|
||
/// <see cref="RejectUnsupportedTagDataTypes"/> is a single grep target for the
|
||
/// follow-up PR that wires each through.
|
||
/// </summary>
|
||
private static readonly HashSet<S7DataType> UnimplementedDataTypes = new()
|
||
{
|
||
S7DataType.Int64,
|
||
S7DataType.UInt64,
|
||
S7DataType.Float64,
|
||
S7DataType.String,
|
||
S7DataType.DateTime,
|
||
};
|
||
|
||
/// <summary>
|
||
/// Approximate memory footprint. The Plc instance + one 240-960 byte PDU buffer is
|
||
/// under 4 KB; return 0 because the <see cref="IDriver"/> contract asks for a
|
||
/// driver-attributable growth number and S7.Net doesn't expose one.
|
||
/// </summary>
|
||
public long GetMemoryFootprint() => 0;
|
||
|
||
// ---- IReadable ----
|
||
|
||
/// <summary>Reads values from the specified tag references.</summary>
|
||
/// <param name="fullReferences">Tag references to read.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation returning a list of data value snapshots.</returns>
|
||
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
|
||
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
|
||
{
|
||
// Validate the list before RequirePlc() so a null argument produces an
|
||
// ArgumentNullException (consistent with DiscoverAsync) rather than an
|
||
// InvalidOperationException from the not-initialized check — Driver.S7-003.
|
||
ArgumentNullException.ThrowIfNull(fullReferences);
|
||
var plc = RequirePlc();
|
||
var now = DateTime.UtcNow;
|
||
var results = new DataValueSnapshot[fullReferences.Count];
|
||
|
||
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||
try
|
||
{
|
||
for (var i = 0; i < fullReferences.Count; i++)
|
||
{
|
||
var name = fullReferences[i];
|
||
if (!_resolver.TryResolve(name, out var tag))
|
||
{
|
||
results[i] = new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now);
|
||
continue;
|
||
}
|
||
try
|
||
{
|
||
var value = await ReadOneAsync(plc, tag, cancellationToken).ConfigureAwait(false);
|
||
results[i] = new DataValueSnapshot(value, 0u, now, now);
|
||
_health = new DriverHealth(DriverState.Healthy, now, null);
|
||
}
|
||
catch (NotSupportedException)
|
||
{
|
||
results[i] = new DataValueSnapshot(null, StatusBadNotSupported, null, now);
|
||
}
|
||
catch (PlcException pex) when (IsAccessDenied(pex))
|
||
{
|
||
// PUT/GET-disabled (S7-1200/1500) / access-protection — a permanent
|
||
// configuration fault, NOT a transient one. Blind retry is wasted effort,
|
||
// so map it to BadNotSupported and flag the driver as a config alert
|
||
// (Faulted) rather than Degraded — per driver-specs.md §5 and
|
||
// code-review finding Driver.S7-007.
|
||
results[i] = new DataValueSnapshot(null, StatusBadNotSupported, null, now);
|
||
_health = new DriverHealth(DriverState.Faulted, _health.LastSuccessfulRead,
|
||
"S7 access denied — enable PUT/GET communication in TIA Portal " +
|
||
$"(Protection & Security) for this CPU. PLC reported: {pex.Message}");
|
||
}
|
||
catch (PlcException pex)
|
||
{
|
||
// A genuine device-layer fault (CPU error, hardware fault) — transient
|
||
// enough to keep retrying; report BadDeviceFailure and degrade health.
|
||
results[i] = new DataValueSnapshot(null, StatusBadDeviceFailure, null, now);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, pex.Message);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
}
|
||
}
|
||
}
|
||
finally { _gate.Release(); }
|
||
return results;
|
||
}
|
||
|
||
private async Task<object> ReadOneAsync(Plc 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
|
||
// on demand. S7AddressParser.Parse throws FormatException on a bad address, which the
|
||
// caller's catch maps to BadCommunicationError — the same surface a bad authored tag
|
||
// would have hit at init (transient defs aren't init-validated).
|
||
var addr = _parsedByName.TryGetValue(tag.Name, out var parsed)
|
||
? parsed
|
||
: S7AddressParser.Parse(tag.Address);
|
||
|
||
// Array path: a tag with a declared count >= 1 reads a CONTIGUOUS block of
|
||
// count × element-bytes in a SINGLE round-trip (Plc.ReadBytesAsync), then decodes each
|
||
// element from its big-endian slice into an element-typed CLR array. The scalar path
|
||
// (count null) is left byte-for-byte unchanged below. A count of 1 IS a valid 1-element
|
||
// array (the foundation materialises a [1] OPC UA array node when isArray:true).
|
||
if (tag.ArrayCount is >= 1)
|
||
return await ReadArrayAsync(plc, tag, addr, ct).ConfigureAwait(false);
|
||
|
||
// S7.Net's string-based ReadAsync returns object where the boxed .NET type depends on
|
||
// the size suffix: DBX=bool, DBB=byte, DBW=ushort, DBD=uint. Our S7DataType enum
|
||
// specifies the SEMANTIC type (Int16 vs UInt16 vs Float32 etc.); the reinterpret below
|
||
// converts the raw unsigned boxed value into the requested type without issuing an
|
||
// extra PLC round-trip.
|
||
var raw = await plc.ReadAsync(tag.Address, ct).ConfigureAwait(false)
|
||
?? throw new System.IO.InvalidDataException($"S7.Net returned null for '{tag.Address}'");
|
||
|
||
return ReinterpretRawValue(tag, addr, raw);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Reads a 1-D array tag as ONE contiguous block (<c>count × element-bytes</c>) via
|
||
/// S7.Net's buffer-based <c>Plc.ReadBytesAsync(DataType, db, startByteAdr, count, ct)</c>
|
||
/// — a single PLC round-trip, NOT <c>N</c> string reads — then hands the raw byte block
|
||
/// 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)
|
||
{
|
||
var count = tag.ArrayCount!.Value;
|
||
var elementBytes = ElementByteSize(addr.Size);
|
||
var totalBytes = count * elementBytes;
|
||
|
||
// ReadBytesAsync addresses by (area, db, startByteOffset, byteCount). The parser already
|
||
// normalised the start to a BYTE offset (ByteOffset) for DB/M/I/Q; a Bit array starts at
|
||
// its byte and consumes one byte per element (byte-granular contiguous bit access). S7.Net
|
||
// transparently splits a > PDU-sized block into multiple wire requests, so the driver
|
||
// doesn't have to chunk.
|
||
var area = ToS7NetArea(addr.Area);
|
||
var block = await plc.ReadBytesAsync(area, addr.DbNumber, addr.ByteOffset, totalBytes, ct)
|
||
.ConfigureAwait(false)
|
||
?? throw new System.IO.InvalidDataException($"S7.Net returned null block for '{tag.Address}'");
|
||
|
||
return DecodeArrayBlock(tag, addr, block);
|
||
}
|
||
|
||
/// <summary>Width in bytes of one array element for the given access size. Bit elements are
|
||
/// byte-granular over the wire (one byte per bool), so they cost 1 byte each.</summary>
|
||
/// <param name="size">The parsed access width.</param>
|
||
/// <returns>Element byte size: Bit/Byte = 1, Word = 2, DWord = 4.</returns>
|
||
internal static int ElementByteSize(S7Size size) => size switch
|
||
{
|
||
S7Size.Bit => 1,
|
||
S7Size.Byte => 1,
|
||
S7Size.Word => 2,
|
||
S7Size.DWord => 4,
|
||
_ => throw new InvalidOperationException($"Unknown S7Size {size}"),
|
||
};
|
||
|
||
/// <summary>
|
||
/// Maps the driver's <see cref="S7Area"/> to S7.Net's <c>DataType</c> for the
|
||
/// buffer-based block read. Timer/Counter are rejected at init so they never reach the
|
||
/// array path.
|
||
/// </summary>
|
||
private static S7NetDataType ToS7NetArea(S7Area area) => area switch
|
||
{
|
||
S7Area.DataBlock => S7NetDataType.DataBlock,
|
||
S7Area.Memory => S7NetDataType.Memory,
|
||
S7Area.Input => S7NetDataType.Input,
|
||
S7Area.Output => S7NetDataType.Output,
|
||
_ => throw new NotSupportedException(
|
||
$"S7 area {area} is not supported for array block reads (Timer/Counter are rejected at init)"),
|
||
};
|
||
|
||
/// <summary>
|
||
/// Pure decode loop — turns a raw S7 (big-endian) byte block into an element-typed CLR
|
||
/// array (<c>short[]</c> / <c>ushort[]</c> / <c>int[]</c> / <c>uint[]</c> / <c>float[]</c>
|
||
/// / <c>byte[]</c> / <c>bool[]</c>), boxed as <see cref="object"/>. No network I/O —
|
||
/// factored out of <see cref="ReadArrayAsync"/> so the block-decode is unit-testable
|
||
/// against a known byte block without a live PLC (S7.Net ships no in-process fake).
|
||
/// Each element is read from its <c>i × element-bytes</c> slice using S7 big-endian byte
|
||
/// order, identical to the per-element semantics of <see cref="ReinterpretRawValue"/>.
|
||
/// </summary>
|
||
/// <param name="tag">Tag definition carrying the element <see cref="S7DataType"/> and array count.</param>
|
||
/// <param name="addr">Parsed address carrying the access <see cref="S7Size"/>.</param>
|
||
/// <param name="block">Raw contiguous byte block read from the PLC (length == count × element-bytes).</param>
|
||
/// <returns>An element-typed CLR array boxed as <see cref="object"/>.</returns>
|
||
internal static object DecodeArrayBlock(S7TagDefinition tag, S7ParsedAddress addr, byte[] block)
|
||
{
|
||
var count = tag.ArrayCount!.Value;
|
||
var elementBytes = ElementByteSize(addr.Size);
|
||
|
||
switch (tag.DataType, addr.Size)
|
||
{
|
||
case (S7DataType.Bool, S7Size.Bit):
|
||
{
|
||
var a = new bool[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = (block[i] & 0x01) != 0;
|
||
return a;
|
||
}
|
||
case (S7DataType.Byte, S7Size.Byte):
|
||
{
|
||
var a = new byte[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = block[i];
|
||
return a;
|
||
}
|
||
case (S7DataType.UInt16, S7Size.Word):
|
||
{
|
||
var a = new ushort[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = ReadBeUInt16(block, i * elementBytes);
|
||
return a;
|
||
}
|
||
case (S7DataType.Int16, S7Size.Word):
|
||
{
|
||
var a = new short[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = unchecked((short)ReadBeUInt16(block, i * elementBytes));
|
||
return a;
|
||
}
|
||
case (S7DataType.UInt32, S7Size.DWord):
|
||
{
|
||
var a = new uint[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = ReadBeUInt32(block, i * elementBytes);
|
||
return a;
|
||
}
|
||
case (S7DataType.Int32, S7Size.DWord):
|
||
{
|
||
var a = new int[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = unchecked((int)ReadBeUInt32(block, i * elementBytes));
|
||
return a;
|
||
}
|
||
case (S7DataType.Float32, S7Size.DWord):
|
||
{
|
||
var a = new float[count];
|
||
for (var i = 0; i < count; i++)
|
||
a[i] = BitConverter.UInt32BitsToSingle(ReadBeUInt32(block, i * elementBytes));
|
||
return a;
|
||
}
|
||
|
||
case (S7DataType.Int64, _):
|
||
case (S7DataType.UInt64, _):
|
||
case (S7DataType.Float64, _):
|
||
case (S7DataType.String, _):
|
||
case (S7DataType.DateTime, _):
|
||
throw new NotSupportedException(
|
||
$"S7 array reads of {tag.DataType} land in a follow-up PR");
|
||
|
||
default:
|
||
throw new System.IO.InvalidDataException(
|
||
$"S7 array Read type-mismatch: tag '{tag.Name}' declared {tag.DataType} but address " +
|
||
$"'{tag.Address}' parsed as Size={addr.Size}");
|
||
}
|
||
}
|
||
|
||
/// <summary>Reads a big-endian 16-bit word from <paramref name="block"/> at <paramref name="offset"/>.</summary>
|
||
private static ushort ReadBeUInt16(byte[] block, int offset) =>
|
||
(ushort)((block[offset] << 8) | block[offset + 1]);
|
||
|
||
/// <summary>Reads a big-endian 32-bit dword from <paramref name="block"/> at <paramref name="offset"/>.</summary>
|
||
private static uint ReadBeUInt32(byte[] block, int offset) =>
|
||
((uint)block[offset] << 24) | ((uint)block[offset + 1] << 16)
|
||
| ((uint)block[offset + 2] << 8) | block[offset + 3];
|
||
|
||
/// <summary>
|
||
/// Pure reinterpret step — converts the boxed value that S7.Net returns (always an
|
||
/// unsigned type: <c>bool</c>, <c>byte</c>, <c>ushort</c>, <c>uint</c>) into the
|
||
/// SEMANTIC type declared by the tag's <see cref="S7DataType"/>. No network I/O.
|
||
/// Factored out of <see cref="ReadOneAsync"/> so it can be exercised in unit tests
|
||
/// without a live PLC (Driver.S7-014).
|
||
/// </summary>
|
||
/// <param name="tag">Tag definition containing type information.</param>
|
||
/// <param name="addr">Parsed tag address.</param>
|
||
/// <param name="raw">Raw value from S7.Net.</param>
|
||
/// <returns>The reinterpreted value in the target semantic type.</returns>
|
||
internal static object ReinterpretRawValue(S7TagDefinition tag, S7ParsedAddress addr, object raw) =>
|
||
(tag.DataType, addr.Size, raw) switch
|
||
{
|
||
(S7DataType.Bool, S7Size.Bit, bool b) => b,
|
||
(S7DataType.Byte, S7Size.Byte, byte by) => by,
|
||
(S7DataType.UInt16, S7Size.Word, ushort u16) => u16,
|
||
(S7DataType.Int16, S7Size.Word, ushort u16) => unchecked((short)u16),
|
||
(S7DataType.UInt32, S7Size.DWord, uint u32) => u32,
|
||
(S7DataType.Int32, S7Size.DWord, uint u32) => unchecked((int)u32),
|
||
(S7DataType.Float32, S7Size.DWord, uint u32) => BitConverter.UInt32BitsToSingle(u32),
|
||
|
||
(S7DataType.Int64, _, _) => throw new NotSupportedException("S7 Int64 reads land in a follow-up PR"),
|
||
(S7DataType.UInt64, _, _) => throw new NotSupportedException("S7 UInt64 reads land in a follow-up PR"),
|
||
(S7DataType.Float64, _, _) => throw new NotSupportedException("S7 Float64 (LReal) reads land in a follow-up PR"),
|
||
(S7DataType.String, _, _) => throw new NotSupportedException("S7 STRING reads land in a follow-up PR"),
|
||
(S7DataType.DateTime, _, _) => throw new NotSupportedException("S7 DateTime reads land in a follow-up PR"),
|
||
|
||
_ => throw new System.IO.InvalidDataException(
|
||
$"S7 Read type-mismatch: tag '{tag.Name}' declared {tag.DataType} but address '{tag.Address}' " +
|
||
$"parsed as Size={addr.Size}; S7.Net returned {raw.GetType().Name}"),
|
||
};
|
||
|
||
// ---- IWritable ----
|
||
|
||
/// <summary>Writes values to the specified tags.</summary>
|
||
/// <param name="writes">Write requests containing tag references and values.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation returning a list of write results.</returns>
|
||
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
|
||
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
|
||
{
|
||
// Same as ReadAsync — validate before RequirePlc() so a null argument is a
|
||
// typed argument error, not the "not initialized" surface (Driver.S7-003).
|
||
ArgumentNullException.ThrowIfNull(writes);
|
||
var plc = RequirePlc();
|
||
var results = new WriteResult[writes.Count];
|
||
|
||
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||
try
|
||
{
|
||
for (var i = 0; i < writes.Count; i++)
|
||
{
|
||
var w = writes[i];
|
||
if (!_resolver.TryResolve(w.FullReference, out var tag))
|
||
{
|
||
results[i] = new WriteResult(StatusBadNodeIdUnknown);
|
||
continue;
|
||
}
|
||
if (!tag.Writable)
|
||
{
|
||
results[i] = new WriteResult(StatusBadNotWritable);
|
||
continue;
|
||
}
|
||
try
|
||
{
|
||
await WriteOneAsync(plc, tag, w.Value, cancellationToken).ConfigureAwait(false);
|
||
results[i] = new WriteResult(0u);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// Driver.S7-008: let cancellation propagate rather than turning it into
|
||
// a status code — the gate is still held so Release() runs in finally.
|
||
throw;
|
||
}
|
||
catch (NotSupportedException)
|
||
{
|
||
results[i] = new WriteResult(StatusBadNotSupported);
|
||
}
|
||
catch (PlcException pex) when (IsAccessDenied(pex))
|
||
{
|
||
// PUT/GET-disabled / access-protection on write — same permanent
|
||
// configuration fault as on read (Driver.S7-007). BadNotSupported +
|
||
// a config-alert health state, not a transient device failure.
|
||
results[i] = new WriteResult(StatusBadNotSupported);
|
||
_health = new DriverHealth(DriverState.Faulted, _health.LastSuccessfulRead,
|
||
"S7 access denied — enable PUT/GET communication in TIA Portal " +
|
||
$"(Protection & Security) for this CPU. PLC reported: {pex.Message}");
|
||
}
|
||
catch (PlcException pex)
|
||
{
|
||
// 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 — Driver.S7-008).
|
||
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 — Driver.S7-008.
|
||
results[i] = new WriteResult(StatusBadCommunicationError);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
}
|
||
}
|
||
}
|
||
finally { _gate.Release(); }
|
||
return results;
|
||
}
|
||
|
||
private async Task WriteOneAsync(Plc plc, S7TagDefinition tag, object? value, CancellationToken ct)
|
||
{
|
||
// S7.Net's Plc.WriteAsync(string address, object value) expects the boxed value to
|
||
// match the address's size-suffix type: DBX=bool, DBB=byte, DBW=ushort, DBD=uint.
|
||
// Our S7DataType lets the caller pass short/int/float; convert to the unsigned
|
||
// wire representation before handing off.
|
||
var boxed = BoxValueForWrite(tag.DataType, value);
|
||
await plc.WriteAsync(tag.Address, boxed, ct).ConfigureAwait(false);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Pure boxing step — converts the caller's value into the unsigned wire type that
|
||
/// S7.Net's <c>Plc.WriteAsync</c> expects for each address size (bool → bool, byte
|
||
/// → byte, short → ushort, int → uint, float → uint-bits). No network I/O.
|
||
/// Factored out of <see cref="WriteOneAsync"/> so it can be exercised in unit tests
|
||
/// without a live PLC (Driver.S7-014).
|
||
/// </summary>
|
||
/// <param name="dataType">Target S7 data type.</param>
|
||
/// <param name="value">Value to box.</param>
|
||
/// <returns>The boxed value in the wire type expected by S7.Net.</returns>
|
||
internal static object BoxValueForWrite(S7DataType dataType, object? value) => dataType switch
|
||
{
|
||
S7DataType.Bool => (object)Convert.ToBoolean(value),
|
||
S7DataType.Byte => (object)Convert.ToByte(value),
|
||
S7DataType.UInt16 => (object)Convert.ToUInt16(value),
|
||
S7DataType.Int16 => (object)unchecked((ushort)Convert.ToInt16(value)),
|
||
S7DataType.UInt32 => (object)Convert.ToUInt32(value),
|
||
S7DataType.Int32 => (object)unchecked((uint)Convert.ToInt32(value)),
|
||
S7DataType.Float32 => (object)BitConverter.SingleToUInt32Bits(Convert.ToSingle(value)),
|
||
|
||
S7DataType.Int64 => throw new NotSupportedException("S7 Int64 writes land in a follow-up PR"),
|
||
S7DataType.UInt64 => throw new NotSupportedException("S7 UInt64 writes land in a follow-up PR"),
|
||
S7DataType.Float64 => throw new NotSupportedException("S7 Float64 (LReal) writes land in a follow-up PR"),
|
||
S7DataType.String => throw new NotSupportedException("S7 STRING writes land in a follow-up PR"),
|
||
S7DataType.DateTime => throw new NotSupportedException("S7 DateTime writes land in a follow-up PR"),
|
||
_ => throw new InvalidOperationException($"Unknown S7DataType {dataType}"),
|
||
};
|
||
|
||
private Plc RequirePlc() =>
|
||
Plc ?? throw new InvalidOperationException("S7Driver not initialized");
|
||
|
||
/// <summary>
|
||
/// Detects an S7 PUT/GET-disabled / access-protection fault inside an S7.Net
|
||
/// <see cref="PlcException"/>. S7.Net's read/write paths wrap every PLC-side error in a
|
||
/// <c>PlcException</c> with <see cref="ErrorCode.ReadData"/> / <see cref="ErrorCode.WriteData"/>;
|
||
/// the response-code validator throws a plain <see cref="Exception"/> for the S7
|
||
/// <c>AccessingObjectNotAllowed</c> status, which lands as the inner exception. There is
|
||
/// no typed error code for it, so the inner message is the only discriminator
|
||
/// S7.Net exposes — see code-review finding Driver.S7-007.
|
||
/// </summary>
|
||
private static bool IsAccessDenied(PlcException pex)
|
||
{
|
||
for (Exception? e = pex; e is not null; e = e.InnerException)
|
||
{
|
||
if (e.Message.Contains("Accessing object not allowed", StringComparison.OrdinalIgnoreCase)
|
||
|| e.Message.Contains("not allowed", StringComparison.OrdinalIgnoreCase))
|
||
{
|
||
return true;
|
||
}
|
||
}
|
||
return false;
|
||
}
|
||
|
||
// ---- ITagDiscovery ----
|
||
|
||
/// <summary>Discovers tags and builds the OPC UA address space.</summary>
|
||
/// <param name="builder">Address space builder.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(builder);
|
||
var folder = builder.Folder("S7", "S7");
|
||
foreach (var t in _options.Tags)
|
||
{
|
||
// 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
|
||
// materialises a [1] OPC UA array node when isArray:true, so the driver must agree.
|
||
var isArray = t.ArrayCount is >= 1;
|
||
folder.Variable(t.Name, t.Name, new DriverAttributeInfo(
|
||
FullName: t.Name,
|
||
DriverDataType: MapDataType(t.DataType),
|
||
IsArray: isArray,
|
||
ArrayDim: isArray ? (uint)t.ArrayCount!.Value : null,
|
||
SecurityClass: t.Writable ? SecurityClassification.Operate : SecurityClassification.ViewOnly,
|
||
IsHistorized: false,
|
||
IsAlarm: false,
|
||
WriteIdempotent: t.WriteIdempotent));
|
||
}
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
private static DriverDataType MapDataType(S7DataType t) => t switch
|
||
{
|
||
S7DataType.Bool => DriverDataType.Boolean,
|
||
S7DataType.Byte => DriverDataType.Int32, // no 8-bit in DriverDataType yet
|
||
// Driver.S7-002: UInt32 values > int.MaxValue (2^31-1) wrap negative when surfaced as
|
||
// Int32. This is lossy in the same way as the Int64/UInt64 mapping below — both are
|
||
// acknowledged limitations until unsigned DriverDataType members ship.
|
||
S7DataType.Int16 or S7DataType.UInt16 or S7DataType.Int32 or S7DataType.UInt32 => DriverDataType.Int32,
|
||
S7DataType.Int64 or S7DataType.UInt64 => DriverDataType.Int32, // lossy for values > 2^31-1; tracked for follow-up
|
||
S7DataType.Float32 => DriverDataType.Float32,
|
||
S7DataType.Float64 => DriverDataType.Float64,
|
||
S7DataType.String => DriverDataType.String,
|
||
S7DataType.DateTime => DriverDataType.DateTime,
|
||
_ => DriverDataType.Int32,
|
||
};
|
||
|
||
// ---- ISubscribable (polling overlay) ----
|
||
|
||
/// <summary>Subscribes to changes on the specified tag references.</summary>
|
||
/// <param name="fullReferences">Tag references to subscribe to.</param>
|
||
/// <param name="publishingInterval">Polling interval.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation returning a subscription handle.</returns>
|
||
public Task<ISubscriptionHandle> SubscribeAsync(
|
||
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
|
||
{
|
||
var id = Interlocked.Increment(ref _nextSubscriptionId);
|
||
var cts = new CancellationTokenSource();
|
||
// Floor at 100 ms — S7 CPUs scan 2-10 ms but the comms mailbox is processed at most
|
||
// once per scan; sub-100 ms polling just queues wire-side with worse latency.
|
||
var interval = publishingInterval < TimeSpan.FromMilliseconds(100)
|
||
? TimeSpan.FromMilliseconds(100)
|
||
: publishingInterval;
|
||
var handle = new S7SubscriptionHandle(id);
|
||
var state = new SubscriptionState(handle, [.. fullReferences], interval, cts);
|
||
_subscriptions[id] = state;
|
||
// Track the poll Task so ShutdownAsync can await it after cancelling — a poll
|
||
// iteration mid-_gate would otherwise race the semaphore's disposal (Driver.S7-006).
|
||
state.PollTask = Task.Run(() => PollLoopAsync(state, cts.Token), CancellationToken.None);
|
||
return Task.FromResult<ISubscriptionHandle>(handle);
|
||
}
|
||
|
||
/// <summary>Unsubscribes from a subscription.</summary>
|
||
/// <param name="handle">Subscription handle.</param>
|
||
/// <param name="cancellationToken">Cancellation token.</param>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
|
||
{
|
||
if (handle is S7SubscriptionHandle h && _subscriptions.TryRemove(h.Id, out var state))
|
||
{
|
||
state.Cts.Cancel();
|
||
state.Cts.Dispose();
|
||
}
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Upper bound on the poll-loop backoff window. After enough consecutive failures the
|
||
/// loop waits this long between retries instead of <see cref="SubscriptionState.Interval"/>,
|
||
/// so a subscription against a dropped / uninitialised driver doesn't spin (Driver.S7-009).
|
||
/// </summary>
|
||
private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30);
|
||
|
||
/// <summary>
|
||
/// Number of consecutive poll failures before the loop transitions the driver's
|
||
/// health to <see cref="DriverState.Degraded"/>. One stray failure can be transient;
|
||
/// a sustained run indicates the operator should see it. Threshold of 1 because the
|
||
/// first failure already lives in the LastError surface — see Driver.S7-009.
|
||
/// </summary>
|
||
private const int PollFailureHealthThreshold = 1;
|
||
|
||
private async Task PollLoopAsync(SubscriptionState state, CancellationToken ct)
|
||
{
|
||
var consecutiveFailures = 0;
|
||
|
||
// Initial-data push per OPC UA Part 4 convention.
|
||
try
|
||
{
|
||
await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false);
|
||
consecutiveFailures = 0;
|
||
}
|
||
catch (OperationCanceledException) { return; }
|
||
catch (Exception ex)
|
||
{
|
||
// First-read error — polling continues; log so the operator has an event trail.
|
||
consecutiveFailures++;
|
||
HandlePollFailure(ex, consecutiveFailures, initial: true);
|
||
}
|
||
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
// Capped exponential backoff: Interval, 2×, 4×, ... up to PollBackoffCap. Healthy
|
||
// ticks reset consecutiveFailures back to 0 so the cadence snaps back to Interval.
|
||
var delay = ComputeBackoffDelay(state.Interval, consecutiveFailures);
|
||
try { await Task.Delay(delay, ct).ConfigureAwait(false); }
|
||
catch (OperationCanceledException) { return; }
|
||
|
||
try
|
||
{
|
||
await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false);
|
||
consecutiveFailures = 0;
|
||
}
|
||
catch (OperationCanceledException) { return; }
|
||
catch (Exception ex)
|
||
{
|
||
// Sustained polling error — loop continues with backoff; log + update health.
|
||
consecutiveFailures++;
|
||
HandlePollFailure(ex, consecutiveFailures, initial: false);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Logs the swallowed poll exception and, once <see cref="PollFailureHealthThreshold"/>
|
||
/// consecutive failures have accumulated, degrades the driver health so the failure
|
||
/// surfaces on the dashboard — see Driver.S7-009. The probe loop owns Running/Stopped
|
||
/// transitions for the host-connectivity surface, so we touch <see cref="_health"/>
|
||
/// rather than the probe state.
|
||
/// </summary>
|
||
private void HandlePollFailure(Exception ex, int consecutiveFailures, bool initial)
|
||
{
|
||
if (initial)
|
||
_logger.LogWarning(ex, "S7 poll initial-read failed. Driver={DriverInstanceId} ConsecutiveFailures={Count}",
|
||
_driverInstanceId, consecutiveFailures);
|
||
else
|
||
_logger.LogWarning(ex, "S7 poll tick failed. Driver={DriverInstanceId} ConsecutiveFailures={Count}",
|
||
_driverInstanceId, consecutiveFailures);
|
||
|
||
if (consecutiveFailures >= PollFailureHealthThreshold)
|
||
{
|
||
// Don't downgrade a Faulted state (e.g. PUT/GET-denied set by ReadAsync) — Faulted
|
||
// is a stronger signal than Degraded and is reserved for permanent config faults.
|
||
if (_health.State != DriverState.Faulted)
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Capped exponential backoff. <c>consecutiveFailures == 0</c> returns the configured
|
||
/// <paramref name="interval"/>; each subsequent failure doubles the wait up to
|
||
/// <see cref="PollBackoffCap"/>. Computed in ticks to avoid overflow at large counts.
|
||
/// </summary>
|
||
/// <param name="interval">Base polling interval.</param>
|
||
/// <param name="consecutiveFailures">Number of consecutive failures.</param>
|
||
/// <returns>The computed backoff delay.</returns>
|
||
internal static TimeSpan ComputeBackoffDelay(TimeSpan interval, int consecutiveFailures)
|
||
{
|
||
if (consecutiveFailures <= 0) return interval;
|
||
// Cap the shift to avoid overflow — at 30 the result already saturates PollBackoffCap
|
||
// for any reasonable Interval.
|
||
var shift = Math.Min(consecutiveFailures - 1, 30);
|
||
var ticks = interval.Ticks << shift;
|
||
if (ticks <= 0 || ticks > PollBackoffCap.Ticks) return PollBackoffCap;
|
||
return TimeSpan.FromTicks(ticks);
|
||
}
|
||
|
||
private async Task PollOnceAsync(SubscriptionState state, bool forceRaise, CancellationToken ct)
|
||
{
|
||
var snapshots = await ReadAsync(state.TagReferences, ct).ConfigureAwait(false);
|
||
for (var i = 0; i < state.TagReferences.Count; i++)
|
||
{
|
||
var tagRef = state.TagReferences[i];
|
||
var current = snapshots[i];
|
||
var lastSeen = state.LastValues.TryGetValue(tagRef, out var prev) ? prev : default;
|
||
|
||
if (forceRaise || !Equals(lastSeen?.Value, current.Value) || lastSeen?.StatusCode != current.StatusCode)
|
||
{
|
||
state.LastValues[tagRef] = current;
|
||
OnDataChange?.Invoke(this, new DataChangeEventArgs(state.Handle, tagRef, current));
|
||
}
|
||
}
|
||
}
|
||
|
||
private sealed record SubscriptionState(
|
||
S7SubscriptionHandle Handle,
|
||
IReadOnlyList<string> TagReferences,
|
||
TimeSpan Interval,
|
||
CancellationTokenSource Cts)
|
||
{
|
||
/// <summary>Gets the last known values for subscribed tags.</summary>
|
||
public ConcurrentDictionary<string, DataValueSnapshot> LastValues { get; }
|
||
= new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
/// <summary>
|
||
/// Handle to this subscription's poll loop. Tracked so <see cref="ShutdownAsync"/>
|
||
/// can await it after cancelling — see code-review finding Driver.S7-006.
|
||
/// </summary>
|
||
public Task PollTask { get; set; } = Task.CompletedTask;
|
||
}
|
||
|
||
private sealed record S7SubscriptionHandle(long Id) : ISubscriptionHandle
|
||
{
|
||
/// <summary>Gets the diagnostic identifier for this subscription.</summary>
|
||
public string DiagnosticId => $"s7-sub-{Id}";
|
||
}
|
||
|
||
// ---- IHostConnectivityProbe ----
|
||
|
||
/// <summary>
|
||
/// Host identifier surfaced in <see cref="GetHostStatuses"/>. <c>host:port</c> format
|
||
/// matches the Modbus driver's convention so the Admin UI dashboard renders both
|
||
/// family's rows uniformly.
|
||
/// </summary>
|
||
public string HostName => $"{_options.Host}:{_options.Port}";
|
||
|
||
/// <summary>Gets the host connectivity statuses.</summary>
|
||
/// <returns>A list containing the current host status.</returns>
|
||
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
|
||
{
|
||
lock (_probeLock)
|
||
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
|
||
}
|
||
|
||
private async Task ProbeLoopAsync(CancellationToken ct)
|
||
{
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
var success = false;
|
||
try
|
||
{
|
||
// Probe via S7.Net's low-cost GetCpuStatus — returns the CPU state (Run/Stop)
|
||
// and is intentionally light on the comms mailbox. Single-word Plc.ReadAsync
|
||
// would also work but GetCpuStatus doubles as a "PLC actually up" check.
|
||
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);
|
||
success = true;
|
||
}
|
||
finally { _gate.Release(); }
|
||
}
|
||
catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
|
||
catch { /* transport/timeout/exception — treated as Stopped below */ }
|
||
|
||
TransitionTo(success ? HostState.Running : HostState.Stopped);
|
||
|
||
try { await Task.Delay(_options.Probe.Interval, ct).ConfigureAwait(false); }
|
||
catch (OperationCanceledException) { return; }
|
||
}
|
||
}
|
||
|
||
private void TransitionTo(HostState newState)
|
||
{
|
||
HostState old;
|
||
lock (_probeLock)
|
||
{
|
||
old = _hostState;
|
||
if (old == newState) return;
|
||
_hostState = newState;
|
||
_hostStateChangedUtc = DateTime.UtcNow;
|
||
}
|
||
_logger.LogInformation("S7 probe transition. Driver={DriverInstanceId} Host={Host} {OldState} → {NewState}",
|
||
_driverInstanceId, HostName, old, newState);
|
||
OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState));
|
||
}
|
||
|
||
/// <summary>Disposes the driver and releases resources.</summary>
|
||
public void Dispose()
|
||
{
|
||
// Driver.S7-010: avoid the sync-over-async DisposeAsync().AsTask().GetAwaiter().GetResult()
|
||
// pattern (a known deadlock surface even when currently safe here). ShutdownAsync's
|
||
// body is effectively synchronous apart from waiting on probe/poll Tasks; do the same
|
||
// teardown directly, blocking only on the drain — and only with a bounded timeout so
|
||
// a wedged loop can't hang Dispose() indefinitely.
|
||
if (_disposed) return;
|
||
_disposed = true;
|
||
SynchronousTeardown();
|
||
_gate.Dispose();
|
||
}
|
||
|
||
/// <summary>Asynchronously disposes the driver and releases resources.</summary>
|
||
/// <returns>A task representing the asynchronous operation.</returns>
|
||
public async ValueTask DisposeAsync()
|
||
{
|
||
if (_disposed) return;
|
||
_disposed = true;
|
||
try { await ShutdownAsync(CancellationToken.None).ConfigureAwait(false); }
|
||
catch { /* disposal is best-effort */ }
|
||
_gate.Dispose();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Synchronous teardown — mirrors <see cref="ShutdownAsync"/> but blocks (with a bounded
|
||
/// timeout) on the probe + poll Tasks instead of awaiting them. Used by the sync
|
||
/// <see cref="Dispose"/> path so we don't sync-over-async <see cref="DisposeAsync"/>
|
||
/// (Driver.S7-010).
|
||
/// </summary>
|
||
private void SynchronousTeardown()
|
||
{
|
||
var drain = new List<Task>();
|
||
|
||
var probeCts = _probeCts;
|
||
var probeTask = _probeTask;
|
||
try { probeCts?.Cancel(); } catch { }
|
||
if (probeTask is not null) drain.Add(probeTask);
|
||
|
||
var subscriptions = _subscriptions.Values.ToArray();
|
||
_subscriptions.Clear();
|
||
foreach (var state in subscriptions)
|
||
{
|
||
try { state.Cts.Cancel(); } catch { }
|
||
drain.Add(state.PollTask);
|
||
}
|
||
|
||
if (drain.Count > 0)
|
||
{
|
||
try { Task.WhenAll(drain).Wait(DrainTimeout); }
|
||
catch { /* timeouts/loop faults are tolerated — teardown continues */ }
|
||
}
|
||
|
||
probeCts?.Dispose();
|
||
_probeCts = null;
|
||
_probeTask = null;
|
||
foreach (var state in subscriptions)
|
||
{
|
||
try { state.Cts.Dispose(); } catch { }
|
||
}
|
||
|
||
try { Plc?.Close(); } catch { /* best-effort — tearing down anyway */ }
|
||
Plc = null;
|
||
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
|
||
}
|
||
}
|