perf(abcip): bounded-parallel tag reads (MaxConcurrentReads, default 8)

2026-08-16 04:02:40 -04:00
parent 5df793e79a
commit d6a4fbd648
4 changed files with 340 additions and 20 deletions
@@ -82,6 +82,18 @@ public sealed class AbCipDriverOptions
    /// </summary>
    public bool EnableDeclarationOnlyUdtGrouping { get; init; }

+    /// <summary>
+    ///     Upper bound on how many planned read work items (whole-UDT groups + per-tag fallbacks)
+    ///     one <c>ReadAsync</c> 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 <c>1</c> restores strictly sequential reads. Values below
+    ///     <c>1</c> are clamped to <c>1</c> at the use site.
+    /// </summary>
+    public int MaxConcurrentReads { get; init; } = 8;
+
    /// <summary>
    ///     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.
@@ -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(
+    /// <summary>
+    ///     One read work item's health verdict. Returned by <see cref="ReadSingleAsync"/> /
+    ///     <see cref="ReadGroupAsync"/> 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 <c>null</c> <see cref="State"/> 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.
+    /// </summary>
+    private readonly record struct ReadOutcome(DriverState? State, string? Error)
+    {
+        /// <summary>The item produced no health signal at all.</summary>
+        public static ReadOutcome None => new(null, null);
+
+        /// <summary>The item read successfully.</summary>
+        public static ReadOutcome Success => new(DriverState.Healthy, null);
+
+        /// <summary>The item failed on the wire.</summary>
+        /// <param name="message">The failure detail surfaced as <c>LastError</c>.</param>
+        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,
+    };
+
+    /// <summary>
+    ///     Write the batch's health ONCE, after every work item has finished: worst outcome wins
+    ///     (Faulted &gt; Degraded &gt; Healthy) and <c>LastSuccessfulRead</c> 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).
+    /// </summary>
+    private void ApplyBatchHealth(ReadOnlySpan<ReadOutcome> 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<ReadOutcome> 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.
    /// </summary>
-    private async Task ReadGroupAsync(
+    private async Task<ReadOutcome> 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);
        }
    }

@@ -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
        /// </summary>
        public bool? EnableDeclarationOnlyUdtGrouping { get; init; }

+        /// <summary>
+        /// 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.
+        /// </summary>
+        public int? MaxConcurrentReads { get; init; }
+
        /// <summary>
        /// Gets or sets the alarm poll interval in milliseconds.
        /// </summary>
@@ -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;
+
+/// <summary>
+///     Task 14 — <c>ReadAsync</c> runs its planned groups + per-tag fallbacks inside a bounded
+///     concurrency window (<see cref="AbCipDriverOptions.MaxConcurrentReads"/>) 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, <c>LastSuccessfulRead</c> advances if any tag succeeded) because per-item writes to
+///     the shared health field would race under the window.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class AbCipParallelReadTests
+{
+    private const string DeviceHost = "ab://10.0.0.5/1,0";
+
+    /// <summary>
+    ///     Holds every in-flight fake read until either <paramref name="target"/> 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).
+    /// </summary>
+    private sealed class ReadGate(int target)
+    {
+        private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int _inFlight;
+
+        /// <summary>Gets the highest number of fake reads observed in flight simultaneously.</summary>
+        public int MaxConcurrent { get; private set; }
+
+        /// <summary>Records one in-flight read and holds it open until the gate releases.</summary>
+        /// <param name="cancellationToken">The read's cancellation token.</param>
+        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);
+            }
+        }
+    }
+
+    /// <summary>A fake whose read blocks on the shared gate and records its OWN overlap.</summary>
+    private sealed class GatedFake(AbCipTagCreateParams p, ReadGate gate) : FakeAbCipTag(p)
+    {
+        private int _own;
+
+        /// <summary>Gets the highest number of concurrent reads seen on THIS handle.</summary>
+        public int MaxOwnConcurrent { get; private set; }
+
+        /// <inheritdoc />
+        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);
+
+    /// <summary>Four distinct tags under a window of eight overlap on the wire.</summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     Falsification control for <see cref="Reads_overlap_up_to_the_window"/> — the same
+    ///     fixture with a window of one must observe no overlap at all, i.e. today's behaviour.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     Health is written once after the batch: the worst per-item outcome wins, and
+    ///     <c>LastSuccessfulRead</c> still advances because at least one tag read succeeded — the
+    ///     contract the comms-loss watchdog reads.
+    /// </summary>
+    [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");
+    }
+}