diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Contracts/AbCipDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Contracts/AbCipDriverOptions.cs index 24c9a74b..4e730422 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Contracts/AbCipDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Contracts/AbCipDriverOptions.cs @@ -82,6 +82,18 @@ public sealed class AbCipDriverOptions /// public bool EnableDeclarationOnlyUdtGrouping { get; init; } + /// + /// Upper bound on how many planned read work items (whole-UDT groups + per-tag fallbacks) + /// one ReadAsync batch may have in flight against the controller at once. Bounds + /// concurrency ACROSS tags only — the per-tag runtime lock (05/STAB-4) still serialises + /// Read → GetStatus → Decode against any one cached libplctag handle, so the same tag + /// requested twice in a batch never overlaps itself. Raising it lets libplctag overlap + /// device round trips (the whole point: a strictly sequential batch pays one full round + /// trip per tag); lowering it to 1 restores strictly sequential reads. Values below + /// 1 are clamped to 1 at the use site. + /// + public int MaxConcurrentReads { get; init; } = 8; + /// /// Timeout for the AdminUI Test Connect probe, in seconds. The AdminUI clamps to a /// 60s server-side maximum; this default is what the form pre-fills for new instances. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs index 25e2625e..b07b76c9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -590,22 +590,125 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, var plan = AbCipUdtReadPlanner.Build( fullReferences, _tagsByRawPath, _options.EnableDeclarationOnlyUdtGrouping); - foreach (var group in plan.Groups) - await ReadGroupAsync(group, results, now, cancellationToken).ConfigureAwait(false); + var groups = plan.Groups; + var fallbacks = plan.Fallbacks; + var itemCount = groups.Count + fallbacks.Count; + if (itemCount == 0) return results; - foreach (var fb in plan.Fallbacks) - await ReadSingleAsync(fb, fullReferences[fb.OriginalIndex], results, now, cancellationToken).ConfigureAwait(false); + // Bounded-parallel window ACROSS work items (groups ∪ fallbacks). Each item still takes its + // OWN per-tag runtime lock inside ReadGroupAsync / ReadSingleAsync, so one libplctag handle + // is never touched concurrently (05/STAB-4) — the window only lets DIFFERENT tags' device + // round trips overlap, which a strictly sequential await chain made impossible. Results are + // index-addressed, so completion order is irrelevant. Per-item health verdicts are COLLECTED + // and written once after the batch: an inline write to the shared _health field would race. + var maxConcurrent = Math.Max(1, _options.MaxConcurrentReads); + var outcomes = new ReadOutcome[itemCount]; + using var window = new SemaphoreSlim(maxConcurrent, maxConcurrent); + var tasks = new Task[itemCount]; + for (var i = 0; i < itemCount; i++) + tasks[i] = RunItemAsync(i); + // Task.WhenAll completes only once EVERY item has finished (it does not short-circuit on the + // first fault), so disposing the window on the way out can never pull the semaphore out from + // under an item still waiting on it. + await Task.WhenAll(tasks).ConfigureAwait(false); + + ApplyBatchHealth(outcomes, now); return results; + + async Task RunItemAsync(int slot) + { + // Acquire OUTSIDE the try: a wait that throws never acquired, so it must not release. + await window.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + if (slot < groups.Count) + { + outcomes[slot] = await ReadGroupAsync(groups[slot], results, now, cancellationToken) + .ConfigureAwait(false); + } + else + { + var fb = fallbacks[slot - groups.Count]; + outcomes[slot] = await ReadSingleAsync( + fb, fullReferences[fb.OriginalIndex], results, now, cancellationToken) + .ConfigureAwait(false); + } + } + finally + { + window.Release(); + } + } } - private async Task ReadSingleAsync( + /// + /// One read work item's health verdict. Returned by / + /// instead of written to the shared health field inline, because + /// under the bounded-parallel read window those writes would race (last-writer-wins across + /// threads could report Healthy for a batch that mostly failed, or vice versa). + /// A null means "no health signal" — a config-level miss such as an + /// unknown reference or unknown device, which never touched the wire and never touched health. + /// + private readonly record struct ReadOutcome(DriverState? State, string? Error) + { + /// The item produced no health signal at all. + public static ReadOutcome None => new(null, null); + + /// The item read successfully. + public static ReadOutcome Success => new(DriverState.Healthy, null); + + /// The item failed on the wire. + /// The failure detail surfaced as LastError. + public static ReadOutcome Degraded(string message) => new(DriverState.Degraded, message); + } + + private static int HealthRank(DriverState state) => state switch + { + DriverState.Faulted => 2, + DriverState.Degraded => 1, + _ => 0, + }; + + /// + /// Write the batch's health ONCE, after every work item has finished: worst outcome wins + /// (Faulted > Degraded > Healthy) and LastSuccessfulRead advances if ANY item + /// succeeded — the contract the comms-loss watchdog reads (a success advances it; a batch + /// where everything failed freezes it, which is what trips the watchdog). + /// + private void ApplyBatchHealth(ReadOnlySpan outcomes, DateTime now) + { + var anySuccess = false; + DriverState? worst = null; + string? worstError = null; + foreach (var outcome in outcomes) + { + if (outcome.State is not { } state) continue; + if (state == DriverState.Healthy) anySuccess = true; + if (worst is null || HealthRank(state) > HealthRank(worst.Value)) + { + worst = state; + worstError = outcome.Error; + } + } + + // Nothing reached the wire (every reference was unknown / unroutable) — leave health alone, + // exactly as the per-item writes did before. + if (worst is not { } worstState) return; + + _health = new DriverHealth( + worstState, + anySuccess ? now : _health.LastSuccessfulRead, + worstState == DriverState.Healthy ? null : worstError); + } + + private async Task ReadSingleAsync( AbCipUdtReadFallback fb, string reference, DataValueSnapshot[] results, DateTime now, CancellationToken ct) { if (!_resolver.TryResolve(reference, out var def)) { results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNodeIdUnknown, null, now); - return; + return ReadOutcome.None; } // A Structure tag whose Members are declared is a container — // its bare name is readable via the whole-UDT grouping path (ReadGroupAsync), not the @@ -614,12 +717,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, if (def.DataType == AbCipDataType.Structure && def.Members is { Count: > 0 }) { results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNotSupported, null, now); - return; + return ReadOutcome.None; } if (!_devicesByRoutingKey.TryGetValue(def.DeviceHostAddress, out var device)) { results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadNodeIdUnknown, null, now); - return; + return ReadOutcome.None; } try @@ -644,13 +747,11 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, EvictRuntime(device, def.Name); results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.MapLibplctagStatus(status), null, now); - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, - $"libplctag status {status} reading {reference}"); _logger.LogWarning( "AbCip read returned non-zero libplctag status {LibplctagStatus} for tag {Tag} on device {Device}; " + "evicting cached runtime so next call re-creates it", status, reference, def.DeviceHostAddress); - return; + return ReadOutcome.Degraded($"libplctag status {status} reading {reference}"); } var tagPath = AbCipTagPath.TryParse(def.TagPath); @@ -662,7 +763,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ? runtime.DecodeArray(def.DataType, Math.Max(1, def.ElementCount)) : runtime.DecodeValue(def.DataType, bitIndex); results[fb.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now); - _health = new DriverHealth(DriverState.Healthy, now, null); + return ReadOutcome.Success; } finally { @@ -679,10 +780,10 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, EvictRuntime(device, def.Name); results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadCommunicationError, null, now); - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); _logger.LogWarning(ex, "AbCip read transport exception for tag {Tag} on device {Device}", reference, def.DeviceHostAddress); + return ReadOutcome.Degraded(ex.Message); } } @@ -693,7 +794,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// the mapped fault across every grouped member only — sibling groups + the /// per-tag fallback list are unaffected. /// - private async Task ReadGroupAsync( + private async Task ReadGroupAsync( AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, CancellationToken ct) { var parent = group.ParentDefinition; @@ -701,7 +802,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, if (!_devicesByRoutingKey.TryGetValue(parent.DeviceHostAddress, out var device)) { StampGroupStatus(group, results, now, AbCipStatusMapper.BadNodeIdUnknown); - return; + return ReadOutcome.None; } try @@ -722,13 +823,11 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, EvictRuntime(device, parent.Name); var mapped = AbCipStatusMapper.MapLibplctagStatus(status); StampGroupStatus(group, results, now, mapped); - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, - $"libplctag status {status} reading UDT {group.ParentName}"); _logger.LogWarning( "AbCip whole-UDT read returned non-zero libplctag status {LibplctagStatus} for parent {Parent} " + "on device {Device}; {MemberCount} member values stamped with mapped status", status, group.ParentName, parent.DeviceHostAddress, group.Members.Count); - return; + return ReadOutcome.Degraded($"libplctag status {status} reading UDT {group.ParentName}"); } foreach (var member in group.Members) @@ -736,7 +835,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, 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); + return ReadOutcome.Success; } finally { @@ -751,10 +850,10 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, { EvictRuntime(device, parent.Name); StampGroupStatus(group, results, now, AbCipStatusMapper.BadCommunicationError); - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); _logger.LogWarning(ex, "AbCip whole-UDT read transport exception for parent {Parent} on device {Device}", group.ParentName, parent.DeviceHostAddress); + return ReadOutcome.Degraded(ex.Message); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriverFactoryExtensions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriverFactoryExtensions.cs index da0fd469..6d8e176c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriverFactoryExtensions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriverFactoryExtensions.cs @@ -85,6 +85,7 @@ public static class AbCipDriverFactoryExtensions EnableAlarmProjection = dto.EnableAlarmProjection ?? false, AlarmPollInterval = TimeSpan.FromMilliseconds(dto.AlarmPollIntervalMs ?? 1_000), EnableDeclarationOnlyUdtGrouping = dto.EnableDeclarationOnlyUdtGrouping ?? false, + MaxConcurrentReads = dto.MaxConcurrentReads ?? 8, }; } @@ -147,6 +148,13 @@ public static class AbCipDriverFactoryExtensions /// public bool? EnableDeclarationOnlyUdtGrouping { get; init; } + /// + /// Gets or sets the maximum number of read work items one ReadAsync batch may have in + /// flight against the controller at once. Bounds concurrency across tags only; the per-tag + /// runtime lock still serialises operations on any one handle. Defaults to 8; 1 is sequential. + /// + public int? MaxConcurrentReads { get; init; } + /// /// Gets or sets the alarm poll interval in milliseconds. /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs new file mode 100644 index 00000000..71630e1f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs @@ -0,0 +1,201 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.AbCip; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests; + +/// +/// Task 14 — ReadAsync runs its planned groups + per-tag fallbacks inside a bounded +/// concurrency window () instead of strictly +/// one device round-trip at a time, so libplctag can overlap requests. The window is ACROSS +/// tags only: the per-tag runtime lock (05/STAB-4) still serialises Read → GetStatus → Decode +/// against any one cached handle. Health is aggregated once after the batch (worst outcome +/// wins, LastSuccessfulRead advances if any tag succeeded) because per-item writes to +/// the shared health field would race under the window. +/// +[Trait("Category", "Unit")] +public sealed class AbCipParallelReadTests +{ + private const string DeviceHost = "ab://10.0.0.5/1,0"; + + /// + /// Holds every in-flight fake read until either reads overlap + /// (proving the window opened) or a short grace period elapses (so a window of one, which + /// can never reach the target, still completes instead of deadlocking). + /// + private sealed class ReadGate(int target) + { + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _inFlight; + + /// Gets the highest number of fake reads observed in flight simultaneously. + public int MaxConcurrent { get; private set; } + + /// Records one in-flight read and holds it open until the gate releases. + /// The read's cancellation token. + public async Task HoldAsync(CancellationToken cancellationToken) + { + var n = Interlocked.Increment(ref _inFlight); + lock (this) + { + if (n > MaxConcurrent) MaxConcurrent = n; + } + try + { + if (n >= target) _release.TrySetResult(); + try + { + await _release.Task.WaitAsync(TimeSpan.FromMilliseconds(150), cancellationToken) + .ConfigureAwait(false); + } + catch (TimeoutException) + { + // The window never reached the target — the sequential control case. + } + } + finally + { + Interlocked.Decrement(ref _inFlight); + } + } + } + + /// A fake whose read blocks on the shared gate and records its OWN overlap. + private sealed class GatedFake(AbCipTagCreateParams p, ReadGate gate) : FakeAbCipTag(p) + { + private int _own; + + /// Gets the highest number of concurrent reads seen on THIS handle. + public int MaxOwnConcurrent { get; private set; } + + /// + public override async Task ReadAsync(CancellationToken cancellationToken) + { + var mine = Interlocked.Increment(ref _own); + lock (this) + { + if (mine > MaxOwnConcurrent) MaxOwnConcurrent = mine; + } + try + { + await gate.HoldAsync(cancellationToken).ConfigureAwait(false); + await base.ReadAsync(cancellationToken).ConfigureAwait(false); + } + finally + { + Interlocked.Decrement(ref _own); + } + } + } + + private static AbCipDriver NewDriver( + IAbCipTagFactory factory, int maxConcurrentReads, params AbCipTagDefinition[] tags) + { + var opts = new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions(DeviceHost)], + RawTags = AbCipRawTags.From(tags), + Probe = new AbCipProbeOptions { Enabled = false }, + MaxConcurrentReads = maxConcurrentReads, + }; + return new AbCipDriver(opts, "drv-1", factory); + } + + private static AbCipTagDefinition Tag(string name) => + new(name, DeviceHost, $"Motor1.{name}", AbCipDataType.DInt); + + /// Four distinct tags under a window of eight overlap on the wire. + [Fact] + public async Task Reads_overlap_up_to_the_window() + { + var gate = new ReadGate(target: 2); + var factory = new FakeAbCipTagFactory { Customise = p => new GatedFake(p, gate) { Value = 1 } }; + var drv = NewDriver(factory, maxConcurrentReads: 8, + Tag("A"), Tag("B"), Tag("C"), Tag("D")); + await drv.InitializeAsync("{}", CancellationToken.None); + + var snapshots = await drv.ReadAsync(["A", "B", "C", "D"], CancellationToken.None); + + gate.MaxConcurrent.ShouldBeGreaterThanOrEqualTo(2, + "a window of 8 must let distinct tags overlap on the device"); + snapshots.Count.ShouldBe(4); + snapshots.ShouldAllBe(s => s.StatusCode == AbCipStatusMapper.Good); + } + + /// + /// Falsification control for — the same + /// fixture with a window of one must observe no overlap at all, i.e. today's behaviour. + /// + [Fact] + public async Task A_window_of_one_is_sequential() + { + var gate = new ReadGate(target: 2); + var factory = new FakeAbCipTagFactory { Customise = p => new GatedFake(p, gate) { Value = 1 } }; + var drv = NewDriver(factory, maxConcurrentReads: 1, + Tag("A"), Tag("B"), Tag("C"), Tag("D")); + await drv.InitializeAsync("{}", CancellationToken.None); + + var snapshots = await drv.ReadAsync(["A", "B", "C", "D"], CancellationToken.None); + + gate.MaxConcurrent.ShouldBe(1, "MaxConcurrentReads = 1 must be observably sequential"); + snapshots.Count.ShouldBe(4); + snapshots.ShouldAllBe(s => s.StatusCode == AbCipStatusMapper.Good); + } + + /// + /// The window parallelises ACROSS tags, never within one — the same reference twice in a + /// batch still serialises on the per-tag runtime lock (05/STAB-4). + /// + [Fact] + public async Task Same_tag_requested_twice_is_still_serialised() + { + var gate = new ReadGate(target: 2); + GatedFake? shared = null; + var factory = new FakeAbCipTagFactory + { + Customise = p => shared = new GatedFake(p, gate) { Value = 5 }, + }; + var drv = NewDriver(factory, maxConcurrentReads: 8, Tag("A")); + await drv.InitializeAsync("{}", CancellationToken.None); + + var snapshots = await drv.ReadAsync(["A", "A"], CancellationToken.None); + + shared.ShouldNotBeNull(); + shared!.MaxOwnConcurrent.ShouldBe(1, + "operations on one shared libplctag Tag handle must never overlap"); + snapshots.Count.ShouldBe(2); + snapshots.ShouldAllBe(s => s.StatusCode == AbCipStatusMapper.Good); + } + + /// + /// Health is written once after the batch: the worst per-item outcome wins, and + /// LastSuccessfulRead still advances because at least one tag read succeeded — the + /// contract the comms-loss watchdog reads. + /// + [Fact] + public async Task Health_after_a_mixed_batch_is_worst_outcome_with_advanced_LastSuccessfulRead() + { + var factory = new FakeAbCipTagFactory + { + Customise = p => p.TagName.Contains("Bad", StringComparison.Ordinal) + ? new FakeAbCipTag(p) { ThrowOnRead = true } + : new FakeAbCipTag(p) { Value = 3 }, + }; + var drv = NewDriver(factory, maxConcurrentReads: 8, Tag("Good"), Tag("Bad")); + await drv.InitializeAsync("{}", CancellationToken.None); + var beforeRead = drv.GetHealth().LastSuccessfulRead; + beforeRead.ShouldNotBeNull(); + await Task.Delay(30, TestContext.Current.CancellationToken); + + var snapshots = await drv.ReadAsync(["Good", "Bad"], CancellationToken.None); + + snapshots[0].StatusCode.ShouldBe(AbCipStatusMapper.Good); + snapshots[1].StatusCode.ShouldBe(AbCipStatusMapper.BadCommunicationError); + var health = drv.GetHealth(); + health.State.ShouldBe(DriverState.Degraded, "the worst per-item outcome wins"); + health.LastSuccessfulRead.ShouldNotBeNull(); + health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(beforeRead!.Value, + "a successful tag in the batch must still advance LastSuccessfulRead"); + } +}