diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs index c8dad722..5c4953d3 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -637,23 +637,23 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // ---- IReadable ---- /// + // Devices are read CONCURRENTLY, each one's tags in order on its own (serialised) socket: a slow CNC — a + // TCP-accept-then-stall control costs its Timeout per tag — used to hold every other device's read in the + // batch hostage, because the whole batch was one sequential walk. public async Task> ReadAsync( IReadOnlyList fullReferences, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(fullReferences); var now = DateTime.UtcNow; var results = new DataValueSnapshot[fullReferences.Count]; + // One snapshot of each swapped table for the whole call (see _devices). + var devices = Devices; + var alarmTags = Volatile.Read(ref _alarmTags); + var fixedTreeActive = _fixedTreeActive; - // Merge adjacent / near PMC reads of one device into fewer pmc_rdpmcrng calls. Indices it - // serves are marked here and skipped by the per-tag loop below; everything else (including every - // single-tag range and any range the CNC rejected) takes the unchanged per-tag path. - var served = _options.PmcCoalesce.Enabled - ? await CoalescePmcReadsAsync(fullReferences, results, now, cancellationToken).ConfigureAwait(false) - : null; - + Dictionary>? byDevice = null; for (var i = 0; i < fullReferences.Count; i++) { - if (served is not null && served[i]) continue; var reference = fullReferences[i]; // Fixed-tree T1 — fixed-tree references are synthesized from the cached @@ -663,25 +663,66 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // This branch matches a reference that IS a "{host}/{path}" fixed-tree key. Authored // tags do not take it — their reference is a RawPath — and are served further down by // their FIXED: address instead. - if (_fixedTreeActive && TryReadFixedTree(reference, now) is { } fx) + if (fixedTreeActive && TryReadFixedTree(devices, reference, now) is { } fx) { results[i] = fx; continue; } // An ALARM: tag is a Part 9 condition, not a value — there is nothing to read. - if (Volatile.Read(ref _alarmTags).ContainsKey(reference)) + if (alarmTags.ContainsKey(reference)) { results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNotSupported, null, now); continue; } - if (!_resolver.TryResolve(reference, out var def)) + if (!_resolver.TryResolve(reference, out var def) + || !devices.TryGetValue(def.DeviceHostAddress, out var device)) { results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now); continue; } - if (!Devices.TryGetValue(def.DeviceHostAddress, out var device)) + + byDevice ??= []; + if (!byDevice.TryGetValue(device, out var indices)) byDevice[device] = indices = []; + indices.Add(i); + } + + if (byDevice is null) return results; + if (byDevice.Count == 1) + { + foreach (var (device, indices) in byDevice) + await ReadDeviceAsync(device, indices, fullReferences, results, now, cancellationToken).ConfigureAwait(false); + } + else + { + await Task.WhenAll(byDevice.Select(kv => + ReadDeviceAsync(kv.Key, kv.Value, fullReferences, results, now, cancellationToken))).ConfigureAwait(false); + } + return results; + } + + /// + /// Reads one device's share of a batch, in request order: the PMC coalescing pre-pass first, then every + /// index it did not serve through the per-tag path. Each result lands at its own index, so concurrent + /// device reads never touch the same slot. + /// + private async Task ReadDeviceAsync( + DeviceState device, List indices, IReadOnlyList fullReferences, + DataValueSnapshot[] results, DateTime now, CancellationToken cancellationToken) + { + // Merge adjacent / near PMC reads of the device into fewer pmc_rdpmcrng calls. Indices it serves are + // marked here and skipped by the per-tag loop below; everything else (including every single-tag range + // and any range the CNC rejected) takes the unchanged per-tag path. + var served = _options.PmcCoalesce.Enabled + ? await CoalescePmcReadsAsync(device, indices, fullReferences, results, now, cancellationToken).ConfigureAwait(false) + : null; + + foreach (var i in indices) + { + if (served is not null && served.Contains(i)) continue; + var reference = fullReferences[i]; + if (!_resolver.TryResolve(reference, out var def)) { results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now); continue; @@ -738,12 +779,10 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, Volatile.Read(ref _health).LastSuccessfulRead, ex.Message)); } } - - return results; } /// - /// PMC coalescing pre-pass. Groups the batch's PMC tags by device, plans merged ranges + /// PMC coalescing pre-pass for ONE device's share of a batch. Plans merged ranges /// () and serves every multi-member range with ONE /// pmc_rdpmcrng, slicing each tag's value out of the returned slots. /// @@ -756,106 +795,98 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// leaves the members UNSERVED, so the per-tag loop reads them individually and each gets its own /// true status — and the range is remembered so later polls skip the doomed merged read. /// - /// Returns null when nothing was served. + /// Returns the served indices, or null when nothing was served. /// - private async Task CoalescePmcReadsAsync( - IReadOnlyList references, DataValueSnapshot[] results, DateTime now, CancellationToken ct) + private async Task?> CoalescePmcReadsAsync( + DeviceState device, List indices, IReadOnlyList references, DataValueSnapshot[] results, + DateTime now, CancellationToken ct) { - Dictionary>? byDevice = null; - for (var i = 0; i < references.Count; i++) + List? items = null; + foreach (var i in indices) { try { - var reference = references[i]; - if (_fixedTreeActive && TryReadFixedTree(reference, now) is not null) continue; - if (!_resolver.TryResolve(reference, out var def)) continue; - if (!Devices.TryGetValue(def.DeviceHostAddress, out var device)) continue; + if (!_resolver.TryResolve(references[i], out var def)) continue; var parsed = ResolveParsedAddress(def); if (parsed.Kind != FocasAreaKind.Pmc || parsed.PmcLetter is null) continue; - (byDevice ??= []).TryAdd(device, []); - byDevice[device].Add(new FocasPmcReadItem(i, parsed.PmcLetter, parsed.Number, def.DataType, parsed.BitIndex)); + (items ??= []).Add(new FocasPmcReadItem(i, parsed.PmcLetter, parsed.Number, def.DataType, parsed.BitIndex)); } catch (Exception ex) when (ex is not OperationCanceledException) { /* malformed / unresolvable — the per-tag loop reports it with its usual status */ } } - if (byDevice is null) return null; + if (items is null || items.Count < 2) return null; - bool[]? served = null; - foreach (var (device, items) in byDevice) + var plans = FocasPmcCoalescer.Plan(items, _options.PmcCoalesce.MaxSpanBytes, _options.PmcCoalesce.MaxGapBytes); + if (!plans.Any(p => p.Members.Count >= 2 && !device.CoalesceRejected.ContainsKey(p.Key))) return null; + + HashSet? served = null; + // One connect attempt per device per batch: if it fails, every PMC tag of the device fails with + // that status here instead of paying a connect attempt per plan and again per tag. + IFocasClient client; + try { client = await EnsureConnectedAsync(device, ct).ConfigureAwait(false); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } + catch (Exception) { - if (items.Count < 2) continue; - var plans = FocasPmcCoalescer.Plan(items, _options.PmcCoalesce.MaxSpanBytes, _options.PmcCoalesce.MaxGapBytes); - if (!plans.Any(p => p.Members.Count >= 2 && !device.CoalesceRejected.ContainsKey(p.Key))) continue; + FailMembers(items, FocasStatusMapper.BadCommunicationError, + $"FOCAS connect failed reading PMC tags of {device.Options.HostAddress}"); + return served; + } - // One connect attempt per device per batch: if it fails, every PMC tag of the device fails with - // that status here instead of paying a connect attempt per plan and again per tag. - IFocasClient client; - try { client = await EnsureConnectedAsync(device, ct).ConfigureAwait(false); } + foreach (var plan in plans) + { + if (plan.Members.Count < 2) continue; + if (device.CoalesceRejected.ContainsKey(plan.Key)) continue; + + var members = plan.Members; + try + { + var (slots, status) = await client + .ReadPmcSlotsAsync(plan.Letter, plan.SlotType, plan.Start, plan.End, ct) + .ConfigureAwait(false); + + if (status == FocasStatusMapper.Good && slots is not null) + { + foreach (var m in members) + { + var slot = plan.SlotOf(m); + results[m.Index] = slot < slots.Count + ? new DataValueSnapshot(FocasPmcCoalescer.MapSlot(slots[slot], m.Type, m.BitIndex), status, now, now) + : new DataValueSnapshot(null, FocasStatusMapper.BadOutOfRange, now, now); + (served ??= []).Add(m.Index); + } + Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, now, null)); + } + else + { + // The CNC REFUSED the merged range (any return code, mapped or not). Don't guess which + // member is at fault: let the per-tag loop give each its own answer. Remember the range + // so later polls skip the doomed read — except for a transient refusal (EW_BUSY maps to + // BadDeviceFailure), which says nothing about the range. + if (status != FocasStatusMapper.BadDeviceFailure) + device.CoalesceRejected[plan.Key] = 0; + _logger.LogDebug( + "FOCAS PMC range {Key} on {Host} rejected (0x{Status:X8}) — reading its {Count} tags individually", + plan.Key, device.Options.HostAddress, status, members.Count); + } + } catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } catch (Exception) { - FailMembers(items, FocasStatusMapper.BadCommunicationError, references, results, now, ref served, - $"FOCAS connect failed reading PMC tags of {device.Options.HostAddress}"); - continue; - } - - foreach (var plan in plans) - { - if (plan.Members.Count < 2) continue; - if (device.CoalesceRejected.ContainsKey(plan.Key)) continue; - - var members = plan.Members; - try - { - var (slots, status) = await client - .ReadPmcSlotsAsync(plan.Letter, plan.SlotType, plan.Start, plan.End, ct) - .ConfigureAwait(false); - - if (status == FocasStatusMapper.Good && slots is not null) - { - foreach (var m in members) - { - var slot = plan.SlotOf(m); - results[m.Index] = slot < slots.Count - ? new DataValueSnapshot(FocasPmcCoalescer.MapSlot(slots[slot], m.Type, m.BitIndex), status, now, now) - : new DataValueSnapshot(null, FocasStatusMapper.BadOutOfRange, now, now); - (served ??= new bool[references.Count])[m.Index] = true; - } - Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, now, null)); - } - else - { - // The CNC REFUSED the merged range (any return code, mapped or not). Don't guess which - // member is at fault: let the per-tag loop give each its own answer. Remember the range - // so later polls skip the doomed read — except for a transient refusal (EW_BUSY maps to - // BadDeviceFailure), which says nothing about the range. - if (status != FocasStatusMapper.BadDeviceFailure) - device.CoalesceRejected[plan.Key] = 0; - _logger.LogDebug( - "FOCAS PMC range {Key} on {Host} rejected (0x{Status:X8}) — reading its {Count} tags individually", - plan.Key, device.Options.HostAddress, status, members.Count); - } - } - catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; } - catch (Exception) - { - // Transport failure / timeout: the link is suspect, so fail the members without per-tag retries. - FailMembers(members, FocasStatusMapper.BadCommunicationError, references, results, now, ref served, - $"FOCAS transport failure reading PMC range {plan.Key}"); - } + // Transport failure / timeout: the link is suspect, so fail the members without per-tag retries. + FailMembers(members, FocasStatusMapper.BadCommunicationError, + $"FOCAS transport failure reading PMC range {plan.Key}"); } } return served; - void FailMembers(IEnumerable failed, uint status, IReadOnlyList refs, - DataValueSnapshot[] res, DateTime stamp, ref bool[]? servedRef, string reason) + void FailMembers(IEnumerable failed, uint status, string reason) { foreach (var m in failed) { - res[m.Index] = new DataValueSnapshot(null, status, null, stamp); - (servedRef ??= new bool[refs.Count])[m.Index] = true; + results[m.Index] = new DataValueSnapshot(null, status, null, now); + (served ??= []).Add(m.Index); } Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, Volatile.Read(ref _health).LastSuccessfulRead, @@ -1604,12 +1635,13 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// the reference looks like a fixed-tree FullName; null when it doesn't /// (callers fall through to the user-authored tag path). /// - private DataValueSnapshot? TryReadFixedTree(string reference, DateTime now) + private DataValueSnapshot? TryReadFixedTree( + IReadOnlyDictionary devices, string reference, DateTime now) { - foreach (var state in Devices.Values) + foreach (var state in devices.Values) { - if (!reference.StartsWith(state.Options.HostAddress + "/", StringComparison.OrdinalIgnoreCase)) continue; - if (TryReadFixedTreeOnDevice(state, reference[(state.Options.HostAddress.Length + 1)..], now) is { } hit) + if (!reference.StartsWith(state.FixedTreePrefix, StringComparison.OrdinalIgnoreCase)) continue; + if (TryReadFixedTreeOnDevice(state, reference[state.FixedTreePrefix.Length..], now) is { } hit) return hit; } return null; @@ -2088,6 +2120,11 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, public FocasHostAddress ParsedAddress { get; } = parsedAddress; /// Gets the device configuration options. public FocasDeviceOptions Options { get; } = options; + /// + /// "{HostAddress}/" — the prefix of this device's fixed-tree cache keys, built once rather than + /// per reference per poll (twice per poll before #740). + /// + public string FixedTreePrefix { get; } = options.HostAddress + "/"; /// Gets or sets the FOCAS client instance. public IFocasClient? Client { get; set; } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs new file mode 100644 index 00000000..9ca35cf7 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs @@ -0,0 +1,115 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.FOCAS; + +namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests; + +/// +/// #740 / R07:P9 — walked the whole batch sequentially, so a SLOW CNC +/// (TCP-accept-then-stall, the live 31i-B case) cost its timeout per tag per poll during which no other +/// device was read. Devices are now read concurrently, each still serialised on its own socket. +/// +[Trait("Category", "Unit")] +public sealed class FocasConcurrentDeviceReadTests +{ + private const string HostA = "focas://10.0.0.5:8193"; + private const string HostB = "focas://10.0.0.6:8193"; + + /// + /// A client that records which device it dialled. Device A's read only completes once device B's read has + /// STARTED — so a sequential walk (A first) can never finish A, while a concurrent one does. Also counts + /// the peak number of reads in flight on any one client. + /// + private sealed class OrderingClient(TaskCompletionSource bStarted) : FakeFocasClient + { + private string _host = ""; + private int _inFlight; + public int PeakInFlight; + + public override Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct) + { + _host = address.ToString(); + return base.ConnectAsync(address, timeout, ct); + } + + public override async Task<(object? value, uint status)> ReadAsync( + FocasAddress address, FocasDataType type, CancellationToken ct) + { + var now = Interlocked.Increment(ref _inFlight); + InterlockedMax(ref PeakInFlight, now); + try + { + if (_host.Contains("10.0.0.6", StringComparison.Ordinal)) bStarted.TrySetResult(); + else await bStarted.Task.WaitAsync(TimeSpan.FromSeconds(2), ct).ConfigureAwait(false); + await Task.Delay(20, ct).ConfigureAwait(false); + return ((sbyte)1, FocasStatusMapper.Good); + } + finally { Interlocked.Decrement(ref _inFlight); } + } + + private static void InterlockedMax(ref int target, int value) + { + int seen; + while ((seen = Volatile.Read(ref target)) < value + && Interlocked.CompareExchange(ref target, value, seen) != seen) { } + } + } + + /// A stalled device no longer holds every other device's read hostage — and each device stays serialised. + [Fact] + public async Task Devices_are_read_concurrently_and_each_stays_serialised() + { + var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var factory = new FakeFocasClientFactory { Customise = () => new OrderingClient(bStarted) }; + var drv = new FocasDriver(new FocasDriverOptions + { + Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")], + RawTags = + [ + FocasRawTags.Entry(new FocasTagDefinition("raw/A1", "A", "R100", FocasDataType.Byte)), + FocasRawTags.Entry(new FocasTagDefinition("raw/A2", "A", "R101", FocasDataType.Byte)), + FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R100", FocasDataType.Byte)), + ], + Probe = new FocasProbeOptions { Enabled = false }, + // Off so every tag takes the per-tag path — this pins device-level concurrency, not coalescing. + PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = false }, + }, "focas-p9", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var snaps = await drv.ReadAsync(["raw/A1", "raw/A2", "raw/B1"], CancellationToken.None); + + snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.Good, + "device A's read can only finish once device B's has started — a sequential walk times A out"); + factory.Clients.Count.ShouldBe(2); + factory.Clients.Cast().ShouldAllBe(c => c.PeakInFlight == 1, + "one device's socket is still request→response: never two reads in flight on it"); + await drv.ShutdownAsync(CancellationToken.None); + } + + /// Results land at their own indices whatever order the devices finish in. + [Fact] + public async Task Results_keep_their_request_order_across_devices() + { + var factory = new FakeFocasClientFactory(); + var drv = new FocasDriver(new FocasDriverOptions + { + Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")], + RawTags = + [ + FocasRawTags.Entry(new FocasTagDefinition("raw/A1", "A", "R100", FocasDataType.Byte)), + FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R200", FocasDataType.Byte)), + ], + Probe = new FocasProbeOptions { Enabled = false }, + }, "focas-p9-order", factory) + ; + factory.Customise = () => new FakeFocasClient { Values = { ["R100"] = (sbyte)1, ["R200"] = (sbyte)2 } }; + await drv.InitializeAsync("{}", CancellationToken.None); + + var snaps = await drv.ReadAsync(["raw/B1", "nope", "raw/A1"], CancellationToken.None); + + snaps[0].Value.ShouldBe((sbyte)2); + snaps[1].StatusCode.ShouldBe(FocasStatusMapper.BadNodeIdUnknown); + snaps[2].Value.ShouldBe((sbyte)1); + await drv.ShutdownAsync(CancellationToken.None); + } +}