2045 lines
103 KiB
C#
2045 lines
103 KiB
C#
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
using ZB.MOM.WW.OtOpcUa.Driver.AbCip.Import;
|
||
using ZB.MOM.WW.OtOpcUa.Driver.AbCip.PlcFamilies;
|
||
|
||
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
|
||
|
||
/// <summary>
|
||
/// Allen-Bradley CIP / EtherNet-IP driver for ControlLogix / CompactLogix / Micro800 /
|
||
/// GuardLogix families. Implements <see cref="IDriver"/> only for now — read/write/
|
||
/// subscribe/discover capabilities ship in subsequent PRs (3–8) and family-specific quirk
|
||
/// profiles ship in PRs 9–12.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>Wire layer is libplctag 1.6.x (plan decision #11). Per-device host addresses use
|
||
/// the <c>ab://gateway[:port]/cip-path</c> canonical form parsed via
|
||
/// <see cref="AbCipHostAddress.TryParse"/>; those strings become the <c>hostName</c> key
|
||
/// for Polly bulkhead + circuit-breaker isolation per plan decision #144.</para>
|
||
///
|
||
/// <para>Tier A per plan decisions #143–145 — in-process, shares server lifetime, no
|
||
/// sidecar. <see cref="ReinitializeAsync"/> is the Tier-B escape hatch for recovering
|
||
/// from native-heap growth that the CLR allocator can't see; it tears down every
|
||
/// <see cref="PlcTagHandle"/> and reconnects each device.</para>
|
||
/// </remarks>
|
||
public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable,
|
||
IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, IDriverControl, IDisposable, IAsyncDisposable
|
||
{
|
||
private readonly AbCipDriverOptions _options;
|
||
private readonly string _driverInstanceId;
|
||
private readonly IAbCipTagFactory _tagFactory;
|
||
private readonly IAbCipTagEnumeratorFactory _enumeratorFactory;
|
||
private readonly IAbCipTemplateReaderFactory _templateReaderFactory;
|
||
private readonly AbCipTemplateCache _templateCache = new();
|
||
private readonly PollGroupEngine _poll;
|
||
private readonly Dictionary<string, DeviceState> _devices = new(StringComparer.OrdinalIgnoreCase);
|
||
private readonly Dictionary<string, AbCipTagDefinition> _tagsByName = new(StringComparer.OrdinalIgnoreCase);
|
||
private readonly AbCipAlarmProjection _alarmProjection;
|
||
private readonly SemaphoreSlim _discoverySemaphore = new(1, 1);
|
||
private readonly AbCipWriteCoalescer _writeCoalescer = new();
|
||
private readonly AbCipSystemTagSource _systemTagSource = new();
|
||
private DriverHealth _health = new(DriverState.Unknown, null, null);
|
||
|
||
public event EventHandler<DataChangeEventArgs>? OnDataChange;
|
||
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
|
||
public event EventHandler<AlarmEventArgs>? OnAlarmEvent;
|
||
|
||
/// <summary>Internal seam for the alarm projection to raise events through the driver.</summary>
|
||
internal void InvokeAlarmEvent(AlarmEventArgs args) => OnAlarmEvent?.Invoke(this, args);
|
||
|
||
public AbCipDriver(AbCipDriverOptions options, string driverInstanceId,
|
||
IAbCipTagFactory? tagFactory = null,
|
||
IAbCipTagEnumeratorFactory? enumeratorFactory = null,
|
||
IAbCipTemplateReaderFactory? templateReaderFactory = null)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(options);
|
||
_options = options;
|
||
_driverInstanceId = driverInstanceId;
|
||
_tagFactory = tagFactory ?? new LibplctagTagFactory();
|
||
_enumeratorFactory = enumeratorFactory ?? new LibplctagTagEnumeratorFactory();
|
||
_templateReaderFactory = templateReaderFactory ?? new LibplctagTemplateReaderFactory();
|
||
_poll = new PollGroupEngine(
|
||
reader: ReadAsync,
|
||
onChange: (handle, tagRef, snapshot) =>
|
||
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)));
|
||
_alarmProjection = new AbCipAlarmProjection(this, _options.AlarmPollInterval);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Fetch + cache the shape of a Logix UDT by template instance id. First call reads
|
||
/// the Template Object off the controller; subsequent calls for the same
|
||
/// <c>(deviceHostAddress, templateInstanceId)</c> return the cached shape without
|
||
/// additional network traffic. <c>null</c> on template-not-found / decode failure so
|
||
/// callers can fall back to declaration-driven UDT fan-out.
|
||
/// </summary>
|
||
internal async Task<AbCipUdtShape?> FetchUdtShapeAsync(
|
||
string deviceHostAddress, uint templateInstanceId, CancellationToken cancellationToken)
|
||
{
|
||
var cached = _templateCache.TryGet(deviceHostAddress, templateInstanceId);
|
||
if (cached is not null) return cached;
|
||
|
||
if (!_devices.TryGetValue(deviceHostAddress, out var device)) return null;
|
||
|
||
// UDT template reads target the @udt/{id} pseudo-tag, which the controller already
|
||
// serves via a logical-segment path of its own. Force Symbolic addressing so we don't
|
||
// overlay the driver's Logical mode on top — libplctag knows how to dereference the
|
||
// pseudo-tag directly.
|
||
var deviceParams = new AbCipTagCreateParams(
|
||
Gateway: device.ParsedAddress.Gateway,
|
||
Port: device.ParsedAddress.Port,
|
||
CipPath: device.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
|
||
TagName: $"@udt/{templateInstanceId}",
|
||
Timeout: _options.Timeout,
|
||
ConnectionSize: device.ConnectionSize,
|
||
AddressingMode: AddressingMode.Symbolic);
|
||
|
||
try
|
||
{
|
||
using var reader = _templateReaderFactory.Create();
|
||
var buffer = await reader.ReadAsync(deviceParams, templateInstanceId, cancellationToken).ConfigureAwait(false);
|
||
var shape = CipTemplateObjectDecoder.Decode(buffer);
|
||
if (shape is not null)
|
||
_templateCache.Put(deviceHostAddress, templateInstanceId, shape);
|
||
return shape;
|
||
}
|
||
catch (OperationCanceledException) { throw; }
|
||
catch
|
||
{
|
||
// Template read failure — log via the driver's health surface so operators see it,
|
||
// but don't propagate since callers should fall back to declaration-driven UDT
|
||
// semantics rather than failing the whole discovery run.
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/// <summary>Shared UDT template cache. Exposed for PR 6 (UDT reader) + diagnostics.</summary>
|
||
internal AbCipTemplateCache TemplateCache => _templateCache;
|
||
|
||
public string DriverInstanceId => _driverInstanceId;
|
||
public string DriverType => "AbCip";
|
||
|
||
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||
{
|
||
_health = new DriverHealth(DriverState.Initializing, null, null);
|
||
try
|
||
{
|
||
foreach (var device in _options.Devices)
|
||
{
|
||
var addr = AbCipHostAddress.TryParse(device.HostAddress)
|
||
?? throw new InvalidOperationException(
|
||
$"AbCip device has invalid HostAddress '{device.HostAddress}' — expected 'ab://gateway[:port]/cip-path'.");
|
||
var profile = AbCipPlcFamilyProfile.ForFamily(device.PlcFamily);
|
||
// PR abcip-3.1 — validate the optional ConnectionSize override before stamping
|
||
// the device. The Kepware-supported range [500..4002] is what the libplctag
|
||
// ControlLogix driver supports; anything outside that fails the Forward Open at
|
||
// runtime, so we reject it loudly at config time instead.
|
||
if (device.ConnectionSize is int explicitSize)
|
||
{
|
||
if (explicitSize < AbCipConnectionSize.Min || explicitSize > AbCipConnectionSize.Max)
|
||
throw new InvalidOperationException(
|
||
$"AbCip device '{device.HostAddress}' has ConnectionSize {explicitSize} outside the supported range " +
|
||
$"[{AbCipConnectionSize.Min}..{AbCipConnectionSize.Max}].");
|
||
// Legacy-firmware warning: families whose profile default is 504 (CompactLogix
|
||
// narrow cap, also where v19-and-earlier ControlLogix lives) can't actually
|
||
// raise their CIP buffer above 511 bytes — the controller rejects the Forward
|
||
// Open. Ship the override anyway so newer firmware can use it, but flag the
|
||
// mismatch so operators see it in the warning sink.
|
||
if (explicitSize > AbCipConnectionSize.LegacyFirmwareCap
|
||
&& profile.DefaultConnectionSize <= AbCipConnectionSize.LegacyFirmwareCap)
|
||
{
|
||
_options.OnWarning?.Invoke(
|
||
$"AbCip device '{device.HostAddress}' family '{device.PlcFamily}' uses a narrow-buffer profile " +
|
||
$"(default ConnectionSize {profile.DefaultConnectionSize}); the configured ConnectionSize {explicitSize} " +
|
||
$"exceeds the {AbCipConnectionSize.LegacyFirmwareCap}-byte legacy-firmware cap and will fail the " +
|
||
"Forward Open on v19-and-earlier ControlLogix or 5069-L1/L2/L3 CompactLogix firmware.");
|
||
}
|
||
}
|
||
// PR abcip-3.2 — resolve AddressingMode at the device level. Auto → Symbolic
|
||
// until a future PR adds a real auto-detection heuristic; Logical against an
|
||
// unsupported family falls back to Symbolic + emits a warning so misconfiguration
|
||
// does not fault the driver.
|
||
var resolvedAddressing = ResolveAddressingMode(device, profile);
|
||
// PR abcip-3.3 — resolve ReadStrategy at the device level. User-forced MultiPacket
|
||
// against a non-packing family (Micro800 et al) falls back to WholeUdt with a
|
||
// warning. Auto stays as-is — the planner re-evaluates per-batch using the
|
||
// device's MultiPacketSparsityThreshold.
|
||
var resolvedReadStrategy = ResolveReadStrategy(device, profile);
|
||
_devices[device.HostAddress] = new DeviceState(
|
||
addr, device, profile, resolvedAddressing, resolvedReadStrategy);
|
||
}
|
||
// Pre-declared tags first; L5K imports fill in only the names not already covered
|
||
// (operators can override an imported entry by re-declaring it under Tags).
|
||
var declaredNames = new HashSet<string>(
|
||
_options.Tags.Select(t => t.Name),
|
||
StringComparer.OrdinalIgnoreCase);
|
||
var allTags = new List<AbCipTagDefinition>(_options.Tags);
|
||
foreach (var import in _options.L5kImports)
|
||
{
|
||
MergeImport(
|
||
deviceHost: import.DeviceHostAddress,
|
||
filePath: import.FilePath,
|
||
inlineText: import.InlineText,
|
||
namePrefix: import.NamePrefix,
|
||
parse: L5kParser.Parse,
|
||
formatLabel: "L5K",
|
||
declaredNames: declaredNames,
|
||
allTags: allTags);
|
||
}
|
||
foreach (var import in _options.L5xImports)
|
||
{
|
||
MergeImport(
|
||
deviceHost: import.DeviceHostAddress,
|
||
filePath: import.FilePath,
|
||
inlineText: import.InlineText,
|
||
namePrefix: import.NamePrefix,
|
||
parse: L5xParser.Parse,
|
||
formatLabel: "L5X",
|
||
declaredNames: declaredNames,
|
||
allTags: allTags);
|
||
}
|
||
foreach (var import in _options.CsvImports)
|
||
{
|
||
MergeCsvImport(import, declaredNames, allTags);
|
||
}
|
||
|
||
foreach (var tag in allTags)
|
||
{
|
||
_tagsByName[tag.Name] = tag;
|
||
if (tag.DataType == AbCipDataType.Structure && tag.Members is { Count: > 0 })
|
||
{
|
||
foreach (var member in tag.Members)
|
||
{
|
||
var memberTag = new AbCipTagDefinition(
|
||
Name: $"{tag.Name}.{member.Name}",
|
||
DeviceHostAddress: tag.DeviceHostAddress,
|
||
TagPath: $"{tag.TagPath}.{member.Name}",
|
||
DataType: member.DataType,
|
||
Writable: member.Writable,
|
||
WriteIdempotent: member.WriteIdempotent,
|
||
StringLength: member.StringLength,
|
||
// PR abcip-4.1 — inherit per-tag scan rate from parent so a UDT
|
||
// declared at 100 ms publishes every member at 100 ms without the
|
||
// operator having to repeat ScanRateMs on every member.
|
||
ScanRateMs: tag.ScanRateMs);
|
||
_tagsByName[memberTag.Name] = memberTag;
|
||
}
|
||
}
|
||
}
|
||
|
||
// PR abcip-4.3 — seed each device's system-tag snapshot before the probe / read loops
|
||
// start so an immediate _System read returns a stable shape (Unknown / 0 / "") instead
|
||
// of "no snapshot recorded yet". TransitionDeviceState + ReadAsync refresh from here.
|
||
foreach (var state in _devices.Values)
|
||
RefreshSystemTagSnapshot(state, lastScanTimeMs: 0.0);
|
||
|
||
// Probe loops — one per device when enabled + a ProbeTagPath is configured.
|
||
if (_options.Probe.Enabled && !string.IsNullOrWhiteSpace(_options.Probe.ProbeTagPath))
|
||
{
|
||
foreach (var state in _devices.Values)
|
||
{
|
||
state.ProbeCts = new CancellationTokenSource();
|
||
var ct = state.ProbeCts.Token;
|
||
_ = Task.Run(() => ProbeLoopAsync(state, ct), ct);
|
||
}
|
||
}
|
||
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_health = new DriverHealth(DriverState.Faulted, null, ex.Message);
|
||
throw;
|
||
}
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.2 — resolve <see cref="AbCipDeviceOptions.AddressingMode"/> against the
|
||
/// family profile. <see cref="AddressingMode.Auto"/> resolves to <see cref="AddressingMode.Symbolic"/>
|
||
/// today (the same behaviour every previous build had); a future PR will plumb a real
|
||
/// auto-detection heuristic and document it in <c>docs/drivers/AbCip-Performance.md</c>.
|
||
/// <see cref="AddressingMode.Logical"/> against a family whose profile sets
|
||
/// <see cref="AbCipPlcFamilyProfile.SupportsLogicalAddressing"/> = <c>false</c> (Micro800,
|
||
/// SLC500, PLC5) falls back to <see cref="AddressingMode.Symbolic"/> with a warning so
|
||
/// the operator sees the misconfiguration in the log without the driver faulting.
|
||
/// </summary>
|
||
private AddressingMode ResolveAddressingMode(AbCipDeviceOptions device, AbCipPlcFamilyProfile profile)
|
||
{
|
||
switch (device.AddressingMode)
|
||
{
|
||
case AddressingMode.Logical:
|
||
if (!profile.SupportsLogicalAddressing)
|
||
{
|
||
_options.OnWarning?.Invoke(
|
||
$"AbCip device '{device.HostAddress}' family '{device.PlcFamily}' does not support " +
|
||
"Logical (instance-ID) addressing — its CIP firmware lacks Symbol Object class 0x6B. " +
|
||
"Falling back to Symbolic addressing for this device.");
|
||
return AddressingMode.Symbolic;
|
||
}
|
||
return AddressingMode.Logical;
|
||
case AddressingMode.Symbolic:
|
||
return AddressingMode.Symbolic;
|
||
case AddressingMode.Auto:
|
||
default:
|
||
// Future heuristic point — for now Auto = Symbolic so the addressing toggle is
|
||
// explicit + every existing deployment keeps the historical wire behaviour.
|
||
return AddressingMode.Symbolic;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.3 — resolve <see cref="AbCipDeviceOptions.ReadStrategy"/> against the
|
||
/// family profile. <see cref="ReadStrategy.MultiPacket"/> against a family whose profile
|
||
/// sets <see cref="AbCipPlcFamilyProfile.SupportsRequestPacking"/> = <c>false</c>
|
||
/// (Micro800 today; SLC500 / PLC5 when those profiles ship) falls back to
|
||
/// <see cref="ReadStrategy.WholeUdt"/> with a warning so the operator sees the
|
||
/// misconfiguration in the log without the driver faulting. <see cref="ReadStrategy.Auto"/>
|
||
/// stays as-is — the planner re-evaluates the choice per-batch from the device's
|
||
/// <see cref="AbCipDeviceOptions.MultiPacketSparsityThreshold"/>; we still need the
|
||
/// resolution step here so a future PR can cap Auto at WholeUdt on non-packing families
|
||
/// in one place rather than scattering the check across the read path.
|
||
/// </summary>
|
||
private ReadStrategy ResolveReadStrategy(AbCipDeviceOptions device, AbCipPlcFamilyProfile profile)
|
||
{
|
||
switch (device.ReadStrategy)
|
||
{
|
||
case ReadStrategy.MultiPacket:
|
||
if (!profile.SupportsRequestPacking)
|
||
{
|
||
_options.OnWarning?.Invoke(
|
||
$"AbCip device '{device.HostAddress}' family '{device.PlcFamily}' does not support " +
|
||
"Multi-Service Packet request packing — its CIP firmware lacks the 0x0A service. " +
|
||
"Falling back to WholeUdt read strategy for this device.");
|
||
return ReadStrategy.WholeUdt;
|
||
}
|
||
return ReadStrategy.MultiPacket;
|
||
case ReadStrategy.WholeUdt:
|
||
return ReadStrategy.WholeUdt;
|
||
case ReadStrategy.Auto:
|
||
default:
|
||
// Auto on a non-packing family stays Auto here so the planner's per-batch
|
||
// heuristic still runs; the heuristic itself never picks MultiPacket against a
|
||
// device whose AddressingMode-style guard tripped — but for ReadStrategy the guard
|
||
// lives in the device-init resolution above (user-forced MultiPacket → WholeUdt).
|
||
// For Auto, the planner consults device.Profile.SupportsRequestPacking before
|
||
// emitting MultiPacket so non-packing families always read WholeUdt under Auto.
|
||
return ReadStrategy.Auto;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Shared L5K / L5X import path — keeps source-format selection (parser delegate) the
|
||
/// only behavioural axis between the two formats. Adds the parser's tags to
|
||
/// <paramref name="allTags"/> while skipping any name already covered by an earlier
|
||
/// declaration or import (declared > L5K > L5X precedence falls out from call order).
|
||
/// </summary>
|
||
private static void MergeImport(
|
||
string deviceHost,
|
||
string? filePath,
|
||
string? inlineText,
|
||
string namePrefix,
|
||
Func<IL5kSource, L5kDocument> parse,
|
||
string formatLabel,
|
||
HashSet<string> declaredNames,
|
||
List<AbCipTagDefinition> allTags)
|
||
{
|
||
if (string.IsNullOrWhiteSpace(deviceHost))
|
||
throw new InvalidOperationException(
|
||
$"AbCip {formatLabel} import is missing DeviceHostAddress — every imported tag needs a target device.");
|
||
IL5kSource? src = null;
|
||
if (!string.IsNullOrEmpty(filePath))
|
||
src = new FileL5kSource(filePath);
|
||
else if (!string.IsNullOrEmpty(inlineText))
|
||
src = new StringL5kSource(inlineText);
|
||
if (src is null) return;
|
||
|
||
var doc = parse(src);
|
||
var ingest = new L5kIngest
|
||
{
|
||
DefaultDeviceHostAddress = deviceHost,
|
||
NamePrefix = namePrefix,
|
||
};
|
||
var result = ingest.Ingest(doc);
|
||
foreach (var importedTag in result.Tags)
|
||
{
|
||
if (declaredNames.Contains(importedTag.Name)) continue;
|
||
allTags.Add(importedTag);
|
||
declaredNames.Add(importedTag.Name);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// CSV-import variant of <see cref="MergeImport"/>. The CSV path produces
|
||
/// <see cref="AbCipTagDefinition"/> records directly (no intermediate document) so we
|
||
/// can't share the L5K/L5X parser-delegate signature. Merge semantics are identical:
|
||
/// a name already covered by a declaration or an earlier import is left untouched so
|
||
/// the precedence chain (declared > L5K > L5X > CSV) holds.
|
||
/// </summary>
|
||
private static void MergeCsvImport(
|
||
AbCipCsvImportOptions import,
|
||
HashSet<string> declaredNames,
|
||
List<AbCipTagDefinition> allTags)
|
||
{
|
||
if (string.IsNullOrWhiteSpace(import.DeviceHostAddress))
|
||
throw new InvalidOperationException(
|
||
"AbCip CSV import is missing DeviceHostAddress — every imported tag needs a target device.");
|
||
|
||
string? csvText = null;
|
||
if (!string.IsNullOrEmpty(import.FilePath))
|
||
csvText = System.IO.File.ReadAllText(import.FilePath);
|
||
else if (!string.IsNullOrEmpty(import.InlineText))
|
||
csvText = import.InlineText;
|
||
if (csvText is null) return;
|
||
|
||
var importer = new CsvTagImporter
|
||
{
|
||
DefaultDeviceHostAddress = import.DeviceHostAddress,
|
||
NamePrefix = import.NamePrefix,
|
||
};
|
||
var result = importer.Import(csvText);
|
||
foreach (var tag in result.Tags)
|
||
{
|
||
if (declaredNames.Contains(tag.Name)) continue;
|
||
allTags.Add(tag);
|
||
declaredNames.Add(tag.Name);
|
||
}
|
||
}
|
||
|
||
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
|
||
{
|
||
await ShutdownAsync(cancellationToken).ConfigureAwait(false);
|
||
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
|
||
}
|
||
|
||
public async Task ShutdownAsync(CancellationToken cancellationToken)
|
||
{
|
||
await _alarmProjection.DisposeAsync().ConfigureAwait(false);
|
||
await _poll.DisposeAsync().ConfigureAwait(false);
|
||
foreach (var state in _devices.Values)
|
||
{
|
||
try { state.ProbeCts?.Cancel(); } catch { }
|
||
state.ProbeCts?.Dispose();
|
||
state.ProbeCts = null;
|
||
state.DisposeHandles();
|
||
}
|
||
_devices.Clear();
|
||
_tagsByName.Clear();
|
||
// PR abcip-4.2 — wipe the write-coalescer cache on shutdown. Reinitializing the driver
|
||
// (Tier-B remediation) starts from a clean slate so the first write after restart pays
|
||
// the full round-trip rather than reusing stale cached state.
|
||
_writeCoalescer.ResetAll();
|
||
_health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null);
|
||
}
|
||
|
||
// ---- ISubscribable (polling overlay via shared engine) ----
|
||
|
||
/// <summary>Per-bucket subscription handles owned by one composite <see cref="AbCipCompositeSubscriptionHandle"/>.</summary>
|
||
private readonly Dictionary<long, IReadOnlyList<ISubscriptionHandle>> _compositeSubscriptions = new();
|
||
private long _nextCompositeId;
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.1 — partitions <paramref name="fullReferences"/> by the resolved publishing
|
||
/// interval (per-tag <see cref="AbCipTagDefinition.ScanRateMs"/> override falling back
|
||
/// to <paramref name="publishingInterval"/>) and registers one
|
||
/// <see cref="PollGroupEngine"/> subscription per distinct interval. The returned handle
|
||
/// wraps every per-bucket subscription so <see cref="UnsubscribeAsync"/> tears them all
|
||
/// down together — callers see one logical subscription, the engine sees N independent
|
||
/// poll loops at their own cadence.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// Approach B from the PR plan — keeps <see cref="PollGroupEngine"/> unchanged and
|
||
/// handles the multi-rate split entirely at the driver level. The engine already floors
|
||
/// each call's interval at 100 ms, so a misconfigured <c>ScanRateMs < 100</c> is
|
||
/// clamped per-bucket without driver-side validation. Tags whose <c>ScanRateMs</c>
|
||
/// equals the subscription default (or that have no override) collapse into the
|
||
/// default-rate bucket — the legacy single-rate path is preserved for callers that
|
||
/// don't set <c>ScanRateMs</c> on any tag.
|
||
/// </remarks>
|
||
public Task<ISubscriptionHandle> SubscribeAsync(
|
||
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(fullReferences);
|
||
|
||
// Bucket tags by resolved interval. Unknown tags (not in _tagsByName, e.g. typo)
|
||
// and tags with no ScanRateMs fall back to the subscription default — matches the
|
||
// S7 driver's "config typo degrades, doesn't break" stance.
|
||
var buckets = new Dictionary<TimeSpan, List<string>>();
|
||
foreach (var tagRef in fullReferences)
|
||
{
|
||
var interval = ResolveTagInterval(tagRef, publishingInterval);
|
||
if (!buckets.TryGetValue(interval, out var list))
|
||
{
|
||
list = [];
|
||
buckets[interval] = list;
|
||
}
|
||
list.Add(tagRef);
|
||
}
|
||
|
||
var innerHandles = new List<ISubscriptionHandle>(buckets.Count);
|
||
foreach (var (interval, refs) in buckets)
|
||
{
|
||
innerHandles.Add(_poll.Subscribe(refs, interval));
|
||
}
|
||
|
||
var compositeId = Interlocked.Increment(ref _nextCompositeId);
|
||
lock (_compositeSubscriptions)
|
||
_compositeSubscriptions[compositeId] = innerHandles;
|
||
return Task.FromResult<ISubscriptionHandle>(new AbCipCompositeSubscriptionHandle(compositeId, innerHandles.Count));
|
||
}
|
||
|
||
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
|
||
{
|
||
if (handle is AbCipCompositeSubscriptionHandle composite)
|
||
{
|
||
IReadOnlyList<ISubscriptionHandle>? inner;
|
||
lock (_compositeSubscriptions)
|
||
{
|
||
_compositeSubscriptions.TryGetValue(composite.Id, out inner);
|
||
_compositeSubscriptions.Remove(composite.Id);
|
||
}
|
||
if (inner is not null)
|
||
{
|
||
foreach (var h in inner)
|
||
_poll.Unsubscribe(h);
|
||
}
|
||
}
|
||
else
|
||
{
|
||
// Defensive — older callers (or tests stubbing in a raw PollGroupEngine handle)
|
||
// can still unsubscribe directly through the engine.
|
||
_poll.Unsubscribe(handle);
|
||
}
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Resolve the publishing interval for one tag — per-tag <see cref="AbCipTagDefinition.ScanRateMs"/>
|
||
/// wins, otherwise fall back to the subscription default. The engine's 100 ms floor still
|
||
/// applies at <see cref="PollGroupEngine.Subscribe"/> time so this method does NOT clamp.
|
||
/// A negative or zero <c>ScanRateMs</c> is treated as null (use default) — mis-typed
|
||
/// overrides degrade rather than fault.
|
||
/// </summary>
|
||
internal TimeSpan ResolveTagInterval(string tagRef, TimeSpan defaultInterval)
|
||
{
|
||
if (_tagsByName.TryGetValue(tagRef, out var def) &&
|
||
def.ScanRateMs is { } ms && ms > 0)
|
||
{
|
||
return TimeSpan.FromMilliseconds(ms);
|
||
}
|
||
return defaultInterval;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Test-only: count of distinct poll-engine subscriptions a composite handle owns.
|
||
/// Used by <c>AbCipPerTagScanRateTests</c> to assert that 2 tags at 2 rates produce
|
||
/// 2 buckets (and 2 tags at 1 rate produce 1 bucket).
|
||
/// </summary>
|
||
internal int GetSubscriptionBucketCount(ISubscriptionHandle handle) =>
|
||
handle is AbCipCompositeSubscriptionHandle composite ? composite.BucketCount : 0;
|
||
|
||
/// <summary>
|
||
/// Composite handle returned by <see cref="SubscribeAsync"/>. Wraps one or more
|
||
/// <see cref="PollGroupEngine"/> handles so the driver can fan out multi-rate
|
||
/// subscriptions while presenting a single token to OPC UA-side callers.
|
||
/// </summary>
|
||
internal sealed record AbCipCompositeSubscriptionHandle(long Id, int BucketCount) : ISubscriptionHandle
|
||
{
|
||
public string DiagnosticId => $"abcip-sub-{Id}({BucketCount}b)";
|
||
}
|
||
|
||
// ---- IAlarmSource (ALMD projection, #177) ----
|
||
|
||
/// <summary>
|
||
/// Subscribe to ALMD alarm transitions on <paramref name="sourceNodeIds"/>. Each id
|
||
/// names a declared ALMD UDT tag; the projection polls the tag's <c>InFaulted</c> +
|
||
/// <c>Severity</c> members at <see cref="AbCipDriverOptions.AlarmPollInterval"/> and
|
||
/// fires <see cref="OnAlarmEvent"/> on 0→1 (raise) + 1→0 (clear) transitions.
|
||
/// Feature-gated — when <see cref="AbCipDriverOptions.EnableAlarmProjection"/> is
|
||
/// <c>false</c> (the default), returns a handle wrapping a no-op subscription so
|
||
/// capability negotiation still works; <see cref="OnAlarmEvent"/> never fires.
|
||
/// </summary>
|
||
public Task<IAlarmSubscriptionHandle> SubscribeAlarmsAsync(
|
||
IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
|
||
{
|
||
if (!_options.EnableAlarmProjection)
|
||
{
|
||
var disabled = new AbCipAlarmSubscriptionHandle(0);
|
||
return Task.FromResult<IAlarmSubscriptionHandle>(disabled);
|
||
}
|
||
return _alarmProjection.SubscribeAsync(sourceNodeIds, cancellationToken);
|
||
}
|
||
|
||
public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) =>
|
||
_options.EnableAlarmProjection
|
||
? _alarmProjection.UnsubscribeAsync(handle, cancellationToken)
|
||
: Task.CompletedTask;
|
||
|
||
public Task AcknowledgeAsync(
|
||
IReadOnlyList<AlarmAcknowledgeRequest> acknowledgements, CancellationToken cancellationToken) =>
|
||
_options.EnableAlarmProjection
|
||
? _alarmProjection.AcknowledgeAsync(acknowledgements, cancellationToken)
|
||
: Task.CompletedTask;
|
||
|
||
// ---- IHostConnectivityProbe ----
|
||
|
||
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses() =>
|
||
[.. _devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))];
|
||
|
||
private async Task ProbeLoopAsync(DeviceState state, CancellationToken ct)
|
||
{
|
||
// Probe handles always run in Symbolic mode regardless of the device's resolved
|
||
// AddressingMode — the probe tag (e.g. @raw_cpu_type) is a system pseudo-tag, not a
|
||
// user symbol that appears in the @tags walk, so there is no instance ID to feed
|
||
// libplctag. Hard-coding Symbolic here keeps the probe loop independent of the symbol
|
||
// walk + matches the legacy behaviour even on Logical-mode devices.
|
||
var probeParams = new AbCipTagCreateParams(
|
||
Gateway: state.ParsedAddress.Gateway,
|
||
Port: state.ParsedAddress.Port,
|
||
CipPath: state.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: state.Profile.LibplctagPlcAttribute,
|
||
TagName: _options.Probe.ProbeTagPath!,
|
||
Timeout: _options.Probe.Timeout,
|
||
ConnectionSize: state.ConnectionSize,
|
||
AddressingMode: AddressingMode.Symbolic);
|
||
|
||
IAbCipTagRuntime? probeRuntime = null;
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
var success = false;
|
||
try
|
||
{
|
||
probeRuntime ??= _tagFactory.Create(probeParams);
|
||
// Lazy-init on first attempt; re-init after a transport failure has caused the
|
||
// native handle to be destroyed.
|
||
if (!state.ProbeInitialized)
|
||
{
|
||
await probeRuntime.InitializeAsync(ct).ConfigureAwait(false);
|
||
state.ProbeInitialized = true;
|
||
}
|
||
await probeRuntime.ReadAsync(ct).ConfigureAwait(false);
|
||
success = probeRuntime.GetStatus() == 0;
|
||
}
|
||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||
{
|
||
break;
|
||
}
|
||
catch
|
||
{
|
||
// Wire / init error — tear down the probe runtime so the next tick re-creates it.
|
||
try { probeRuntime?.Dispose(); } catch { }
|
||
probeRuntime = null;
|
||
state.ProbeInitialized = false;
|
||
}
|
||
|
||
TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped);
|
||
|
||
try { await Task.Delay(_options.Probe.Interval, ct).ConfigureAwait(false); }
|
||
catch (OperationCanceledException) { break; }
|
||
}
|
||
|
||
try { probeRuntime?.Dispose(); } catch { }
|
||
}
|
||
|
||
private void TransitionDeviceState(DeviceState state, HostState newState)
|
||
{
|
||
HostState old;
|
||
lock (state.ProbeLock)
|
||
{
|
||
old = state.HostState;
|
||
if (old == newState) return;
|
||
state.HostState = newState;
|
||
state.HostStateChangedUtc = DateTime.UtcNow;
|
||
}
|
||
// PR abcip-4.2 — drop the per-device write-coalescer cache when we lose the wire. The
|
||
// PLC may have been restarted while we were offline + our cached "we already wrote 42"
|
||
// is no longer valid PLC state. Reset on the Stopped transition (and again on the
|
||
// recovery edge for safety) so the first post-reconnect write of any value pays the
|
||
// full round-trip + the coalescer rebuilds its cache from the new baseline.
|
||
if (newState == HostState.Stopped || newState == HostState.Running)
|
||
_writeCoalescer.Reset(state.Options.HostAddress);
|
||
// PR abcip-4.3 — refresh the diagnostic-tag snapshot on every transition so a client
|
||
// subscribed to _System/_ConnectionStatus sees the new state immediately + the
|
||
// _DeviceError mirror the driver's most-recent fault message. Pass newState explicitly
|
||
// so the refresh doesn't race the lock-release with state.HostState.
|
||
RefreshSystemTagSnapshot(state, overrideHostState: newState);
|
||
OnHostStatusChanged?.Invoke(this,
|
||
new HostStatusChangedEventArgs(state.Options.HostAddress, old, newState));
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.3 — rebuild a single device's <see cref="SystemTagSnapshot"/> from the
|
||
/// live <see cref="DeviceState.HostState"/>, the configured probe interval, the
|
||
/// count of discovered tags excluding <c>_System/*</c>, the most-recent driver-error
|
||
/// message, and the supplied last-scan duration. Called from probe transitions, the
|
||
/// end of <see cref="ReadAsync"/>, and at <see cref="InitializeAsync"/> seed time.
|
||
/// </summary>
|
||
private void RefreshSystemTagSnapshot(
|
||
DeviceState state, double? lastScanTimeMs = null, HostState? overrideHostState = null)
|
||
{
|
||
var tagCount = 0;
|
||
foreach (var t in _tagsByName.Values)
|
||
{
|
||
if (string.Equals(t.DeviceHostAddress, state.Options.HostAddress, StringComparison.OrdinalIgnoreCase)
|
||
&& !AbCipSystemTagSource.IsSystemReference(t.Name))
|
||
tagCount++;
|
||
}
|
||
var scanRateMs = _options.Probe.Interval.TotalMilliseconds;
|
||
var deviceError = _health.LastError ?? string.Empty;
|
||
// Caller can pass overrideHostState to dodge the read-from-volatile-state race that
|
||
// would otherwise sit between TransitionDeviceState's lock release + this refresh.
|
||
var connectionStatus = (overrideHostState ?? state.HostState).ToString();
|
||
var resolvedScan = lastScanTimeMs ?? state.LastScanTimeMs;
|
||
_systemTagSource.Update(state.Options.HostAddress, new SystemTagSnapshot(
|
||
ConnectionStatus: connectionStatus,
|
||
ScanRateMs: scanRateMs,
|
||
TagCount: tagCount,
|
||
DeviceError: deviceError,
|
||
LastScanTimeMs: resolvedScan));
|
||
}
|
||
|
||
// ---- IPerCallHostResolver ----
|
||
|
||
/// <summary>
|
||
/// Resolve the device host address for a given tag full-reference. Per plan decision #144
|
||
/// the Phase 6.1 resilience pipeline keys its bulkhead + breaker on
|
||
/// <c>(DriverInstanceId, hostName)</c> so multi-PLC drivers get per-device isolation —
|
||
/// one dead PLC trips only its own breaker. Unknown references fall back to the
|
||
/// first configured device's host address rather than throwing — the invoker handles the
|
||
/// mislookup at the capability level when the actual read returns BadNodeIdUnknown.
|
||
/// </summary>
|
||
public string ResolveHost(string fullReference)
|
||
{
|
||
// PR abcip-4.3 — _System/<deviceHostAddress>/<name> carries the device in its
|
||
// address path, so route on the embedded host directly rather than falling back
|
||
// to "first configured device" + having the bulkhead key collide across devices.
|
||
if (AbCipSystemTagSource.IsSystemReference(fullReference))
|
||
{
|
||
var host = ExtractSystemDeviceHost(fullReference);
|
||
if (host is not null) return host;
|
||
}
|
||
if (_tagsByName.TryGetValue(fullReference, out var def))
|
||
return def.DeviceHostAddress;
|
||
return _options.Devices.FirstOrDefault()?.HostAddress ?? DriverInstanceId;
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.3 — pull the device host address out of a <c>_System/<host>/<name></c>
|
||
/// reference. Splits on the last <c>'/'</c> so device hosts that themselves contain a
|
||
/// forward-slash (the canonical <c>ab://gateway/cip-path</c> form does) survive the
|
||
/// round-trip. Returns <c>null</c> when the reference doesn't match the expected shape.
|
||
/// </summary>
|
||
internal static string? ExtractSystemDeviceHost(string reference)
|
||
{
|
||
if (!AbCipSystemTagSource.IsSystemReference(reference)) return null;
|
||
var withoutPrefix = reference[AbCipSystemTagSource.SystemFolderPrefix.Length..];
|
||
var lastSlash = withoutPrefix.LastIndexOf('/');
|
||
if (lastSlash <= 0) return null;
|
||
return withoutPrefix[..lastSlash];
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.3 — pull the trailing system-tag name (e.g. <c>_ConnectionStatus</c>) out
|
||
/// of a <c>_System/<host>/<name></c> reference. Pairs with
|
||
/// <see cref="ExtractSystemDeviceHost"/>.
|
||
/// </summary>
|
||
internal static string? ExtractSystemTagName(string reference)
|
||
{
|
||
if (!AbCipSystemTagSource.IsSystemReference(reference)) return null;
|
||
var withoutPrefix = reference[AbCipSystemTagSource.SystemFolderPrefix.Length..];
|
||
var lastSlash = withoutPrefix.LastIndexOf('/');
|
||
if (lastSlash <= 0 || lastSlash >= withoutPrefix.Length - 1) return null;
|
||
return withoutPrefix[(lastSlash + 1)..];
|
||
}
|
||
|
||
// ---- IReadable ----
|
||
|
||
/// <summary>
|
||
/// Read each <c>fullReference</c> in order. Unknown tags surface as
|
||
/// <c>BadNodeIdUnknown</c>; libplctag-layer failures map through
|
||
/// <see cref="AbCipStatusMapper.MapLibplctagStatus"/>; any other exception becomes
|
||
/// <c>BadCommunicationError</c>. The driver health surface is updated per-call so the
|
||
/// Admin UI sees a tight feedback loop between read failures + the driver's state.
|
||
/// </summary>
|
||
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
|
||
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(fullReferences);
|
||
var now = DateTime.UtcNow;
|
||
var results = new DataValueSnapshot[fullReferences.Count];
|
||
var scanStart = System.Diagnostics.Stopwatch.GetTimestamp();
|
||
|
||
// PR abcip-3.2 — first-read symbol-walk for Logical-mode devices. Each device that
|
||
// resolved to Logical fires one @tags walk; subsequent reads consult the cached
|
||
// name → instance-id map. Devices in Symbolic mode skip the walk entirely.
|
||
await EnsureLogicalMappingsAsync(fullReferences, cancellationToken).ConfigureAwait(false);
|
||
|
||
// Task #194 — plan the batch: members of the same parent UDT get collapsed into one
|
||
// whole-UDT read + in-memory member decode; every other reference falls back to the
|
||
// per-tag path that's been here since PR 3. Planner is a pure function over the
|
||
// current tag map; BOOL/String/Structure members stay on the fallback path because
|
||
// declaration-only offsets can't place them under Logix alignment rules.
|
||
//
|
||
// PR abcip-3.3 — dispatch by device's resolved ReadStrategy:
|
||
// WholeUdt — every group goes through the whole-UDT planner (task #194 default).
|
||
// MultiPacket — every group goes through the multi-packet planner; one read per
|
||
// subscribed member, bundled per parent.
|
||
// Auto — per-group heuristic on subscribedMembers / totalMembers.
|
||
await ExecuteReadPlanAsync(fullReferences, results, now, cancellationToken).ConfigureAwait(false);
|
||
|
||
// PR abcip-4.3 — track wall-clock scan time per device that owned at least one ref in
|
||
// this batch. Surfaces as _System/_LastScanTimeMs; the snapshot refresh also picks up
|
||
// any health transitions / error messages that happened during the read.
|
||
var elapsedMs = (System.Diagnostics.Stopwatch.GetTimestamp() - scanStart)
|
||
* 1000.0 / System.Diagnostics.Stopwatch.Frequency;
|
||
var touched = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||
foreach (var fr in fullReferences)
|
||
{
|
||
string? deviceHost = null;
|
||
if (AbCipSystemTagSource.IsSystemReference(fr))
|
||
{
|
||
deviceHost = ExtractSystemDeviceHost(fr);
|
||
}
|
||
else if (_tagsByName.TryGetValue(fr, out var def))
|
||
{
|
||
deviceHost = def.DeviceHostAddress;
|
||
}
|
||
if (deviceHost is null || !touched.Add(deviceHost)) continue;
|
||
if (_devices.TryGetValue(deviceHost, out var state))
|
||
{
|
||
state.LastScanTimeMs = elapsedMs;
|
||
RefreshSystemTagSnapshot(state, elapsedMs);
|
||
}
|
||
}
|
||
return results;
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.3 — strategy-aware dispatch wrapper around the WholeUdt + MultiPacket
|
||
/// planners. Both planners produce the same shape of "groups + per-tag fallbacks" so
|
||
/// fallbacks always run through <see cref="ReadSingleAsync"/>; only the group shape
|
||
/// differs. Auto resolves per-group: members of the same parent UDT either flow through
|
||
/// <see cref="ReadGroupAsync"/> (one whole-UDT read) or
|
||
/// <see cref="ReadMultiPacketBatchAsync"/> (per-member reads bundled per parent).
|
||
/// </summary>
|
||
private async Task ExecuteReadPlanAsync(
|
||
IReadOnlyList<string> fullReferences, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||
{
|
||
// First pass — segregate references by parent UDT vs everything-else, identical to the
|
||
// shape both planners produce; we can then route each parent group through the chosen
|
||
// planner. Reuse AbCipUdtReadPlanner.Build for the WholeUdt+Auto path because it already
|
||
// demotes single-member groups to fallback, and the MultiPacket planner does NOT demote
|
||
// (sparse reads of one member are still a win on the wire).
|
||
var wholeUdtPlan = AbCipUdtReadPlanner.Build(fullReferences, _tagsByName);
|
||
var multiPacketPlan = AbCipMultiPacketReadPlanner.Build(fullReferences, _tagsByName);
|
||
|
||
// Determine per-parent strategy. WholeUdt planner emits a Group only when ≥2 members of
|
||
// the same parent are subscribed; MultiPacket planner emits a Batch for every parent
|
||
// touched. Index multiPacket batches by parent name so we can co-route them with the
|
||
// WholeUdt grouping decisions.
|
||
var multiPacketByParent = new Dictionary<string, AbCipMultiPacketReadBatch>(
|
||
multiPacketPlan.Batches.Count, StringComparer.OrdinalIgnoreCase);
|
||
foreach (var b in multiPacketPlan.Batches)
|
||
multiPacketByParent[b.ParentName] = b;
|
||
|
||
// Treat each parent that survived the WholeUdt planner as the candidate for whole-UDT
|
||
// dispatch; per-strategy routing decides whether it goes WholeUdt or MultiPacket.
|
||
var routedParents = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||
foreach (var group in wholeUdtPlan.Groups)
|
||
{
|
||
_devices.TryGetValue(group.ParentDefinition.DeviceHostAddress, out var device);
|
||
|
||
var strategy = ChooseEffectiveStrategy(device, group);
|
||
if (strategy == ReadStrategy.MultiPacket
|
||
&& multiPacketByParent.TryGetValue(group.ParentName, out var mpBatch))
|
||
{
|
||
if (device is not null) Interlocked.Increment(ref device.MultiPacketGroupsExecuted);
|
||
await ReadMultiPacketBatchAsync(mpBatch, results, now, ct).ConfigureAwait(false);
|
||
}
|
||
else
|
||
{
|
||
if (device is not null) Interlocked.Increment(ref device.WholeUdtGroupsExecuted);
|
||
await ReadGroupAsync(group, results, now, ct).ConfigureAwait(false);
|
||
}
|
||
routedParents.Add(group.ParentName);
|
||
}
|
||
|
||
// Per-tag fallbacks from the WholeUdt planner are the union of (a) tags with no UDT
|
||
// parent, (b) UDT members where the parent had only one subscribed member, and (c)
|
||
// unknown references. We re-route singletons through MultiPacket too when the device
|
||
// strategy is MultiPacket, because a 1-of-50 read is exactly the sparse case the planner
|
||
// is designed for.
|
||
foreach (var fb in wholeUdtPlan.Fallbacks)
|
||
{
|
||
// Is this a UDT-member fallback that the MultiPacket planner can route?
|
||
if (multiPacketPlan.Batches.Count > 0
|
||
&& _tagsByName.TryGetValue(fb.Reference, out var def))
|
||
{
|
||
var dot = fb.Reference.IndexOf('.');
|
||
if (dot > 0 && dot < fb.Reference.Length - 1)
|
||
{
|
||
var parentName = fb.Reference[..dot];
|
||
if (!routedParents.Contains(parentName)
|
||
&& multiPacketByParent.TryGetValue(parentName, out var mpBatch)
|
||
&& _devices.TryGetValue(def.DeviceHostAddress, out var device))
|
||
{
|
||
// Singleton parent — Auto picks based on the same heuristic; MultiPacket
|
||
// wins by default because the WholeUdt planner already demoted single
|
||
// members under the assumption that one whole-UDT read is no cheaper
|
||
// than one member read. With explicit MultiPacket the answer flips.
|
||
var strategy = ChooseEffectiveStrategyForSingleton(device, mpBatch);
|
||
if (strategy == ReadStrategy.MultiPacket)
|
||
{
|
||
Interlocked.Increment(ref device.MultiPacketGroupsExecuted);
|
||
await ReadMultiPacketBatchAsync(mpBatch, results, now, ct).ConfigureAwait(false);
|
||
routedParents.Add(parentName);
|
||
continue;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
await ReadSingleAsync(fb, fullReferences[fb.OriginalIndex], results, now, ct).ConfigureAwait(false);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.3 — pick the effective <see cref="ReadStrategy"/> for one parent UDT group.
|
||
/// <see cref="ReadStrategy.WholeUdt"/> + <see cref="ReadStrategy.MultiPacket"/> are
|
||
/// forced explicitly (already family-compat-checked at device init). <see cref="ReadStrategy.Auto"/>
|
||
/// consults the planner heuristic on subscribed-member fraction, but only when the
|
||
/// family supports request packing — non-packing families always read WholeUdt regardless
|
||
/// of sparsity because they have no Multi-Service-Packet path on the wire.
|
||
/// </summary>
|
||
private static ReadStrategy ChooseEffectiveStrategy(DeviceState? device, AbCipUdtReadGroup group)
|
||
{
|
||
if (device is null) return ReadStrategy.WholeUdt;
|
||
switch (device.ReadStrategy)
|
||
{
|
||
case ReadStrategy.MultiPacket:
|
||
return ReadStrategy.MultiPacket;
|
||
case ReadStrategy.WholeUdt:
|
||
return ReadStrategy.WholeUdt;
|
||
case ReadStrategy.Auto:
|
||
default:
|
||
if (!device.Profile.SupportsRequestPacking) return ReadStrategy.WholeUdt;
|
||
var totalMembers = group.ParentDefinition.Members?.Count ?? 0;
|
||
return AbCipMultiPacketReadPlanner.ChooseStrategyForGroup(
|
||
subscribedMembers: group.Members.Count,
|
||
totalMembers: totalMembers,
|
||
threshold: device.Options.MultiPacketSparsityThreshold);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.3 — strategy pick for a singleton (one-member) UDT batch. Only relevant
|
||
/// when the device strategy is explicit <see cref="ReadStrategy.MultiPacket"/> or Auto
|
||
/// produces a MultiPacket result; otherwise the per-tag fallback path runs as before.
|
||
/// </summary>
|
||
private static ReadStrategy ChooseEffectiveStrategyForSingleton(
|
||
DeviceState device, AbCipMultiPacketReadBatch batch)
|
||
{
|
||
switch (device.ReadStrategy)
|
||
{
|
||
case ReadStrategy.MultiPacket:
|
||
return ReadStrategy.MultiPacket;
|
||
case ReadStrategy.WholeUdt:
|
||
return ReadStrategy.WholeUdt;
|
||
case ReadStrategy.Auto:
|
||
default:
|
||
if (!device.Profile.SupportsRequestPacking) return ReadStrategy.WholeUdt;
|
||
var totalMembers = batch.ParentDefinition.Members?.Count ?? 0;
|
||
return AbCipMultiPacketReadPlanner.ChooseStrategyForGroup(
|
||
subscribedMembers: batch.Members.Count,
|
||
totalMembers: totalMembers,
|
||
threshold: device.Options.MultiPacketSparsityThreshold);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.3 — execute a Multi-Service-Packet batch. Today this issues one libplctag
|
||
/// read per member (the same N reads the per-tag fallback path does), keyed on the
|
||
/// batch's parent so the diagnostic counters track which strategy ran. Wire-level
|
||
/// Multi-Service-Packet bundling depends on the libplctag .NET wrapper exposing the
|
||
/// 0x0A service explicitly — same wrapper limitation as PR abcip-3.1's connection_size
|
||
/// and PR abcip-3.2's instance-ID addressing. The planner's grouping is still
|
||
/// load-bearing because it gives the runtime the correct plan when an upstream wrapper
|
||
/// release exposes the bundling primitive.
|
||
/// </summary>
|
||
private async Task ReadMultiPacketBatchAsync(
|
||
AbCipMultiPacketReadBatch batch, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||
{
|
||
var parent = batch.ParentDefinition;
|
||
if (!_devices.TryGetValue(parent.DeviceHostAddress, out var device))
|
||
{
|
||
foreach (var m in batch.Members)
|
||
results[m.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNodeIdUnknown, null, now);
|
||
return;
|
||
}
|
||
|
||
foreach (var member in batch.Members)
|
||
{
|
||
var memberFullName = member.Definition.Name;
|
||
var fb = new AbCipUdtReadFallback(member.OriginalIndex, memberFullName);
|
||
await ReadSingleAsync(fb, memberFullName, results, now, ct).ConfigureAwait(false);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.2 — for each Logical-mode device touched by this read batch, fire the
|
||
/// one-time <c>@tags</c> symbol-table walk + populate <see cref="DeviceState.LogicalInstanceMap"/>.
|
||
/// Subsequent reads short-circuit on <see cref="DeviceState.LogicalWalkComplete"/>;
|
||
/// concurrent first reads on the same device serialise on
|
||
/// <see cref="DeviceState.LogicalWalkLock"/> so the walk is dispatched once even under
|
||
/// parallel load.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>The walk uses the same <see cref="LibplctagTagEnumerator"/> as discovery —
|
||
/// reading <c>@tags</c> + decoding the Symbol Object response. Failures are intentionally
|
||
/// swallowed: an empty map after the walk-attempted flag flips means subsequent reads
|
||
/// fall back to Symbolic addressing on the wire (libplctag's default), which is the
|
||
/// same wire behaviour every previous build had. Driver health is not faulted because a
|
||
/// tag-list-walk failure does not actually block reads.</para>
|
||
/// </remarks>
|
||
private async Task EnsureLogicalMappingsAsync(
|
||
IReadOnlyList<string> fullReferences, CancellationToken ct)
|
||
{
|
||
// Find the unique set of Logical-mode devices the batch touches. Most batches touch
|
||
// one device, so the HashSet is a small allocation.
|
||
HashSet<DeviceState>? pending = null;
|
||
foreach (var reference in fullReferences)
|
||
{
|
||
if (!_tagsByName.TryGetValue(reference, out var def)) continue;
|
||
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue;
|
||
if (device.AddressingMode != AddressingMode.Logical) continue;
|
||
if (device.LogicalWalkComplete) continue;
|
||
(pending ??= []).Add(device);
|
||
}
|
||
if (pending is null) return;
|
||
|
||
foreach (var device in pending)
|
||
{
|
||
await device.LogicalWalkLock.WaitAsync(ct).ConfigureAwait(false);
|
||
try
|
||
{
|
||
if (device.LogicalWalkComplete) continue;
|
||
try
|
||
{
|
||
using var enumerator = _enumeratorFactory.Create();
|
||
var deviceParams = new AbCipTagCreateParams(
|
||
Gateway: device.ParsedAddress.Gateway,
|
||
Port: device.ParsedAddress.Port,
|
||
CipPath: device.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
|
||
TagName: "@tags",
|
||
Timeout: _options.Timeout,
|
||
ConnectionSize: device.ConnectionSize,
|
||
AddressingMode: AddressingMode.Symbolic);
|
||
|
||
// Symbol Object instance IDs aren't surfaced on AbCipDiscoveredTag yet — the
|
||
// record carries Name / ProgramScope / DataType / ReadOnly. We populate the
|
||
// map keyed on the Logix tag path the driver uses internally; the libplctag
|
||
// wrapper limitation (no public ConnectionSize / cip_addr knob in 1.5.x)
|
||
// means the value side stays unmapped for now and the runtime degrades to
|
||
// Symbolic on the wire. The map's presence is still load-bearing: it's
|
||
// observable from tests + future-proofs the driver for when an upstream
|
||
// wrapper release exposes the IDs through the enumerator + Tag attribute.
|
||
await foreach (var discovered in enumerator.EnumerateAsync(deviceParams, ct)
|
||
.ConfigureAwait(false))
|
||
{
|
||
if (discovered.IsSystemTag) continue;
|
||
if (AbCipSystemTagFilter.IsSystemTag(discovered.Name)) continue;
|
||
var fullName = discovered.ProgramScope is null
|
||
? discovered.Name
|
||
: $"Program:{discovered.ProgramScope}.{discovered.Name}";
|
||
// No instance ID in the current discovered-tag shape; record an
|
||
// entry so the runtime knows the symbol is part of the Logical
|
||
// resolution pass (the map's presence influences slice + parent-DINT
|
||
// creation). 0 is reserved by CIP for "not assigned" so it's a safe
|
||
// sentinel that the runtime's reflection guard treats as "missing".
|
||
device.LogicalInstanceMap[fullName] = 0u;
|
||
}
|
||
}
|
||
catch (OperationCanceledException) { throw; }
|
||
catch
|
||
{
|
||
// Walk failure is non-fatal — the driver keeps Logical mode set but
|
||
// every per-tag handle ends up using Symbolic addressing on the wire.
|
||
}
|
||
finally
|
||
{
|
||
device.LogicalWalkComplete = true;
|
||
}
|
||
}
|
||
finally
|
||
{
|
||
device.LogicalWalkLock.Release();
|
||
}
|
||
}
|
||
}
|
||
|
||
private async Task ReadSingleAsync(
|
||
AbCipUdtReadFallback fb, string reference, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||
{
|
||
// PR abcip-4.3 — synthetic _System/<deviceHost>/<name> reference; serve from the
|
||
// diagnostic snapshot instead of materialising a libplctag runtime.
|
||
if (AbCipSystemTagSource.IsSystemReference(reference))
|
||
{
|
||
var deviceHost = ExtractSystemDeviceHost(reference);
|
||
var nameUnderSystem = ExtractSystemTagName(reference);
|
||
if (deviceHost is not null && nameUnderSystem is not null
|
||
&& _systemTagSource.TryRead(nameUnderSystem, deviceHost, out var sysValue))
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(sysValue, AbCipStatusMapper.Good, now, now);
|
||
}
|
||
else
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNodeIdUnknown, null, now);
|
||
}
|
||
return;
|
||
}
|
||
|
||
if (!_tagsByName.TryGetValue(reference, out var def))
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNodeIdUnknown, null, now);
|
||
return;
|
||
}
|
||
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device))
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNodeIdUnknown, null, now);
|
||
return;
|
||
}
|
||
|
||
// PR abcip-1.3 — array-slice path. A tag whose TagPath ends in [N..M] dispatches to
|
||
// AbCipArrayReadPlanner: one libplctag tag-create with ElementCount=N issues one
|
||
// Rockwell array read; the contiguous buffer is decoded at element stride into a
|
||
// single snapshot whose Value is an object[] of the N elements.
|
||
var parsedPath = AbCipTagPath.TryParse(def.TagPath);
|
||
if (parsedPath?.Slice is not null)
|
||
{
|
||
await ReadSliceAsync(fb, def, parsedPath, device, results, now, ct).ConfigureAwait(false);
|
||
return;
|
||
}
|
||
|
||
try
|
||
{
|
||
var runtime = await EnsureTagRuntimeAsync(device, def, ct).ConfigureAwait(false);
|
||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||
|
||
var status = runtime.GetStatus();
|
||
if (status != 0)
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||
AbCipStatusMapper.MapLibplctagStatus(status), null, now);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
|
||
$"libplctag status {status} reading {reference}");
|
||
return;
|
||
}
|
||
|
||
var bitIndex = parsedPath?.BitIndex;
|
||
var value = runtime.DecodeValue(def.DataType, bitIndex);
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now);
|
||
_health = new DriverHealth(DriverState.Healthy, now, null);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
throw;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||
AbCipStatusMapper.BadCommunicationError, null, now);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-1.3 — slice read path. Builds an <see cref="AbCipArrayReadPlan"/> from the
|
||
/// parsed slice path, materialises a per-tag runtime keyed by the tag's full name (so
|
||
/// repeat reads reuse the same libplctag handle), issues one PLC array read, and
|
||
/// decodes the contiguous buffer into <c>object?[]</c> at element stride. Unsupported
|
||
/// element types fall back to <see cref="AbCipStatusMapper.BadNotSupported"/>.
|
||
/// </summary>
|
||
private async Task ReadSliceAsync(
|
||
AbCipUdtReadFallback fb, AbCipTagDefinition def, AbCipTagPath parsedPath,
|
||
DeviceState device, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||
{
|
||
// PR abcip-3.2 — slice reads piggyback on the device's resolved addressing mode. Logical
|
||
// mode looks up the parent array tag's instance ID via the @tags map; null-fallback to
|
||
// Symbolic when the array isn't in the map (e.g. @tags walk hasn't populated the entry).
|
||
uint? sliceLogicalId = null;
|
||
if (device.AddressingMode == AddressingMode.Logical
|
||
&& device.LogicalInstanceMap.TryGetValue(def.TagPath, out var sliceId))
|
||
{
|
||
sliceLogicalId = sliceId;
|
||
}
|
||
var baseParams = new AbCipTagCreateParams(
|
||
Gateway: device.ParsedAddress.Gateway,
|
||
Port: device.ParsedAddress.Port,
|
||
CipPath: device.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
|
||
TagName: parsedPath.ToLibplctagName(),
|
||
Timeout: _options.Timeout,
|
||
ConnectionSize: device.ConnectionSize,
|
||
AddressingMode: device.AddressingMode,
|
||
LogicalInstanceId: sliceLogicalId);
|
||
|
||
var plan = AbCipArrayReadPlanner.TryBuild(def, parsedPath, baseParams);
|
||
if (plan is null)
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||
AbCipStatusMapper.BadNotSupported, null, now);
|
||
return;
|
||
}
|
||
|
||
try
|
||
{
|
||
var runtime = await EnsureSliceRuntimeAsync(device, def.Name, plan.CreateParams, ct)
|
||
.ConfigureAwait(false);
|
||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||
|
||
var status = runtime.GetStatus();
|
||
if (status != 0)
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||
AbCipStatusMapper.MapLibplctagStatus(status), null, now);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
|
||
$"libplctag status {status} reading slice {def.Name}");
|
||
return;
|
||
}
|
||
|
||
var values = AbCipArrayReadPlanner.Decode(plan, runtime);
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(values, AbCipStatusMapper.Good, now, now);
|
||
_health = new DriverHealth(DriverState.Healthy, now, null);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
throw;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
results[fb.OriginalIndex] = new DataValueSnapshot(null,
|
||
AbCipStatusMapper.BadCommunicationError, null, now);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Idempotently materialise a slice-read runtime. Slice runtimes share the device's
|
||
/// <see cref="DeviceState.Runtimes"/> dict keyed by the tag's full name so repeated
|
||
/// reads reuse the same libplctag handle without re-creating the native tag every poll.
|
||
/// </summary>
|
||
private async Task<IAbCipTagRuntime> EnsureSliceRuntimeAsync(
|
||
DeviceState device, string tagName, AbCipTagCreateParams createParams, CancellationToken ct)
|
||
{
|
||
if (device.Runtimes.TryGetValue(tagName, out var existing)) return existing;
|
||
|
||
var runtime = _tagFactory.Create(createParams);
|
||
try
|
||
{
|
||
await runtime.InitializeAsync(ct).ConfigureAwait(false);
|
||
}
|
||
catch
|
||
{
|
||
runtime.Dispose();
|
||
throw;
|
||
}
|
||
device.Runtimes[tagName] = runtime;
|
||
return runtime;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Task #194 — perform one whole-UDT read on the parent tag, then decode each
|
||
/// grouped member from the runtime's buffer at its computed byte offset. A per-group
|
||
/// failure (parent read raised, non-zero libplctag status, or missing device) stamps
|
||
/// the mapped fault across every grouped member only — sibling groups + the
|
||
/// per-tag fallback list are unaffected.
|
||
/// </summary>
|
||
private async Task ReadGroupAsync(
|
||
AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
|
||
{
|
||
var parent = group.ParentDefinition;
|
||
|
||
if (!_devices.TryGetValue(parent.DeviceHostAddress, out var device))
|
||
{
|
||
StampGroupStatus(group, results, now, AbCipStatusMapper.BadNodeIdUnknown);
|
||
return;
|
||
}
|
||
|
||
try
|
||
{
|
||
var runtime = await EnsureTagRuntimeAsync(device, parent, ct).ConfigureAwait(false);
|
||
await runtime.ReadAsync(ct).ConfigureAwait(false);
|
||
|
||
var status = runtime.GetStatus();
|
||
if (status != 0)
|
||
{
|
||
var mapped = AbCipStatusMapper.MapLibplctagStatus(status);
|
||
StampGroupStatus(group, results, now, mapped);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead,
|
||
$"libplctag status {status} reading UDT {group.ParentName}");
|
||
return;
|
||
}
|
||
|
||
foreach (var member in group.Members)
|
||
{
|
||
var value = runtime.DecodeValueAt(member.Definition.DataType, member.Offset, bitIndex: null);
|
||
results[member.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now);
|
||
}
|
||
_health = new DriverHealth(DriverState.Healthy, now, null);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
throw;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
StampGroupStatus(group, results, now, AbCipStatusMapper.BadCommunicationError);
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
}
|
||
}
|
||
|
||
private static void StampGroupStatus(
|
||
AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, uint statusCode)
|
||
{
|
||
foreach (var member in group.Members)
|
||
results[member.OriginalIndex] = new DataValueSnapshot(null, statusCode, null, now);
|
||
}
|
||
|
||
// ---- IWritable ----
|
||
|
||
/// <summary>
|
||
/// Write each request in the batch. Writes are NOT auto-retried by the driver — per
|
||
/// plan decisions #44, #45, #143 the caller opts in via
|
||
/// <see cref="AbCipTagDefinition.WriteIdempotent"/> and the resilience pipeline (layered
|
||
/// above the driver) decides whether to replay. Non-writable configurations surface as
|
||
/// <c>BadNotWritable</c>; type-conversion failures as <c>BadTypeMismatch</c>; transport
|
||
/// errors as <c>BadCommunicationError</c>.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// PR abcip-1.4 — multi-tag write packing. Writes are grouped by device via
|
||
/// <see cref="AbCipMultiWritePlanner"/>. Devices whose family
|
||
/// <see cref="AbCipPlcFamilyProfile.SupportsRequestPacking"/> is <c>true</c> dispatch
|
||
/// their packable writes concurrently so libplctag's native scheduler can coalesce them
|
||
/// onto one CIP Multi-Service Packet (0x0A) per round-trip; Micro800 (no packing) still
|
||
/// issues writes one-at-a-time. BOOL-within-DINT writes always go through the RMW path
|
||
/// under a per-parent semaphore, regardless of the family flag, because two concurrent
|
||
/// RMWs on the same DINT could lose one another's update. Per-tag StatusCodes are
|
||
/// preserved in the caller's input order on partial failures.
|
||
/// </remarks>
|
||
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
|
||
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(writes);
|
||
var results = new WriteResult[writes.Count];
|
||
|
||
var plans = AbCipMultiWritePlanner.Build(
|
||
writes, _tagsByName, _devices,
|
||
reportPreflight: (idx, code) => results[idx] = new WriteResult(code));
|
||
|
||
foreach (var plan in plans)
|
||
{
|
||
if (!_devices.TryGetValue(plan.DeviceHostAddress, out var device))
|
||
{
|
||
foreach (var e in plan.Packable) results[e.OriginalIndex] = new WriteResult(AbCipStatusMapper.BadNodeIdUnknown);
|
||
foreach (var e in plan.BitRmw) results[e.OriginalIndex] = new WriteResult(AbCipStatusMapper.BadNodeIdUnknown);
|
||
continue;
|
||
}
|
||
|
||
// Bit-RMW writes always serialise per-parent — never packed.
|
||
foreach (var entry in plan.BitRmw)
|
||
results[entry.OriginalIndex] = new WriteResult(
|
||
await ExecuteBitRmwWriteAsync(device, entry, cancellationToken).ConfigureAwait(false));
|
||
|
||
if (plan.Packable.Count == 0) continue;
|
||
|
||
if (plan.Profile.SupportsRequestPacking && plan.Packable.Count > 1)
|
||
{
|
||
// Concurrent dispatch — libplctag's native scheduler packs same-connection writes
|
||
// into one Multi-Service Packet when the family supports it.
|
||
var tasks = new Task<(int idx, uint code)>[plan.Packable.Count];
|
||
for (var i = 0; i < plan.Packable.Count; i++)
|
||
{
|
||
var entry = plan.Packable[i];
|
||
tasks[i] = ExecutePackableWriteAsync(device, entry, cancellationToken);
|
||
}
|
||
var outcomes = await Task.WhenAll(tasks).ConfigureAwait(false);
|
||
foreach (var (idx, code) in outcomes)
|
||
results[idx] = new WriteResult(code);
|
||
}
|
||
else
|
||
{
|
||
// Single-write groups + Micro800 (SupportsRequestPacking=false) — sequential.
|
||
foreach (var entry in plan.Packable)
|
||
{
|
||
var code = await ExecutePackableWriteAsync(device, entry, cancellationToken)
|
||
.ConfigureAwait(false);
|
||
results[entry.OriginalIndex] = new WriteResult(code.code);
|
||
}
|
||
}
|
||
}
|
||
|
||
return results;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Execute one packable write — encode the value into the per-tag runtime, flush, and
|
||
/// map the resulting libplctag status. Exception-to-StatusCode mapping mirrors the
|
||
/// pre-1.4 per-tag loop so callers see no behaviour change for individual writes.
|
||
/// </summary>
|
||
private async Task<(int idx, uint code)> ExecutePackableWriteAsync(
|
||
DeviceState device, AbCipMultiWritePlanner.ClassifiedWrite entry, CancellationToken ct)
|
||
{
|
||
var def = entry.Definition;
|
||
var w = entry.Request;
|
||
var now = DateTime.UtcNow;
|
||
|
||
// PR abcip-4.2 — write deadband / write-on-change. Consult the coalescer first; a
|
||
// suppression decision returns Good without hitting libplctag so the OPC UA client sees
|
||
// the same write semantics it always has, the wire just doesn't move. Driver health is
|
||
// intentionally left alone on suppression — a coalesced write is neither a success nor
|
||
// a failure of the underlying connection. Bit-RMW writes go through their own path
|
||
// (ExecuteBitRmwWriteAsync) which has its own coalescer call site.
|
||
if (_writeCoalescer.ShouldSuppress(def.DeviceHostAddress, def, w.Value))
|
||
return (entry.OriginalIndex, AbCipStatusMapper.Good);
|
||
|
||
try
|
||
{
|
||
var runtime = await EnsureTagRuntimeAsync(device, def, ct).ConfigureAwait(false);
|
||
runtime.EncodeValue(def.DataType, entry.ParsedPath?.BitIndex, w.Value);
|
||
await runtime.WriteAsync(ct).ConfigureAwait(false);
|
||
|
||
var status = runtime.GetStatus();
|
||
if (status == 0)
|
||
{
|
||
_health = new DriverHealth(DriverState.Healthy, now, null);
|
||
_writeCoalescer.Record(def.DeviceHostAddress, def, w.Value);
|
||
return (entry.OriginalIndex, AbCipStatusMapper.Good);
|
||
}
|
||
return (entry.OriginalIndex, AbCipStatusMapper.MapLibplctagStatus(status));
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
throw;
|
||
}
|
||
catch (NotSupportedException nse)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, nse.Message);
|
||
return (entry.OriginalIndex, AbCipStatusMapper.BadNotSupported);
|
||
}
|
||
catch (FormatException fe)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, fe.Message);
|
||
return (entry.OriginalIndex, AbCipStatusMapper.BadTypeMismatch);
|
||
}
|
||
catch (InvalidCastException ice)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ice.Message);
|
||
return (entry.OriginalIndex, AbCipStatusMapper.BadTypeMismatch);
|
||
}
|
||
catch (OverflowException oe)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, oe.Message);
|
||
return (entry.OriginalIndex, AbCipStatusMapper.BadOutOfRange);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
return (entry.OriginalIndex, AbCipStatusMapper.BadCommunicationError);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Execute one BOOL-within-DINT write through <see cref="WriteBitInDIntAsync"/>, with
|
||
/// the same exception-mapping fan-out as the pre-1.4 per-tag loop. Bit RMWs cannot be
|
||
/// packed because two concurrent writes against the same parent DINT would race their
|
||
/// read-modify-write windows.
|
||
/// </summary>
|
||
private async Task<uint> ExecuteBitRmwWriteAsync(
|
||
DeviceState device, AbCipMultiWritePlanner.ClassifiedWrite entry, CancellationToken ct)
|
||
{
|
||
// PR abcip-4.2 — bit-RMW writes go through the coalescer too. The deadband path is
|
||
// never useful on a single-bit BOOL (deadband < 1 collapses to equality) but
|
||
// WriteOnChange is — a UI that toggles a SetPoint.Reset bit at every cycle benefits
|
||
// from suppressing the redundant pulses.
|
||
var def = entry.Definition;
|
||
if (_writeCoalescer.ShouldSuppress(def.DeviceHostAddress, def, entry.Request.Value))
|
||
return AbCipStatusMapper.Good;
|
||
|
||
try
|
||
{
|
||
var bit = entry.ParsedPath!.BitIndex!.Value;
|
||
var code = await WriteBitInDIntAsync(device, entry.ParsedPath, bit, entry.Request.Value, ct)
|
||
.ConfigureAwait(false);
|
||
if (code == AbCipStatusMapper.Good)
|
||
{
|
||
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
|
||
_writeCoalescer.Record(def.DeviceHostAddress, def, entry.Request.Value);
|
||
}
|
||
return code;
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
throw;
|
||
}
|
||
catch (NotSupportedException nse)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, nse.Message);
|
||
return AbCipStatusMapper.BadNotSupported;
|
||
}
|
||
catch (FormatException fe)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, fe.Message);
|
||
return AbCipStatusMapper.BadTypeMismatch;
|
||
}
|
||
catch (InvalidCastException ice)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ice.Message);
|
||
return AbCipStatusMapper.BadTypeMismatch;
|
||
}
|
||
catch (OverflowException oe)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, oe.Message);
|
||
return AbCipStatusMapper.BadOutOfRange;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
|
||
return AbCipStatusMapper.BadCommunicationError;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Read-modify-write one bit within a DINT parent. Creates / reuses a parallel
|
||
/// parent-DINT runtime (distinct from the bit-selector handle) + serialises concurrent
|
||
/// writers against the same parent via a per-parent <see cref="SemaphoreSlim"/>.
|
||
/// Matches the Modbus BitInRegister + FOCAS PMC Bit pattern shipped in pass 1 of task #181.
|
||
/// </summary>
|
||
private async Task<uint> WriteBitInDIntAsync(
|
||
DeviceState device, AbCipTagPath bitPath, int bit, object? value, CancellationToken ct)
|
||
{
|
||
var parentPath = bitPath with { BitIndex = null };
|
||
var parentName = parentPath.ToLibplctagName();
|
||
|
||
var rmwLock = device.GetRmwLock(parentName);
|
||
await rmwLock.WaitAsync(ct).ConfigureAwait(false);
|
||
try
|
||
{
|
||
var parentRuntime = await EnsureParentRuntimeAsync(device, parentName, ct).ConfigureAwait(false);
|
||
await parentRuntime.ReadAsync(ct).ConfigureAwait(false);
|
||
var readStatus = parentRuntime.GetStatus();
|
||
if (readStatus != 0) return AbCipStatusMapper.MapLibplctagStatus(readStatus);
|
||
|
||
var current = Convert.ToInt32(parentRuntime.DecodeValue(AbCipDataType.DInt, bitIndex: null) ?? 0);
|
||
var updated = Convert.ToBoolean(value)
|
||
? current | (1 << bit)
|
||
: current & ~(1 << bit);
|
||
|
||
parentRuntime.EncodeValue(AbCipDataType.DInt, bitIndex: null, updated);
|
||
await parentRuntime.WriteAsync(ct).ConfigureAwait(false);
|
||
var writeStatus = parentRuntime.GetStatus();
|
||
return writeStatus == 0
|
||
? AbCipStatusMapper.Good
|
||
: AbCipStatusMapper.MapLibplctagStatus(writeStatus);
|
||
}
|
||
finally
|
||
{
|
||
rmwLock.Release();
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Get or lazily create a parent-DINT runtime for a parent tag path, cached per-device
|
||
/// so repeated bit writes against the same DINT share one handle.
|
||
/// </summary>
|
||
private async Task<IAbCipTagRuntime> EnsureParentRuntimeAsync(
|
||
DeviceState device, string parentTagName, CancellationToken ct)
|
||
{
|
||
if (device.ParentRuntimes.TryGetValue(parentTagName, out var existing)) return existing;
|
||
|
||
// PR abcip-3.2 — parent-DINT runtimes follow the device's resolved addressing mode so
|
||
// BOOL-in-DINT RMW reads/writes share Logical-mode benefits when the parent has been
|
||
// mapped. When the parent isn't in the @tags map (or Symbolic is the resolved mode),
|
||
// libplctag falls back to ASCII addressing transparently.
|
||
uint? parentLogicalId = null;
|
||
if (device.AddressingMode == AddressingMode.Logical
|
||
&& device.LogicalInstanceMap.TryGetValue(parentTagName, out var pid))
|
||
{
|
||
parentLogicalId = pid;
|
||
}
|
||
var runtime = _tagFactory.Create(new AbCipTagCreateParams(
|
||
Gateway: device.ParsedAddress.Gateway,
|
||
Port: device.ParsedAddress.Port,
|
||
CipPath: device.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
|
||
TagName: parentTagName,
|
||
Timeout: _options.Timeout,
|
||
ConnectionSize: device.ConnectionSize,
|
||
AddressingMode: device.AddressingMode,
|
||
LogicalInstanceId: parentLogicalId));
|
||
try
|
||
{
|
||
await runtime.InitializeAsync(ct).ConfigureAwait(false);
|
||
}
|
||
catch
|
||
{
|
||
runtime.Dispose();
|
||
throw;
|
||
}
|
||
device.ParentRuntimes[parentTagName] = runtime;
|
||
return runtime;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Idempotently materialise the runtime handle for a tag definition. First call creates
|
||
/// + initialises the libplctag Tag; subsequent calls reuse the cached handle for the
|
||
/// lifetime of the device.
|
||
/// </summary>
|
||
private async Task<IAbCipTagRuntime> EnsureTagRuntimeAsync(
|
||
DeviceState device, AbCipTagDefinition def, CancellationToken ct)
|
||
{
|
||
if (device.Runtimes.TryGetValue(def.Name, out var existing)) return existing;
|
||
|
||
var parsed = AbCipTagPath.TryParse(def.TagPath)
|
||
?? throw new InvalidOperationException(
|
||
$"AbCip tag '{def.Name}' has malformed TagPath '{def.TagPath}'.");
|
||
|
||
// PR abcip-3.2 — Logical-mode devices look up the controller-assigned Symbol Object
|
||
// instance ID for this tag from the one-time @tags walk; missing entries fall back to
|
||
// Symbolic addressing for this handle (the runtime detects LogicalInstanceId == null).
|
||
uint? logicalId = null;
|
||
if (device.AddressingMode == AddressingMode.Logical
|
||
&& device.LogicalInstanceMap.TryGetValue(def.TagPath, out var resolvedId))
|
||
{
|
||
logicalId = resolvedId;
|
||
}
|
||
|
||
var runtime = _tagFactory.Create(new AbCipTagCreateParams(
|
||
Gateway: device.ParsedAddress.Gateway,
|
||
Port: device.ParsedAddress.Port,
|
||
CipPath: device.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
|
||
TagName: parsed.ToLibplctagName(),
|
||
Timeout: _options.Timeout,
|
||
StringMaxCapacity: def.DataType == AbCipDataType.String ? def.StringLength : null,
|
||
ConnectionSize: device.ConnectionSize,
|
||
AddressingMode: device.AddressingMode,
|
||
LogicalInstanceId: logicalId));
|
||
try
|
||
{
|
||
await runtime.InitializeAsync(ct).ConfigureAwait(false);
|
||
}
|
||
catch
|
||
{
|
||
runtime.Dispose();
|
||
throw;
|
||
}
|
||
device.Runtimes[def.Name] = runtime;
|
||
return runtime;
|
||
}
|
||
|
||
public DriverHealth GetHealth() => _health with { Diagnostics = BuildDiagnostics() };
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.2 — driver-attributable counter snapshot exposed via
|
||
/// <see cref="DriverHealth.Diagnostics"/> + the <c>driver-diagnostics</c> RPC. Names use
|
||
/// the <c>"<DriverType>.<Counter>"</c> convention so the Admin UI can render
|
||
/// them alongside Modbus / S7 / OPC UA Client metrics without per-driver special-casing.
|
||
/// Counters today: <c>AbCip.WritesSuppressed</c> (writes the coalescer skipped because
|
||
/// deadband / write-on-change suppressed them) and <c>AbCip.WritesPassedThrough</c>
|
||
/// (writes that hit the wire after consulting the coalescer). Future PRs add CIP-level
|
||
/// counters (Forward Open count, multi-service-packet ratio, etc.) by extending this
|
||
/// dictionary.
|
||
/// </summary>
|
||
private IReadOnlyDictionary<string, double> BuildDiagnostics() => new Dictionary<string, double>
|
||
{
|
||
["AbCip.WritesSuppressed"] = _writeCoalescer.TotalWritesSuppressed,
|
||
["AbCip.WritesPassedThrough"] = _writeCoalescer.TotalWritesPassedThrough,
|
||
};
|
||
|
||
/// <summary>
|
||
/// Test seam — exposes the live coalescer for unit tests that want to inspect counters
|
||
/// without rebuilding the diagnostics dictionary on every assertion.
|
||
/// </summary>
|
||
internal AbCipWriteCoalescer WriteCoalescer => _writeCoalescer;
|
||
|
||
/// <summary>
|
||
/// CLR-visible allocation footprint only — libplctag's native heap is invisible to the
|
||
/// GC. driver-specs.md §3 flags this: operators must watch whole-process RSS for the
|
||
/// full picture, and <see cref="ReinitializeAsync"/> is the Tier-B remediation.
|
||
/// </summary>
|
||
public long GetMemoryFootprint() => 0;
|
||
|
||
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken)
|
||
{
|
||
_templateCache.Clear();
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
// ---- ITagDiscovery ----
|
||
|
||
/// <summary>
|
||
/// Stream the driver's tag set into the builder. Pre-declared tags from
|
||
/// <see cref="AbCipDriverOptions.Tags"/> emit first; optionally, the
|
||
/// <see cref="IAbCipTagEnumerator"/> walks each device's symbol table and adds
|
||
/// controller-discovered tags under a <c>Discovered/</c> sub-folder. System / module /
|
||
/// routine / task tags are hidden via <see cref="AbCipSystemTagFilter"/>.
|
||
/// </summary>
|
||
public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(builder);
|
||
await _discoverySemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||
try
|
||
{
|
||
await DiscoverCoreAsync(builder, cancellationToken).ConfigureAwait(false);
|
||
}
|
||
finally
|
||
{
|
||
_discoverySemaphore.Release();
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-2.5 — operator-triggered rebrowse. Drops the cached UDT template shapes so
|
||
/// the next read re-fetches them from the controller, then runs the same enumerator
|
||
/// walk + builder fan-out that <see cref="DiscoverAsync"/> drives. Serialised against
|
||
/// other rebrowse / discovery passes via <see cref="_discoverySemaphore"/> so two
|
||
/// concurrent triggers don't double-issue the @tags read.
|
||
/// </summary>
|
||
public async Task RebrowseAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(builder);
|
||
await _discoverySemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||
try
|
||
{
|
||
// Stale template shapes can outlive a controller program-download, so a rebrowse
|
||
// is the natural moment to drop them; subsequent UDT reads re-populate on demand.
|
||
_templateCache.Clear();
|
||
await DiscoverCoreAsync(builder, cancellationToken).ConfigureAwait(false);
|
||
}
|
||
finally
|
||
{
|
||
_discoverySemaphore.Release();
|
||
}
|
||
}
|
||
|
||
private async Task DiscoverCoreAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
|
||
{
|
||
var root = builder.Folder("AbCip", "AbCip");
|
||
|
||
foreach (var device in _options.Devices)
|
||
{
|
||
var deviceLabel = device.DeviceName ?? device.HostAddress;
|
||
var deviceFolder = root.Folder(device.HostAddress, deviceLabel);
|
||
|
||
// PR abcip-4.3 — diagnostic / system tags. Five read-only variables under
|
||
// _System/, each FullName-prefixed with _System/<deviceHostAddress>/ so the
|
||
// ReadAsync dispatcher can route by device without an additional registry. PR 4.4
|
||
// will turn _RefreshTagDb into a writeable refresh trigger; everything 4.3 ships
|
||
// is ViewOnly.
|
||
EmitSystemTagFolder(deviceFolder, device.HostAddress);
|
||
|
||
// Pre-declared tags — always emitted; the primary config path. UDT tags with declared
|
||
// Members fan out into a sub-folder + one Variable per member instead of a single
|
||
// Structure Variable (Structure has no useful scalar value + member-addressable paths
|
||
// are what downstream consumers actually want).
|
||
var preDeclared = _options.Tags.Where(t =>
|
||
string.Equals(t.DeviceHostAddress, device.HostAddress, StringComparison.OrdinalIgnoreCase));
|
||
foreach (var tag in preDeclared)
|
||
{
|
||
if (AbCipSystemTagFilter.IsSystemTag(tag.Name)) continue;
|
||
|
||
if (tag.DataType == AbCipDataType.Structure && tag.Members is { Count: > 0 })
|
||
{
|
||
var udtFolder = deviceFolder.Folder(tag.Name, tag.Name);
|
||
// PR abcip-2.6 — AOI-aware fan-out. When any member carries a non-Local
|
||
// AoiQualifier the tag is treated as an AOI instance: Input / Output / InOut
|
||
// members get grouped under sub-folders (Inputs/, Outputs/, InOut/) so the
|
||
// browse tree visually matches Studio 5000's AOI parameter tabs. Plain UDT
|
||
// tags (every member Local) retain the pre-2.6 flat layout under the parent
|
||
// folder so existing browse paths stay stable.
|
||
var hasDirectional = tag.Members.Any(m => m.AoiQualifier != AoiQualifier.Local);
|
||
IAddressSpaceBuilder? inputsFolder = null;
|
||
IAddressSpaceBuilder? outputsFolder = null;
|
||
IAddressSpaceBuilder? inOutFolder = null;
|
||
foreach (var member in tag.Members)
|
||
{
|
||
var parentFolder = udtFolder;
|
||
if (hasDirectional)
|
||
{
|
||
parentFolder = member.AoiQualifier switch
|
||
{
|
||
AoiQualifier.Input => inputsFolder ??= udtFolder.Folder("Inputs", "Inputs"),
|
||
AoiQualifier.Output => outputsFolder ??= udtFolder.Folder("Outputs", "Outputs"),
|
||
AoiQualifier.InOut => inOutFolder ??= udtFolder.Folder("InOut", "InOut"),
|
||
_ => udtFolder, // Local stays at the AOI root
|
||
};
|
||
}
|
||
var memberFullName = $"{tag.Name}.{member.Name}";
|
||
parentFolder.Variable(member.Name, member.Name, new DriverAttributeInfo(
|
||
FullName: memberFullName,
|
||
DriverDataType: member.DataType.ToDriverDataType(),
|
||
IsArray: false,
|
||
ArrayDim: null,
|
||
SecurityClass: member.Writable
|
||
? SecurityClassification.Operate
|
||
: SecurityClassification.ViewOnly,
|
||
IsHistorized: false,
|
||
IsAlarm: false,
|
||
WriteIdempotent: member.WriteIdempotent,
|
||
Description: member.Description));
|
||
}
|
||
continue;
|
||
}
|
||
|
||
deviceFolder.Variable(tag.Name, tag.Name, ToAttributeInfo(tag));
|
||
}
|
||
|
||
// Controller-discovered tags — opt-in via EnableControllerBrowse. The real @tags
|
||
// walker (LibplctagTagEnumerator) is the factory default since task #178 shipped,
|
||
// so leaving the flag off keeps the strict-config path for deployments where only
|
||
// declared tags should appear.
|
||
if (_options.EnableControllerBrowse && _devices.TryGetValue(device.HostAddress, out var state))
|
||
{
|
||
using var enumerator = _enumeratorFactory.Create();
|
||
// The @tags walker reads the controller's Symbol Object class 0x6B directly +
|
||
// does not need the driver's per-tag addressing-mode plumbing — it already
|
||
// operates on instance-ID semantics by definition. Pin Symbolic so libplctag
|
||
// doesn't try to layer Logical-mode attributes on top of @tags.
|
||
var deviceParams = new AbCipTagCreateParams(
|
||
Gateway: state.ParsedAddress.Gateway,
|
||
Port: state.ParsedAddress.Port,
|
||
CipPath: state.ParsedAddress.CipPath,
|
||
LibplctagPlcAttribute: state.Profile.LibplctagPlcAttribute,
|
||
TagName: "@tags",
|
||
Timeout: _options.Timeout,
|
||
ConnectionSize: state.ConnectionSize,
|
||
AddressingMode: AddressingMode.Symbolic);
|
||
|
||
IAddressSpaceBuilder? discoveredFolder = null;
|
||
await foreach (var discovered in enumerator.EnumerateAsync(deviceParams, cancellationToken)
|
||
.ConfigureAwait(false))
|
||
{
|
||
if (discovered.IsSystemTag) continue;
|
||
if (AbCipSystemTagFilter.IsSystemTag(discovered.Name)) continue;
|
||
|
||
discoveredFolder ??= deviceFolder.Folder("Discovered", "Discovered");
|
||
var fullName = discovered.ProgramScope is null
|
||
? discovered.Name
|
||
: $"Program:{discovered.ProgramScope}.{discovered.Name}";
|
||
discoveredFolder.Variable(fullName, discovered.Name, new DriverAttributeInfo(
|
||
FullName: fullName,
|
||
DriverDataType: discovered.DataType.ToDriverDataType(),
|
||
IsArray: false,
|
||
ArrayDim: null,
|
||
SecurityClass: discovered.ReadOnly
|
||
? SecurityClassification.ViewOnly
|
||
: SecurityClassification.Operate,
|
||
IsHistorized: false,
|
||
IsAlarm: false,
|
||
WriteIdempotent: false));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.3 — emit the per-device <c>_System</c> folder + its five read-only
|
||
/// diagnostic variables. The <c>FullName</c> on each variable encodes the owning
|
||
/// device's host address (<c>_System/<host>/<name></c>) so the read path
|
||
/// can route to <see cref="AbCipSystemTagSource.TryRead"/> without a separate
|
||
/// registry. Names + types stay in lockstep with
|
||
/// <see cref="AbCipSystemTagSource.SystemTagNames"/>.
|
||
/// </summary>
|
||
private static void EmitSystemTagFolder(IAddressSpaceBuilder deviceFolder, string deviceHostAddress)
|
||
{
|
||
var systemFolder = deviceFolder.Folder("_System", "_System");
|
||
EmitSystemVariable(systemFolder, deviceHostAddress, "_ConnectionStatus", DriverDataType.String);
|
||
EmitSystemVariable(systemFolder, deviceHostAddress, "_ScanRate", DriverDataType.Float64);
|
||
EmitSystemVariable(systemFolder, deviceHostAddress, "_TagCount", DriverDataType.Int32);
|
||
EmitSystemVariable(systemFolder, deviceHostAddress, "_DeviceError", DriverDataType.String);
|
||
EmitSystemVariable(systemFolder, deviceHostAddress, "_LastScanTimeMs", DriverDataType.Float64);
|
||
}
|
||
|
||
private static void EmitSystemVariable(
|
||
IAddressSpaceBuilder systemFolder, string deviceHostAddress, string name, DriverDataType type)
|
||
{
|
||
var fullName = $"{AbCipSystemTagSource.SystemFolderPrefix}{deviceHostAddress}/{name}";
|
||
systemFolder.Variable(name, name, new DriverAttributeInfo(
|
||
FullName: fullName,
|
||
DriverDataType: type,
|
||
IsArray: false,
|
||
ArrayDim: null,
|
||
// Read-only for now — PR abcip-4.4 will flip _RefreshTagDb to Operate when the
|
||
// refresh trigger lands. Today the AbCip system folder has no writeable members.
|
||
SecurityClass: SecurityClassification.ViewOnly,
|
||
IsHistorized: false,
|
||
IsAlarm: false,
|
||
WriteIdempotent: false,
|
||
Description: name switch
|
||
{
|
||
"_ConnectionStatus" => "Live HostState (Running / Stopped / Unknown / Faulted) — driven by the connectivity probe.",
|
||
"_ScanRate" => "Configured probe / poll interval in milliseconds.",
|
||
"_TagCount" => "Count of discovered tags on this device, excluding _System.",
|
||
"_DeviceError" => "Most recent driver-error message; empty when the device is healthy.",
|
||
"_LastScanTimeMs" => "Wall-clock duration of the most recent ReadAsync iteration on this device, in milliseconds.",
|
||
_ => null,
|
||
}));
|
||
}
|
||
|
||
/// <summary>Test seam — exposes the live system-tag source so unit tests can poke the snapshot directly.</summary>
|
||
internal AbCipSystemTagSource SystemTagSource => _systemTagSource;
|
||
|
||
private static DriverAttributeInfo ToAttributeInfo(AbCipTagDefinition tag) => new(
|
||
FullName: tag.Name,
|
||
DriverDataType: tag.DataType.ToDriverDataType(),
|
||
IsArray: false,
|
||
ArrayDim: null,
|
||
SecurityClass: (tag.Writable && !tag.SafetyTag)
|
||
? SecurityClassification.Operate
|
||
: SecurityClassification.ViewOnly,
|
||
IsHistorized: false,
|
||
IsAlarm: false,
|
||
WriteIdempotent: tag.WriteIdempotent,
|
||
Description: tag.Description);
|
||
|
||
/// <summary>Count of registered devices — exposed for diagnostics + tests.</summary>
|
||
internal int DeviceCount => _devices.Count;
|
||
|
||
/// <summary>Looked-up device state for the given host address. Tests + later-PR capabilities hit this.</summary>
|
||
internal DeviceState? GetDeviceState(string hostAddress) =>
|
||
_devices.TryGetValue(hostAddress, out var s) ? s : null;
|
||
|
||
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
|
||
|
||
public async ValueTask DisposeAsync()
|
||
{
|
||
await ShutdownAsync(CancellationToken.None).ConfigureAwait(false);
|
||
_discoverySemaphore.Dispose();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Per-device runtime state. Holds the parsed host address, family profile, and the
|
||
/// live <see cref="PlcTagHandle"/> cache keyed by tag path. PRs 3–8 populate + consume
|
||
/// this dict via libplctag.
|
||
/// </summary>
|
||
internal sealed class DeviceState(
|
||
AbCipHostAddress parsedAddress,
|
||
AbCipDeviceOptions options,
|
||
AbCipPlcFamilyProfile profile,
|
||
AddressingMode resolvedAddressingMode,
|
||
ReadStrategy resolvedReadStrategy)
|
||
{
|
||
public AbCipHostAddress ParsedAddress { get; } = parsedAddress;
|
||
public AbCipDeviceOptions Options { get; } = options;
|
||
public AbCipPlcFamilyProfile Profile { get; } = profile;
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.1 — effective CIP connection size for this device. Per-device
|
||
/// <see cref="AbCipDeviceOptions.ConnectionSize"/> override wins; otherwise the
|
||
/// family profile's <see cref="AbCipPlcFamilyProfile.DefaultConnectionSize"/>
|
||
/// (4002 / 504 / 488 depending on family). Threaded through every
|
||
/// <see cref="AbCipTagCreateParams"/> the driver builds so libplctag receives a
|
||
/// consistent buffer-size hint across read / write / probe / discovery handles.
|
||
/// </summary>
|
||
public int ConnectionSize { get; } = options.ConnectionSize ?? profile.DefaultConnectionSize;
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.2 — concrete addressing mode in effect for this device. Always
|
||
/// <see cref="AbCip.AddressingMode.Symbolic"/> or <see cref="AbCip.AddressingMode.Logical"/>
|
||
/// after <see cref="AbCipDriver.ResolveAddressingMode"/> has run; <c>Auto</c> +
|
||
/// unsupported-family fall-back collapse to Symbolic at config time so the
|
||
/// read/write hot paths can branch on a single value.
|
||
/// </summary>
|
||
public AddressingMode AddressingMode { get; } = resolvedAddressingMode;
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.3 — resolved read strategy for this device. <see cref="AbCip.ReadStrategy.WholeUdt"/>
|
||
/// or <see cref="AbCip.ReadStrategy.MultiPacket"/> mean "always pick this for every UDT
|
||
/// batch on this device." <see cref="AbCip.ReadStrategy.Auto"/> means "let the planner
|
||
/// pick per-batch using <see cref="AbCipDeviceOptions.MultiPacketSparsityThreshold"/>."
|
||
/// User-forced MultiPacket against a non-packing family (Micro800 et al) was already
|
||
/// collapsed to WholeUdt at <see cref="AbCipDriver.ResolveReadStrategy"/> time, so the
|
||
/// read hot path can branch on this single value without re-checking family compat.
|
||
/// </summary>
|
||
public ReadStrategy ReadStrategy { get; } = resolvedReadStrategy;
|
||
|
||
/// <summary>PR abcip-3.3 — count of UDT groups dispatched through the WholeUdt path on
|
||
/// this device. Surfaced for tests + a future driver-diagnostics RPC.</summary>
|
||
public int WholeUdtGroupsExecuted;
|
||
|
||
/// <summary>PR abcip-3.3 — count of UDT groups dispatched through the MultiPacket path
|
||
/// on this device. Surfaced for tests + a future driver-diagnostics RPC.</summary>
|
||
public int MultiPacketGroupsExecuted;
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.2 — name → Symbol Object instance ID map populated by the one-time
|
||
/// <c>@tags</c> walk that fires on the first read on a Logical-mode device. Empty
|
||
/// for Symbolic-mode devices + before the walk completes; consulted by
|
||
/// <see cref="AbCipDriver.EnsureTagRuntimeAsync"/> when materialising the per-tag
|
||
/// runtime so libplctag receives the resolved instance ID directly. Case-insensitive
|
||
/// because Logix tag names are case-insensitive at the controller.
|
||
/// </summary>
|
||
public Dictionary<string, uint> LogicalInstanceMap { get; } =
|
||
new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
/// <summary>
|
||
/// PR abcip-3.2 — guarded inside <see cref="AbCipDriver.EnsureLogicalMappingsAsync"/>
|
||
/// so the symbol-table walk fires exactly once per device. Setting this to
|
||
/// <c>true</c> means "walk attempted" — the walk's success / failure is captured by
|
||
/// the contents of <see cref="LogicalInstanceMap"/>; an empty map after the flag
|
||
/// flips means the walk yielded nothing and subsequent reads keep falling back to
|
||
/// Symbolic addressing on the wire.
|
||
/// </summary>
|
||
public bool LogicalWalkComplete { get; set; }
|
||
|
||
/// <summary>Serialises concurrent first-read symbol-walks against this device.</summary>
|
||
public SemaphoreSlim LogicalWalkLock { get; } = new(1, 1);
|
||
|
||
public object ProbeLock { get; } = new();
|
||
public HostState HostState { get; set; } = HostState.Unknown;
|
||
public DateTime HostStateChangedUtc { get; set; } = DateTime.UtcNow;
|
||
public CancellationTokenSource? ProbeCts { get; set; }
|
||
public bool ProbeInitialized { get; set; }
|
||
|
||
/// <summary>
|
||
/// PR abcip-4.3 — wall-clock duration of the most recent <see cref="AbCipDriver.ReadAsync"/>
|
||
/// iteration that touched any tag on this device, in milliseconds. Surfaces as
|
||
/// <c>_System/_LastScanTimeMs</c>; <c>0.0</c> until the first read completes so an
|
||
/// unread device shows a stable zero rather than a stale value.
|
||
/// </summary>
|
||
public double LastScanTimeMs;
|
||
|
||
public Dictionary<string, PlcTagHandle> TagHandles { get; } =
|
||
new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
/// <summary>
|
||
/// Per-tag runtime handles owned by this device. One entry per configured tag is
|
||
/// created lazily on first read (see <see cref="AbCipDriver.EnsureTagRuntimeAsync"/>).
|
||
/// </summary>
|
||
public Dictionary<string, IAbCipTagRuntime> Runtimes { get; } =
|
||
new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
/// <summary>
|
||
/// Parent-DINT runtimes created on-demand by <see cref="AbCipDriver.EnsureParentRuntimeAsync"/>
|
||
/// for BOOL-within-DINT RMW writes. Separate from <see cref="Runtimes"/> because a
|
||
/// bit-selector tag name ("Motor.Flags.3") needs a distinct handle from the DINT
|
||
/// parent ("Motor.Flags") used to do the read + write.
|
||
/// </summary>
|
||
public Dictionary<string, IAbCipTagRuntime> ParentRuntimes { get; } =
|
||
new(StringComparer.OrdinalIgnoreCase);
|
||
|
||
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, SemaphoreSlim> _rmwLocks = new();
|
||
|
||
public SemaphoreSlim GetRmwLock(string parentTagName) =>
|
||
_rmwLocks.GetOrAdd(parentTagName, _ => new SemaphoreSlim(1, 1));
|
||
|
||
public void DisposeHandles()
|
||
{
|
||
foreach (var h in TagHandles.Values) h.Dispose();
|
||
TagHandles.Clear();
|
||
foreach (var r in Runtimes.Values) r.Dispose();
|
||
Runtimes.Clear();
|
||
foreach (var r in ParentRuntimes.Values) r.Dispose();
|
||
ParentRuntimes.Clear();
|
||
}
|
||
}
|
||
}
|