using System.Buffers.Binary;
using System.Collections.Concurrent;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
///
/// Modbus TCP implementation of + +
/// + . First native-protocol greenfield
/// driver for the v2 stack — validates the driver-agnostic IAddressSpaceBuilder +
/// IReadable/IWritable abstractions generalize beyond Galaxy.
///
///
/// Scope limits: Historian + alarm capabilities are out of scope (the protocol doesn't
/// express them). Subscriptions overlay a polling loop via the shared
/// since Modbus has no native push model.
///
public sealed class ModbusDriver
: IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable
{
/// Authored tags this driver could not build into its poll table.
private readonly SkippedTagLedger _skippedTags = new();
///
public SkippedTagReport? GetSkippedTags() => _skippedTags.Snapshot();
// ---- instance fields (grouped at top for auditability) ----
/// Mutable so can adopt a re-parsed config on reinitialize.
/// Only ever written on the init path, before any reader/transport uses it.
private ModbusDriverOptions _options;
private readonly Func _transportFactory;
private readonly string _driverInstanceId;
private readonly ILogger _logger;
// Polled subscriptions delegate to the shared PollGroupEngine. The driver only supplies
// the reader + on-change bridge; the engine owns the loop, interval floor, and lifecycle.
private readonly PollGroupEngine _poll;
// v3 authored-tag table, keyed by RawPath (the tag identity + driver wire reference). RawPath is
// case-sensitive ordinal. Built in InitializeAsync from _options.RawTags via the pure mapper.
// BUILD-AND-SWAP: a fresh dictionary is built off to the side and published by reference
// (Volatile.Write, under _stateLock with the generation check); it is never mutated or .Clear()ed
// in place, because retained poll loops read it concurrently across a re-initialize.
private Dictionary _tagsByRawPath = new(StringComparer.Ordinal);
///
/// Lifecycle generation. Every InitializeAsync / ShutdownAsync bumps it under
/// ; every health write and every table/transport publish carries the
/// generation it was computed under and is DROPPED once the generation has moved on — so a
/// superseded (abandoned) initialize can never publish over a newer one, a read that straddled a
/// re-initialize cannot write health about the new transport, and a shutdown's final health
/// write cannot clobber a newer init.
///
private int _generation;
/// Guards , the transport/table publish and the health fields.
private readonly object _stateLock = new();
// ONE comms-fault slot for reads AND writes: a fault is a property of the shared transport, so the
// latest batch of either kind decides — a good read clears a Degraded a write's timeout raised and vice
// versa (two independent sides latched a single write blip as Degraded forever in a read-mostly plant).
// LastSuccessfulRead advances on any evidence the device answered — a good read, a good write, a
// refusal, or a probe answer.
private string? _commsFault;
// Once-per-tag Warning for deterministic device refusals (exception 01/02/03, undecodable data).
// Replaced (never cleared in place) on every table publish.
private ConcurrentDictionary _refusalWarned = new(StringComparer.Ordinal);
// Resolves a read/write/subscribe RawPath reference to its tag definition via a single hit on the
// authored _tagsByRawPath table (v3: the reference is always a RawPath; a miss is a miss).
private readonly EquipmentTagRefResolver _resolver;
// Last-published (value, status) per tag, keyed by FullReference. Used by ShouldPublish to apply
// the deadband filter. The value is stored as object so all numeric types share one map; the
// comparison does a typed cast inside. The status is part of the baseline — see ShouldPublish.
private readonly ConcurrentDictionary _lastPublishedByRef = new(StringComparer.OrdinalIgnoreCase);
// Per-cycle buffer of the items this driver actually PUBLISHED (i.e. that passed ShouldPublish),
// keyed by subscription handle. The poll engine collects its own batch list unconditionally — it
// has no knowledge of the deadband filter — so the batch event is raised from this buffer instead,
// using the engine's onBatch purely as the end-of-cycle signal.
//
// No lock is needed on the per-handle List: the engine runs one loop task per subscription, so for
// ONE handle all of a cycle's onChange calls and that cycle's single onBatch run sequentially on
// that subscription's own task. Different subscriptions run concurrently — hence the per-handle
// ConcurrentDictionary.
private readonly ConcurrentDictionary> _cycleBatchItems = new();
// Last-written value per tag for the WriteOnChangeOnly suppression. Invalidated by reads
// that return a different value (so an HMI-side change doesn't get masked).
private readonly Dictionary _lastWrittenByRef = new(StringComparer.OrdinalIgnoreCase);
private readonly object _lastWrittenLock = new();
// BitInRegister writes need a read-modify-write against the full holding register. A
// per-register lock keeps concurrent bit-write callers from stomping on each other.
private readonly ConcurrentDictionary _rmwLocks = new();
private readonly Dictionary<(byte Unit, ModbusRegion Region, ushort Start, ushort End), ProhibitionState> _autoProhibited = new();
private readonly object _autoProhibitedLock = new();
// Single-host probe state — Modbus driver talks to exactly one endpoint so the "hosts"
// collection has at most one entry. HostName is the Host:Port string so the Admin UI can
// display the PLC endpoint uniformly with Galaxy platforms/engines.
private readonly object _probeLock = new();
private HostState _hostState = HostState.Unknown;
private DateTime _hostStateChangedUtc = DateTime.UtcNow;
private IModbusTransport? _transport;
private CancellationTokenSource? _probeCts;
private CancellationTokenSource? _reprobeCts;
private Task? _probeTask;
private Task? _reprobeTask;
/// Bounded join on the probe / re-probe loops at teardown — they honour their token, so this is a backstop.
private static readonly TimeSpan BackgroundLoopJoinTimeout = TimeSpan.FromSeconds(2);
///
/// Guards / /
/// / against each other. The three
/// fields above are published by one lifecycle operation and released by the next, so two
/// concurrent operations would dispose each other's freshly-built transport or null
/// _transport out from under an in-flight connect. Not hypothetical:
/// DriverInstanceActor dispatches initialize on a fire-and-forget Task.Run and
/// a reconnect retry tick can start a second attempt while the first is still running. Same shape as
/// MqttDriver._lifecycleGate.
///
private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
///
/// Bounded wait the teardown paths take on . A wedged initialize
/// (a connect that never returns) must not turn shutdown or dispose into a hang — past the
/// timeout they tear down anyway, which is the lesser evil of the two.
///
internal TimeSpan LifecycleGateTimeout { get; set; } = TimeSpan.FromSeconds(10);
private DriverHealth _health = new(DriverState.Unknown, null, null);
/// Occurs when a subscribed tag value changes.
public event EventHandler? OnDataChange;
///
/// Occurs once per poll cycle that produced at least one change, carrying that cycle's
/// changed tags. Both data-change events fire by design — this one once per cycle,
/// once per changed tag — so a consumer subscribes to EXACTLY
/// ONE of the two; subscribing both double-delivers.
///
public event EventHandler? OnDataChangeBatch;
/// Occurs when host connectivity status changes.
public event EventHandler? OnHostStatusChanged;
// ---- nested types ----
///
/// Per-prohibition state. SplitPending drives the re-probe loop's
/// bisection: when true and the range spans > 1 register, the next re-probe
/// tries the two halves separately to narrow the actual offending register(s).
/// Single-register prohibitions can't be split further; they stay re-probed as-is.
///
private sealed class ProhibitionState
{
public DateTime LastProbedUtc;
public bool SplitPending;
}
// ---- ctor + identity ----
/// Initializes a new Modbus TCP driver with the specified options and transport factory.
/// Driver configuration options.
/// Unique identifier for this driver instance.
/// Factory to create the Modbus transport; defaults to
/// , which honours .
/// Logger instance; defaults to null logger if not provided.
public ModbusDriver(ModbusDriverOptions options, string driverInstanceId,
Func? transportFactory = null,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
_options = options;
_driverInstanceId = driverInstanceId;
_logger = logger ?? NullLogger.Instance;
_resolver = new EquipmentTagRefResolver(
r => Volatile.Read(ref _tagsByRawPath).TryGetValue(r, out var t) ? t : null);
_transportFactory = transportFactory
?? (o => ModbusTransportFactory.Create(o));
_poll = new PollGroupEngine(
reader: ReadAsync,
onChange: (handle, tagRef, snapshot) =>
{
if (!ShouldPublish(tagRef, snapshot)) return;
// Buffer what we publish so the cycle-end batch carries exactly the same items.
// The append stays OUTSIDE the catch below on purpose: as far as the batch contract
// is concerned the item WAS published — only the per-tag subscriber faulted — so it
// must still appear in this cycle's batch, keeping the two events content-identical.
_cycleBatchItems.GetOrAdd(handle, _ => new List())
.Add(new DataChangeItem(tagRef, snapshot));
// Defensive, mirroring PollGroupEngine.ReportError: the engine invokes this bridge
// UNGUARDED, so a throwing external subscriber would unwind PollOnceAsync mid-loop
// and the cycle would never reach onBatch. Because the buffer above persists across
// cycles (it is drained only by onBatch), those items would then silently MERGE into
// a later cycle's batch — cross-cycle contamination unique to this driver. Swallowing
// is what keeps the cycle running so onBatch drains the buffer on schedule.
try { OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)); }
catch { /* a buggy subscriber must not corrupt the poll cycle */ }
},
onError: HandlePollError,
backoffCap: PollBackoffCap,
// The engine's own items list is deliberately IGNORED: it is collected pre-deadband
// (PollGroupEngine has no knowledge of ShouldPublish), so forwarding it would carry
// changes the per-tag event suppressed — and a consumer that prefers the batch event
// would see deadband silently stop working. The buffer holds exactly what the per-tag
// event published, keeping the two events content-identical per the IBatchSubscribable
// contract. onBatch is used only as the end-of-cycle signal. A cycle whose every change
// was suppressed leaves the buffer empty and raises nothing.
onBatch: (handle, _) =>
{
if (_cycleBatchItems.TryRemove(handle, out var published) && published.Count > 0)
OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, published));
},
// The engine owns no Meter (Core.Abstractions takes no metrics dependency), so this
// closure IS the production consumer of its overrun signal.
onOverrun: _ => OtOpcUaTelemetry.PollOverruns.Add(1,
new KeyValuePair(OtOpcUaTelemetry.PollOverrunDriverTypeTag, DriverTypeNames.Modbus),
new KeyValuePair(OtOpcUaTelemetry.PollOverrunDriverInstanceTag, driverInstanceId)));
}
/// Upper bound on the poll-loop failure backoff — adopts the S7-proven 30 s cap fleet-wide.
private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30);
///
/// Routes a poll-loop reader failure to the driver health surface: logs it and
/// degrades to preserving LastSuccessfulRead.
/// Never downgrades a state (a stronger, config-level signal).
///
/// The exception caught by the poll engine.
internal void HandlePollError(Exception ex)
{
_logger.LogWarning(ex, "Modbus poll reader failed. Driver={DriverInstanceId}", _driverInstanceId);
lock (_stateLock)
{
// Lifecycle-owned states are left alone: Faulted is a stronger, config-level signal, and an
// Initializing/Unknown driver has no transport for a poll to have failed against — the
// failure is the re-initialize window itself, which the lifecycle reports on its own.
var current = _health;
if (current.State is DriverState.Faulted or DriverState.Initializing or DriverState.Unknown) return;
_commsFault = ex.Message;
PublishComposedHealthLocked(current.LastSuccessfulRead);
}
}
// ---- health composition ----
///
/// Per-batch accounting: did the device answer anything, did a comms fault happen, and which units
/// (or the whole endpoint) must the rest of this batch fail fast against.
///
private sealed class BatchOutcome
{
public bool Answered;
public string? CommsFault;
public bool EndpointDown;
private HashSet? _failedUnits;
public void FailUnit(byte unit) => (_failedUnits ??= []).Add(unit);
public bool IsFailedFast(byte unit) => EndpointDown || (_failedUnits?.Contains(unit) ?? false);
}
///
/// Applies one batch's outcome to the driver health, under , and only when
/// is still current. A batch that touched no device (every reference
/// unknown, every write refused before the wire) changes nothing.
///
private void ReportOutcome(int generation, BatchOutcome batch, DateTime now)
{
if (!batch.Answered && batch.CommsFault is null) return;
lock (_stateLock)
{
if (generation != _generation) return;
var current = _health;
if (current.State is DriverState.Faulted or DriverState.Initializing or DriverState.Unknown) return;
_commsFault = batch.CommsFault;
PublishComposedHealthLocked(batch.Answered ? now : current.LastSuccessfulRead);
}
}
/// Writes Healthy/Degraded from the comms-fault slot. Caller holds .
private void PublishComposedHealthLocked(DateTime? lastSuccessfulRead)
{
var fault = _commsFault;
WriteHealth(fault is null
? new DriverHealth(DriverState.Healthy, lastSuccessfulRead, null)
: new DriverHealth(DriverState.Degraded, lastSuccessfulRead, fault));
}
///
/// A probe answer proves the device is reachable: advance LastSuccessfulRead without
/// touching either fault side (it says nothing about the tags).
///
private void NoteDeviceAnswered(int generation)
{
lock (_stateLock)
{
if (generation != _generation) return;
var current = _health;
if (current.State is not (DriverState.Healthy or DriverState.Degraded)) return;
WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow });
}
}
/// Lifecycle health write, dropped when has been superseded.
private void WriteLifecycleHealth(int generation, DriverHealth health)
{
lock (_stateLock)
{
if (generation != _generation) return;
_commsFault = null;
WriteHealth(health);
}
}
///
/// Classifies one failed read/write against the health contract and returns the per-tag status:
///
/// Exception 01/02/03 — a deterministic REFUSAL: the device answered; per-tag Bad, a
/// once-per-tag Warning, no Degraded (it used to latch Degraded and, if every tag refused,
/// trip the 90 s watchdog against a reachable device).
/// Exception 04/05/06 — the device answered but is not serving: Degraded, answered.
/// Exception 0A/0B (and unknown codes) — the gateway answered for a target that did not:
/// Degraded, not answered; 0B fails the rest of that unit's batch fast.
/// Undecodable data — the device answered with a value we cannot represent: per-tag
/// BadDecodingError, once-per-tag Warning, no Degraded.
/// No socket / throttled reconnect — the endpoint is down: Degraded, the rest of the
/// batch fails fast.
/// Any other transport failure (timeout, desync, EOF) — Degraded; the rest of that
/// unit's batch fails fast instead of serialising N timeouts.
///
///
private uint ClassifyFailure(Exception ex, byte unit, string tagRef, BatchOutcome batch, bool isWrite)
{
switch (ex)
{
case ModbusException mex when IsRefusalCode(mex.ExceptionCode):
batch.Answered = true;
WarnRefusalOnce(tagRef, isWrite, $"exception 0x{mex.ExceptionCode:X2} (fc 0x{mex.FunctionCode:X2})");
return MapModbusExceptionToStatus(mex.ExceptionCode);
case ModbusException mex when mex.ExceptionCode is 0x04 or 0x05 or 0x06:
batch.Answered = true;
batch.CommsFault ??= mex.Message;
return MapModbusExceptionToStatus(mex.ExceptionCode);
case ModbusException mex:
batch.CommsFault ??= mex.Message;
if (mex.ExceptionCode == 0x0B) batch.FailUnit(unit);
return MapModbusExceptionToStatus(mex.ExceptionCode);
case ModbusDecodeException dex:
batch.Answered = true;
WarnRefusalOnce(tagRef, isWrite, dex.Message);
return StatusBadDecodingError;
case ModbusTransportUnavailableException or ObjectDisposedException or System.Net.Sockets.SocketException:
batch.CommsFault ??= ex.Message;
batch.EndpointDown = true;
return StatusBadCommunicationError;
default:
batch.CommsFault ??= ex.Message;
// Fail the unit fast only when nothing came back in time (timeout / desync / socket-level IO):
// a malformed-but-delivered reply (byte-count mismatch) proves the unit responds, so its
// siblings are still tried.
if (ex is IOException or TimeoutException or OperationCanceledException) batch.FailUnit(unit);
return StatusBadCommunicationError;
}
}
/// Exception 01 Illegal Function / 02 Illegal Data Address / 03 Illegal Data Value — the device refused the request.
internal static bool IsRefusalCode(byte exceptionCode) => exceptionCode is 0x01 or 0x02 or 0x03;
private void WarnRefusalOnce(string tagRef, bool isWrite, string detail)
{
if (!Volatile.Read(ref _refusalWarned).TryAdd((isWrite ? "w:" : "r:") + tagRef, 0)) return;
_logger.LogWarning(
"Modbus device refused {Operation} of tag {RawPath}: {Detail}. The tag reads/writes Bad; driver health is unaffected (the device answered). Driver={DriverInstanceId}",
isWrite ? "a write" : "a read", tagRef, detail, _driverInstanceId);
}
//
// Resolve a reference to the per-slave host key (host:port/unitN) that keys its
// per-host resilience (bulkhead / circuit breaker). Routes through
// so an equipment-tag reference (raw TagConfig
// JSON) keys its OWN authored unit — via the tag's optional unitId or the driver
// default — rather than the single-slave fallback. The resolver caches parses, so this adds
// no per-call cost after first use.
//
///
public string ResolveHost(string fullReference)
{
if (_resolver.TryResolve(fullReference, out var tag))
return BuildSlaveHostName(ResolveUnitId(tag));
// Unknown reference — fall back to driver-instance host (single-slave behaviour).
return HostName;
}
/// Format a per-slave host string. Multi-slave deployments distinguish breakers by this string.
private string BuildSlaveHostName(byte unitId) => $"{_options.Host}:{_options.Port}/unit{unitId}";
///
/// Deadband filter for the subscribe path. The baseline is keyed on (value, status), never
/// value alone: a Bad/null publish drops the baseline, and a status change always publishes. A
/// device that comes back after a comms drop usually comes back holding the SAME reading, so the
/// old value-only comparison (|10 − 10| < deadband) swallowed the Good recovery and left
/// the node Bad until the value drifted past the deadband — the #527 trap.
///
/// The subscribed reference.
/// The snapshot the poll engine is about to raise.
/// when the snapshot must be published.
internal bool ShouldPublish(string tagRef, DataValueSnapshot snapshot)
{
if (!_resolver.TryResolve(tagRef, out var tag) || tag.Deadband is null) return true;
// Deadband only applies to numeric scalar types — array / Bool / String publishes
// unconditionally. Easier to special-case skip than to enumerate the supported types.
if (tag.ArrayCount.HasValue || tag.DataType is ModbusDataType.Bool or ModbusDataType.BitInRegister or ModbusDataType.String)
return true;
if (snapshot.Value is null || StatusCodeIsBad(snapshot.StatusCode))
{
// A Bad publish always passes AND drops the baseline, so the next Good sample is
// compared against nothing and publishes — whatever value it carries.
_lastPublishedByRef.TryRemove(tagRef, out _);
return true;
}
if (!_lastPublishedByRef.TryGetValue(tagRef, out var prev) || prev.Status != snapshot.StatusCode)
{
// First sample, or a quality transition (Good ↔ Uncertain): passes unconditionally and
// (re)seeds the comparison. The threshold cannot be evaluated across a status change.
_lastPublishedByRef[tagRef] = (snapshot.Value, snapshot.StatusCode);
return true;
}
var newD = Convert.ToDouble(snapshot.Value);
var oldD = Convert.ToDouble(prev.Value);
if (Math.Abs(newD - oldD) < tag.Deadband.Value) return false;
_lastPublishedByRef[tagRef] = (snapshot.Value, snapshot.StatusCode);
return true;
}
/// OPC UA severity bits 31..30 = 10 ⇒ Bad.
private static bool StatusCodeIsBad(uint statusCode) => (statusCode & 0xC0000000u) == 0x80000000u;
///
public string DriverInstanceId => _driverInstanceId;
///
public string DriverType => "Modbus";
/// True when the supplied DriverConfig JSON carries a real body. The bootstrapper always
/// passes a populated document; some unit tests pass "{}" or an empty string to exercise
/// lifecycle shape without a config — those keep the constructor-supplied options.
private static bool HasConfigBody(string? driverConfigJson)
{
if (string.IsNullOrWhiteSpace(driverConfigJson)) return false;
var trimmed = driverConfigJson.Trim();
return trimmed is not "{}" and not "[]";
}
///
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
/// The gate-held body of .
private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// Claim a new generation and take the previous transport out of service atomically: any read or
// write that captured the old generation can no longer write health, and nothing can capture the
// old transport from here on.
int generation;
IModbusTransport? previous;
lock (_stateLock)
{
generation = ++_generation;
previous = _transport;
_transport = null;
_commsFault = null;
WriteHealth(new DriverHealth(DriverState.Initializing, null, null));
}
IModbusTransport? transport = null;
try
{
// Release whatever a PREVIOUS initialize published before building the replacements. All
// fields assigned below (_transport, _probeCts, _reprobeCts) were simply overwritten,
// so a second initialize on a live driver stranded a connected transport plus two running
// background loops that no later dispose could reach. The IDriver contract lets a
// caller re-initialise a live driver without ShutdownAsync, and DriverInstanceActor's retry tick
// can overlap a still-running attempt, so this must not rely on a teardown having happened.
// Deliberately NOT the shared TeardownAsync(): that also disposes _poll, which tears down
// every ACTIVE SUBSCRIPTION — and nothing re-subscribes them afterwards, because a
// mid-life re-initialize is not a shutdown the OPC UA layer knows to re-drive. A caller
// that genuinely wants the subscriptions gone calls ShutdownAsync. A first initialize
// sees nulls, so this block is a no-op on the initial call.
await StopBackgroundLoopsAsync().ConfigureAwait(false);
if (previous is not null) await previous.DisposeAsync().ConfigureAwait(false);
// re-parse the supplied DriverConfig so a config change delivered through the IDriver
// contract is honoured. Without this the driver keeps serving the options it was CONSTRUCTED
// with, so an operator's edit is discarded while the deployment still seals green. An empty /
// placeholder document (the "{}" some unit tests pass) keeps the constructor-supplied options.
if (HasConfigBody(driverConfigJson))
_options = ModbusDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson);
transport = _transportFactory(_options);
await transport.ConnectAsync(cancellationToken).ConfigureAwait(false);
// Build the RawPath → definition table OFF TO THE SIDE from the authored raw tags. Each
// entry's TagConfig blob is mapped by the pure factory; the entry's WriteIdempotent flag is
// threaded onto the def (it lives on the RawTagEntry, not inside the blob). A mapper miss is
// skipped (logged), never thrown — a bad tag must not fail the whole driver init.
var table = new Dictionary(StringComparer.Ordinal);
_skippedTags.BeginRebuild(_options.RawTags.Count);
foreach (var entry in _options.RawTags)
{
if (ModbusTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, _options.Family, _options.MelsecSubFamily, out var def))
table[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent };
else
{
_skippedTags.Skip(entry.RawPath, "TagConfig did not map to a definition");
_logger.LogWarning(
"Modbus tag config did not map to a definition; skipping. Driver={DriverInstanceId} RawPath={RawPath}",
_driverInstanceId, entry.RawPath);
}
}
_skippedTags.Commit();
// Publish transport + table + Healthy as ONE generation-checked step. A superseded attempt
// (a shutdown/initialize ran while this one was abandoned mid-connect) publishes nothing and
// releases its own transport.
lock (_stateLock)
{
if (generation != _generation)
throw new OperationCanceledException(
"Modbus initialize was superseded by a later lifecycle operation; its transport was released");
_transport = transport;
Volatile.Write(ref _tagsByRawPath, table);
Volatile.Write(ref _refusalWarned, new ConcurrentDictionary(StringComparer.Ordinal));
Volatile.Write(ref _fc23Unsupported, new ConcurrentDictionary());
WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
// PR 23: kick off the probe loop once the transport is up. Initial state stays
// Unknown until the first probe tick succeeds — avoids broadcasting a premature
// Running transition before any register round-trip has happened. Started inside the
// generation-checked publish so a teardown that bypassed the lifecycle gate can never
// run between the publish and the loop start and strand a loop.
//
// #729: each loop's CancellationToken VALUE is snapshotted here, before scheduling, and the
// worker delegate closes over that value — never over the CTS (or the mutable _probeCts /
// _reprobeCts fields). The scheduling token only suppresses a still-queued task; once the
// pool has committed to running the delegate, teardown can complete underneath it (the
// bounded join in StopBackgroundLoopsAsync gives up and disposes the CTS). A delegate that
// then read cts.Token would throw ObjectDisposedException; one that read the field would
// see null or adopt the NEXT initialization's token. A token value stays valid after its
// source is disposed and still reports the cancellation, so the late worker exits cleanly
// on its own generation's cancelled token.
if (_options.Probe.Enabled)
{
var cts = new CancellationTokenSource();
var probeToken = cts.Token;
var probeTransport = transport;
_probeCts = cts;
_probeTask = Task.Run(() => ProbeLoopAsync(generation, probeTransport, probeToken), probeToken);
}
if (_options.AutoProhibitReprobeInterval is not null)
{
var cts = new CancellationTokenSource();
var reprobeToken = cts.Token;
_reprobeCts = cts;
_reprobeTask = Task.Run(() => ReprobeLoopAsync(reprobeToken), reprobeToken);
}
}
transport = null; // published — owned by the driver now
}
catch (Exception ex)
{
// A failed attempt releases the transport it built — it is never left half-published.
if (transport is not null)
{
try { await transport.DisposeAsync().ConfigureAwait(false); } catch { }
}
WriteLifecycleHealth(generation, new DriverHealth(DriverState.Faulted, null, ex.Message));
throw;
}
}
/// Cancels, bounded-joins and disposes the probe and re-probe loops. Idempotent.
///
/// Limits (#729): the join is bounded by , so this does NOT
/// guarantee a loop has finished when it returns — a loop stuck in a transport call that ignores its
/// token, or a worker the pool had committed to but not yet invoked, is abandoned, not awaited. Such
/// a worker holds its own already-cancelled token value (never the CTS or the fields), so it exits on
/// its first token check without faulting. Transport lifetime is NOT bound to the loops: the caller
/// disposes the transport after this returns, so a probe send that ignores its token and outlives the
/// join completes (or faults) against a disposed transport; the loop then sees its cancelled token and
/// returns without a health transition.
///
private async Task StopBackgroundLoopsAsync()
{
var probeCts = _probeCts;
var reprobeCts = _reprobeCts;
var tasks = new[] { _probeTask, _reprobeTask }.Where(t => t is not null).Cast().ToArray();
_probeCts = null;
_reprobeCts = null;
_probeTask = null;
_reprobeTask = null;
try { probeCts?.Cancel(); } catch { }
try { reprobeCts?.Cancel(); } catch { }
if (tasks.Length > 0)
{
try { await Task.WhenAll(tasks).WaitAsync(BackgroundLoopJoinTimeout).ConfigureAwait(false); }
catch { /* a loop that ignores its token is abandoned, never awaited forever */ }
}
probeCts?.Dispose();
reprobeCts?.Dispose();
}
///
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// Holds the gate across BOTH halves and calls the ungated cores: taking it once here is what
// makes the shutdown-then-initialize pair atomic. Routing through the public (gated) methods
// instead would both self-deadlock and leave a window between them for a concurrent
// initialize to slip into.
await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
await ShutdownCoreAsync().ConfigureAwait(false);
await InitializeCoreAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
///
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
if (await _lifecycleGate.WaitAsync(LifecycleGateTimeout, cancellationToken).ConfigureAwait(false))
{
try
{
await ShutdownCoreAsync().ConfigureAwait(false);
}
finally
{
_lifecycleGate.Release();
}
}
else
{
_logger.LogWarning(
"Modbus lifecycle gate still held at shutdown; tearing the transport down anyway. Driver={DriverInstanceId}",
_driverInstanceId);
await ShutdownCoreAsync().ConfigureAwait(false);
}
}
/// The gate-held body of , shared with
/// .
private async Task ShutdownCoreAsync()
{
int generation;
DateTime? lastRead;
lock (_stateLock)
{
// A shutdown is a lifecycle operation too: it supersedes any in-flight (abandoned) initialize,
// and its own final write below is dropped if a NEWER initialize started meanwhile.
generation = ++_generation;
lastRead = _health.LastSuccessfulRead;
}
await TeardownAsync().ConfigureAwait(false);
WriteLifecycleHealth(generation, new DriverHealth(DriverState.Unknown, lastRead, null));
}
///
public DriverHealth GetHealth() => ReadHealth();
///
/// Barrier-protected read of the multi-thread _health field.
/// Volatile.Read guarantees GetHealth() and the in-driver self-reads (the
/// Degraded paths that retain LastSuccessfulRead) observe the most recently
/// published snapshot rather than a per-core cached stale copy.
///
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
///
/// Barrier-protected publish of a new _health snapshot.
///
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
// ---- ITagDiscovery ----
///
public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;
///
public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(builder);
var folder = builder.Folder("Modbus", "Modbus");
// Surface the authored raw tags (mapped into the RawPath → def table at Initialize). The def's
// Name is the RawPath, used for both the browse name and the driver FullName.
foreach (var t in _tagsByRawPath.Values)
{
folder.Variable(t.Name, t.Name, new DriverAttributeInfo(
FullName: t.Name,
DriverDataType: MapDataType(t.DataType),
IsArray: t.ArrayCount.HasValue,
ArrayDim: t.ArrayCount.HasValue ? (uint)t.ArrayCount.Value : null,
SecurityClass: t.Writable ? SecurityClassification.Operate : SecurityClassification.ViewOnly,
IsHistorized: false,
IsAlarm: false,
WriteIdempotent: t.WriteIdempotent));
}
return Task.CompletedTask;
}
// ---- IReadable ----
///
public async Task> ReadAsync(
IReadOnlyList fullReferences, CancellationToken cancellationToken)
{
var (generation, transport, table) = CaptureLifecycle();
var now = DateTime.UtcNow;
var results = new DataValueSnapshot[fullReferences.Count];
var batch = new BatchOutcome();
// MaxReadGap 0 = coalescing OFF (its pre-lane meaning, kept for every persisted config that authored
// it); the gap-8 default applies only to an absent key.
var coalesced = !_options.DisableReadCoalescing && _options.MaxReadGap > 0
? await ReadCoalescedAsync(transport, table, fullReferences, results, now, batch, cancellationToken).ConfigureAwait(false)
: new HashSet();
for (var i = 0; i < fullReferences.Count; i++)
{
if (coalesced.Contains(i)) continue;
if (!table.TryGetValue(fullReferences[i], out var tag))
{
results[i] = new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now);
continue;
}
var unit = ResolveUnitId(tag);
if (batch.IsFailedFast(unit))
{
// Fail fast: this unit (or the whole endpoint) already failed at the transport this batch;
// paying the same timeout once per remaining tag would stall the poll N-fold.
results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
InvalidateWriteCache(fullReferences[i]);
continue;
}
try
{
var value = await ReadOneAsync(transport, tag, cancellationToken).ConfigureAwait(false);
results[i] = new DataValueSnapshot(value, 0u, now, now);
batch.Answered = true;
// Invalidate the WriteOnChangeOnly cache when the read returns a different value
// — typically an HMI-side or PLC-internal change. Without this, a setpoint
// tweaked at the panel could be silently re-suppressed when our client tried
// to restore it.
InvalidateWriteCacheIfDiverged(fullReferences[i], value);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception ex)
{
results[i] = new DataValueSnapshot(null, ClassifyFailure(ex, unit, fullReferences[i], batch, isWrite: false), null, now);
InvalidateWriteCache(fullReferences[i]);
}
}
// ONE health write per batch, worst-wins over the batch, generation-checked. It used to be
// last-writer-wins per tag, so health flapped with tag order.
ReportOutcome(generation, batch, now);
return results;
}
///
/// Captures the generation, transport and tag table a read/write batch runs against, atomically
/// with respect to a lifecycle publish.
///
private (int Generation, IModbusTransport Transport, Dictionary Table) CaptureLifecycle()
{
lock (_stateLock)
{
var transport = _transport ?? throw new InvalidOperationException("ModbusDriver not initialized");
return (_generation, transport, _tagsByRawPath);
}
}
/// Runs a codec decode, normalising its failure to so the
/// classifier can tell undecodable device data from a transport fault.
private static object Decode(Func