From 4e9bd9d9665f10866063c1df24cc1f14132dddad Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:19:38 -0400 Subject: [PATCH] perf(abcip): native read window >= 64; cold waiters hold no window slot P8: on the native transport the bounded-parallel read window is also the cap on services per Multiple Service Packet (the device's shared scheduler packs whatever is queued, within the connection's request/reply byte budget), so the default 8 cost a 1000-tag controller ~125 round trips per poll instead of ~10. The effective window (AbCipDriver.ReadWindow) is now max(64, option) when the native tag factory serves reads; libplctag and injected factories keep the option exactly. Option doc + AdminUI help text updated (Razor: Wave 4 live check). P7 addendum: a work item now takes its window slot only after its runtime exists. A cold create queues on its device's single-flight connect gate, and holding a slot there let one dead device's backoff-expiry attempt occupy the whole window for a full Timeout, stalling every other device's reads. Refs #737 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../AbCipDriverOptions.cs | 6 + .../AbCipDriver.cs | 271 ++++++++++-------- .../Drivers/Forms/AbCipDriverForm.razor | 2 +- .../AbCipReadWindowTests.cs | 96 +++++++ 4 files changed, 254 insertions(+), 121 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReadWindowTests.cs 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 9fc101c3..9656a70e 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 @@ -117,6 +117,12 @@ public sealed class AbCipDriverOptions /// 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. + /// + /// On the transport the effective window is at least 64: there the + /// window caps how many services one Multiple Service Packet coalesces (the device's shared scheduler packs + /// what is queued, bounded by the connection's request/reply byte budget), so a small value only multiplies + /// round trips. A value above 64 is honoured; a value below it is raised. + /// /// public int MaxConcurrentReads { get; init; } = 8; 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 e4b6f9bb..b48a619c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -895,7 +895,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // 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 maxConcurrent = ReadWindow; var outcomes = new ReadOutcome[itemCount]; using var window = new SemaphoreSlim(maxConcurrent, maxConcurrent); var tasks = new Task[itemCount]; @@ -910,32 +910,44 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ApplyBatchHealth(outcomes, now); return results; + // The window slot is taken INSIDE each item, after its runtime exists (P7): a cold create queues on its + // device's single-flight connect gate, and holding a slot while queued there let one dead device's + // backoff-expiry attempt occupy the whole window for a full Timeout, stalling every other device's reads. 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) { - 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); - } + outcomes[slot] = await ReadGroupAsync(groups[slot], results, now, window, cancellationToken) + .ConfigureAwait(false); } - finally + else { - window.Release(); + var fb = fallbacks[slot - groups.Count]; + outcomes[slot] = await ReadSingleAsync( + fb, fullReferences[fb.OriginalIndex], results, now, window, cancellationToken) + .ConfigureAwait(false); } } } + /// + /// Floor on the read window when the native transport serves this driver (P8). Its runtimes share one + /// request scheduler per device that packs whatever is queued when the pump wakes into one Multiple Service + /// Packet — bounded by the connection's request AND reply byte budget, so a wide window cannot over-pack. The + /// window is therefore the cap on services per round trip: at 8, a 1000-tag controller paid ~125 round trips + /// per poll instead of ~10. + /// + internal const int NativeMinReadWindow = 64; + + /// + /// The effective bounded-parallel read window: (clamped + /// to ≥ 1), raised to at least when the native tag factory serves the reads. + /// libplctag overlaps per-handle round trips only, so it (and any injected factory) keeps the option as is. + /// + internal int ReadWindow => _tagFactory is NativeAbCipTagFactory + ? Math.Max(NativeMinReadWindow, _options.MaxConcurrentReads) + : Math.Max(1, _options.MaxConcurrentReads); + /// /// One read work item's health verdict. Returned by / /// instead of written to the shared health field inline, because @@ -1015,7 +1027,8 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } private async Task ReadSingleAsync( - AbCipUdtReadFallback fb, string reference, DataValueSnapshot[] results, DateTime now, CancellationToken ct) + AbCipUdtReadFallback fb, string reference, DataValueSnapshot[] results, DateTime now, + SemaphoreSlim window, CancellationToken ct) { if (!_resolver.TryResolve(reference, out var def)) { @@ -1041,82 +1054,91 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, { var runtime = await EnsureTagRuntimeAsync(device, def, ct).ConfigureAwait(false); + // Window slot only now that the runtime exists (see ReadAsync); a wait that throws never acquired. + await window.WaitAsync(ct).ConfigureAwait(false); // Serialise Read → GetStatus → Decode against the shared runtime — the cached libplctag // Tag handle is not concurrency-safe and GetStatus returns the LAST op's status. var opLock = device.GetRuntimeLock(def.Name); - await opLock.WaitAsync(ct).ConfigureAwait(false); try { - await runtime.ReadAsync(ct).ConfigureAwait(false); + await opLock.WaitAsync(ct).ConfigureAwait(false); + try + { + await runtime.ReadAsync(ct).ConfigureAwait(false); - var status = runtime.GetStatus(); - if (status == (int)AbStatus.TypeMismatch) - { - // Final review I4: the controller ANSWERED with a type that disagrees with the authored - // one — a deterministic config error, not a comms fault. Keep the handle (evicting would - // re-create it — and, for a sole lease holder, re-Forward-Open — on every poll), leave - // health alone, warn once per tag. - results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now); - WarnTypeMismatchOnce(def.Name, - "AbCip read of {Tag} on device {Device}: the controller's type does not match the authored " + - "{DataType}; reporting BadTypeMismatch until the tag's dataType is corrected (logged once per tag)", - reference, def.DeviceHostAddress, def.DataType); - return ReadOutcome.Refused; - } - if (AbCipRefusals.IsDeviceRefusal(status)) - { - // The controller answered with a deterministic refusal: per-tag Bad, keep the handle. - results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.MapReadStatus((AbStatus)status), null, now); - WarnRefusalOnce(def.Name, "read", reference, def.DeviceHostAddress, ((AbStatus)status).ToString()); - return ReadOutcome.Refused; - } - if (status != 0) - { - // Evict the stale handle so the next call re-creates it. - // A non-zero status can mean the controller dropped the connection or the tag - // handle became permanently invalid (e.g. after a PLC download). Evicting - // mirrors the probe loop's recreate-on-failure behaviour. - EvictRuntime(device, def.Name); - results[fb.OriginalIndex] = new DataValueSnapshot(null, - AbCipStatusMapper.MapStatus(status), null, now); - _logger.LogWarning( - "AbCip read returned non-zero transport status {Status} for tag {Tag} on device {Device}; " + - "evicting cached runtime so next call re-creates it", - status, reference, def.DeviceHostAddress); - return ReadOutcome.Degraded($"transport status {status} reading {reference}"); - } + var status = runtime.GetStatus(); + if (status == (int)AbStatus.TypeMismatch) + { + // Final review I4: the controller ANSWERED with a type that disagrees with the authored + // one — a deterministic config error, not a comms fault. Keep the handle (evicting would + // re-create it — and, for a sole lease holder, re-Forward-Open — on every poll), leave + // health alone, warn once per tag. + results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now); + WarnTypeMismatchOnce(def.Name, + "AbCip read of {Tag} on device {Device}: the controller's type does not match the authored " + + "{DataType}; reporting BadTypeMismatch until the tag's dataType is corrected (logged once per tag)", + reference, def.DeviceHostAddress, def.DataType); + return ReadOutcome.Refused; + } + if (AbCipRefusals.IsDeviceRefusal(status)) + { + // The controller answered with a deterministic refusal: per-tag Bad, keep the handle. + results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.MapReadStatus((AbStatus)status), null, now); + WarnRefusalOnce(def.Name, "read", reference, def.DeviceHostAddress, ((AbStatus)status).ToString()); + return ReadOutcome.Refused; + } + if (status != 0) + { + // Evict the stale handle so the next call re-creates it. + // A non-zero status can mean the controller dropped the connection or the tag + // handle became permanently invalid (e.g. after a PLC download). Evicting + // mirrors the probe loop's recreate-on-failure behaviour. + EvictRuntime(device, def.Name); + results[fb.OriginalIndex] = new DataValueSnapshot(null, + AbCipStatusMapper.MapStatus(status), null, now); + _logger.LogWarning( + "AbCip read returned non-zero transport status {Status} for tag {Tag} on device {Device}; " + + "evicting cached runtime so next call re-creates it", + status, reference, def.DeviceHostAddress); + return ReadOutcome.Degraded($"transport status {status} reading {reference}"); + } - var tagPath = AbCipTagPath.TryParse(def.TagPath); - var bitIndex = tagPath?.BitIndex; + var tagPath = AbCipTagPath.TryParse(def.TagPath); + var bitIndex = tagPath?.BitIndex; - // libplctag never checks the CIP type the controller returned; it decodes the bytes as - // the authored type. A scalar whose buffer is not that type's width is mis-typed config - // (seen live: a REAL authored as INT read Good -17346), so refuse it rather than publish - // garbage as Good. Same-width mismatches (DINT vs REAL) cannot be seen this way. - if (!IsArrayTag(def) && bitIndex is null - && ExpectedScalarWidth(def.DataType) is { } width - && runtime.GetSize() is var size && size >= 0 && size != width) - { - // Same class as the native TypeMismatch above (final review I4): deterministic, device - // answered — no Degraded, warn once per tag. - results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now); - WarnTypeMismatchOnce(def.Name, - "AbCip read of {Tag} on device {Device} returned {Size} bytes but its authored type {DataType} is {Width} bytes wide; check the tag's dataType (logged once per tag)", - reference, def.DeviceHostAddress, size, def.DataType, width); - return ReadOutcome.Refused; + // libplctag never checks the CIP type the controller returned; it decodes the bytes as + // the authored type. A scalar whose buffer is not that type's width is mis-typed config + // (seen live: a REAL authored as INT read Good -17346), so refuse it rather than publish + // garbage as Good. Same-width mismatches (DINT vs REAL) cannot be seen this way. + if (!IsArrayTag(def) && bitIndex is null + && ExpectedScalarWidth(def.DataType) is { } width + && runtime.GetSize() is var size && size >= 0 && size != width) + { + // Same class as the native TypeMismatch above (final review I4): deterministic, device + // answered — no Degraded, warn once per tag. + results[fb.OriginalIndex] = new DataValueSnapshot(null, AbCipStatusMapper.BadTypeMismatch, null, now); + WarnTypeMismatchOnce(def.Name, + "AbCip read of {Tag} on device {Device} returned {Size} bytes but its authored type {DataType} is {Width} bytes wide; check the tag's dataType (logged once per tag)", + reference, def.DeviceHostAddress, size, def.DataType, width); + return ReadOutcome.Refused; + } + // Review I-1 — an array tag (the EXPLICIT IsArray flag) decodes the whole buffer into an + // element-typed CLR array (int[]/float[]/bool[]/string[]…), INCLUDING a 1-element array + // (ElementCount 1). Scalar tags keep the single-value path. + var value = IsArrayTag(def) + ? runtime.DecodeArray(def.DataType, Math.Max(1, def.ElementCount)) + : runtime.DecodeValue(def.DataType, bitIndex); + results[fb.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now); + return ReadOutcome.Success; + } + finally + { + opLock.Release(); } - // Review I-1 — an array tag (the EXPLICIT IsArray flag) decodes the whole buffer into an - // element-typed CLR array (int[]/float[]/bool[]/string[]…), INCLUDING a 1-element array - // (ElementCount 1). Scalar tags keep the single-value path. - var value = IsArrayTag(def) - ? runtime.DecodeArray(def.DataType, Math.Max(1, def.ElementCount)) - : runtime.DecodeValue(def.DataType, bitIndex); - results[fb.OriginalIndex] = new DataValueSnapshot(value, AbCipStatusMapper.Good, now, now); - return ReadOutcome.Success; } finally { - opLock.Release(); + window.Release(); } } catch (OperationCanceledException) @@ -1185,7 +1207,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// per-tag fallback list are unaffected. /// private async Task ReadGroupAsync( - AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, CancellationToken ct) + AbCipUdtReadGroup group, DataValueSnapshot[] results, DateTime now, SemaphoreSlim window, CancellationToken ct) { var parent = group.ParentDefinition; @@ -1199,53 +1221,62 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, { var runtime = await EnsureTagRuntimeAsync(device, parent, ct).ConfigureAwait(false); + // Window slot only now that the runtime exists (see ReadAsync); a wait that throws never acquired. + await window.WaitAsync(ct).ConfigureAwait(false); // Serialise the whole-UDT Read → GetStatus → per-member DecodeValueAt against the shared // parent runtime — the members decode from that one handle's buffer. var opLock = device.GetRuntimeLock(parent.Name); - await opLock.WaitAsync(ct).ConfigureAwait(false); try { - await runtime.ReadAsync(ct).ConfigureAwait(false); + await opLock.WaitAsync(ct).ConfigureAwait(false); + try + { + await runtime.ReadAsync(ct).ConfigureAwait(false); - var status = runtime.GetStatus(); - if (status == (int)AbStatus.TypeMismatch) - { - // Final review I4: deterministic — keep the handle, no health signal, warn once. - StampGroupStatus(group, results, now, AbCipStatusMapper.BadTypeMismatch); - WarnTypeMismatchOnce(parent.Name, - "AbCip whole-UDT read of {Parent} on device {Device}: the controller's type does not match " + - "the authored type; {MemberCount} members report BadTypeMismatch (logged once per tag)", - group.ParentName, parent.DeviceHostAddress, group.Members.Count); - return ReadOutcome.Refused; - } - if (AbCipRefusals.IsDeviceRefusal(status)) - { - StampGroupStatus(group, results, now, AbCipStatusMapper.MapReadStatus((AbStatus)status)); - WarnRefusalOnce(parent.Name, "whole-UDT read", group.ParentName, parent.DeviceHostAddress, ((AbStatus)status).ToString()); - return ReadOutcome.Refused; - } - if (status != 0) - { - EvictRuntime(device, parent.Name); - var mapped = AbCipStatusMapper.MapStatus(status); - StampGroupStatus(group, results, now, mapped); - _logger.LogWarning( - "AbCip whole-UDT read returned non-zero transport status {Status} for parent {Parent} " + - "on device {Device}; {MemberCount} member values stamped with mapped status", - status, group.ParentName, parent.DeviceHostAddress, group.Members.Count); - return ReadOutcome.Degraded($"transport status {status} reading UDT {group.ParentName}"); - } + var status = runtime.GetStatus(); + if (status == (int)AbStatus.TypeMismatch) + { + // Final review I4: deterministic — keep the handle, no health signal, warn once. + StampGroupStatus(group, results, now, AbCipStatusMapper.BadTypeMismatch); + WarnTypeMismatchOnce(parent.Name, + "AbCip whole-UDT read of {Parent} on device {Device}: the controller's type does not match " + + "the authored type; {MemberCount} members report BadTypeMismatch (logged once per tag)", + group.ParentName, parent.DeviceHostAddress, group.Members.Count); + return ReadOutcome.Refused; + } + if (AbCipRefusals.IsDeviceRefusal(status)) + { + StampGroupStatus(group, results, now, AbCipStatusMapper.MapReadStatus((AbStatus)status)); + WarnRefusalOnce(parent.Name, "whole-UDT read", group.ParentName, parent.DeviceHostAddress, ((AbStatus)status).ToString()); + return ReadOutcome.Refused; + } + if (status != 0) + { + EvictRuntime(device, parent.Name); + var mapped = AbCipStatusMapper.MapStatus(status); + StampGroupStatus(group, results, now, mapped); + _logger.LogWarning( + "AbCip whole-UDT read returned non-zero transport status {Status} for parent {Parent} " + + "on device {Device}; {MemberCount} member values stamped with mapped status", + status, group.ParentName, parent.DeviceHostAddress, group.Members.Count); + return ReadOutcome.Degraded($"transport status {status} reading UDT {group.ParentName}"); + } - 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); + 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); + } + return ReadOutcome.Success; + } + finally + { + opLock.Release(); } - return ReadOutcome.Success; } finally { - opLock.Release(); + window.Release(); } } catch (OperationCanceledException) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/AbCipDriverForm.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/AbCipDriverForm.razor index 49ecfd06..496a2bed 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/AbCipDriverForm.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/AbCipDriverForm.razor @@ -21,7 +21,7 @@
-
Read work items in flight at once. Default 8; 1 = sequential.
+
Read work items in flight at once. Default 8; 1 = sequential (libplctag). The native transport always uses at least 64 — it packs concurrent reads into one request.
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReadWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReadWindowTests.cs new file mode 100644 index 00000000..45cd3489 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReadWindowTests.cs @@ -0,0 +1,96 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests; + +/// +/// Review 07 P8 / P7: the bounded-parallel read window. On the native transport every concurrent tag read of a +/// device rides ONE shared request scheduler that packs whatever is queued into a Multiple Service Packet, so a +/// window of 8 capped MSP coalescing at 8 services per round trip; it is widened to at least 64 there (the +/// scheduler's reply budget already prevents over-packing). And a cold create waiting on a device's connect gate +/// must not occupy a window slot, or one dead device's backoff-expiry attempt stalls every other device's reads. +/// +[Trait("Category", "Unit")] +public sealed class AbCipReadWindowTests +{ + private static AbCipDriverOptions Options(int maxConcurrentReads, AbCipTransport? transport = null) => new() + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions { Enabled = false }, + MaxConcurrentReads = maxConcurrentReads, + Transport = transport, + }; + + /// The window per transport: native ≥ 64; libplctag and injected factories honour the option exactly. + [Fact] + public void The_native_window_is_at_least_64_and_other_transports_keep_the_option() + { + new AbCipDriver(Options(8, AbCipTransport.Native), "a").ReadWindow.ShouldBe(AbCipDriver.NativeMinReadWindow); + AbCipDriver.NativeMinReadWindow.ShouldBeGreaterThanOrEqualTo(64); + new AbCipDriver(Options(1, AbCipTransport.Native), "a").ReadWindow.ShouldBe(AbCipDriver.NativeMinReadWindow); + new AbCipDriver(Options(200, AbCipTransport.Native), "a").ReadWindow.ShouldBe(200); + new AbCipDriver(Options(8), "a", defaultTransport: AbCipTransport.Native).ReadWindow.ShouldBe(AbCipDriver.NativeMinReadWindow); + + // Positive controls: libplctag (per-handle concurrency) and an injected test factory keep the option. + new AbCipDriver(Options(8, AbCipTransport.Libplctag), "a").ReadWindow.ShouldBe(8); + new AbCipDriver(Options(0, AbCipTransport.Libplctag), "a").ReadWindow.ShouldBe(1); + new AbCipDriver(Options(3, AbCipTransport.Native), "a", new FakeAbCipTagFactory()).ReadWindow.ShouldBe(3); + } + + /// A tag whose initialisation blocks until released (a cold connect to a slow / dead device). + private sealed class BlockingInitTag(AbCipTagCreateParams p, Task release) : FakeAbCipTag(p) + { + public override async Task InitializeAsync(CancellationToken cancellationToken) + { + await base.InitializeAsync(cancellationToken); + await release.WaitAsync(cancellationToken); + } + } + + /// + /// With a window of ONE, device B's blocked cold create (and its queued sibling) must not stop device A's tag + /// from being read: cold waiters sit on B's connect gate, not in the window. + /// + [Fact] + public async Task A_blocked_cold_create_on_one_device_does_not_hold_the_window_against_another() + { + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var factory = new FakeAbCipTagFactory + { + Customise = p => p.Gateway == "10.0.0.6" + ? new BlockingInitTag(p, release.Task) { Value = 2 } + : new FakeAbCipTag(p) { Value = 1 }, + }; + const string a = "ab://10.0.0.5/1,0", b = "ab://10.0.0.6/1,0"; + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions(a), new AbCipDeviceOptions(b)], + RawTags = AbCipRawTags.From( + new AbCipTagDefinition("B1", b, "B1", AbCipDataType.DInt), + new AbCipTagDefinition("B2", b, "B2", AbCipDataType.DInt), + new AbCipTagDefinition("A1", a, "A1", AbCipDataType.DInt)), + Probe = new AbCipProbeOptions { Enabled = false }, + MaxConcurrentReads = 1, + }, "abcip-window", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var read = drv.ReadAsync(["B1", "B2", "A1"], CancellationToken.None); + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!(factory.Tags.TryGetValue("A1", out var tag) && tag.ReadCount > 0)) + { + if (DateTime.UtcNow > deadline) + { + release.TrySetResult(); + await read; + throw new ShouldAssertException("device A's read waited behind device B's blocked cold create"); + } + await Task.Delay(5); + } + read.IsCompleted.ShouldBeFalse(); // B is still blocked — A was read regardless + + release.SetResult(); + var r = await read; + r.Select(x => x.Value).ShouldBe([2, 2, 1]); // positive control: B's reads complete once it connects + await drv.DisposeAsync(); + } +}