dohertj2/lmxopcua
1
0
Fork 0
Code Issues 43 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(focas): read devices concurrently within a batch (R07:P9)

Browse Source 
ReadAsync walked the whole batch sequentially, so one SLOW CNC (TCP-accept-then-stall,
the live 31i-B case) cost Timeout per tag per poll while no other device was read.
ReadAsync now resolves every reference against one table snapshot, buckets the rest by
device and reads the devices with Task.WhenAll; each device's tags still go in order on
its own serialised socket (the PMC coalescing pre-pass runs per device, first). Results
land at their own indices.

Also precomputes each device's fixed-tree key prefix (it was built per reference, twice
per poll).

Refs #740
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:18:16 -04:00
parent cc140593a0
commit 256915b7e4
2 changed files with 245 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs
+130 -93
View File
@@ -637,23 +637,23 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
// ---- IReadable ----
/// <inheritdoc />
// 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<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> 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<DeviceState, List<int>>? 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;
}
/// <summary>
/// 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.
/// </summary>
private async Task ReadDeviceAsync(
DeviceState device, List<int> indices, IReadOnlyList<string> 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;
}
/// <summary>
/// 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
/// (<see cref="FocasPmcCoalescer"/>) and serves every multi-member range with ONE
/// <c>pmc_rdpmcrng</c>, slicing each tag's value out of the returned slots.
/// </summary>
@@ -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.</item>
/// </list>
/// Returns null when nothing was served.
/// Returns the served indices, or null when nothing was served.
/// </remarks>
private async Task<bool[]?> CoalescePmcReadsAsync(
IReadOnlyList<string> references, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
private async Task<HashSet<int>?> CoalescePmcReadsAsync(
DeviceState device, List<int> indices, IReadOnlyList<string> references, DataValueSnapshot[] results,
DateTime now, CancellationToken ct)
{
Dictionary<DeviceState, List<FocasPmcReadItem>>? byDevice = null;
for (var i = 0; i < references.Count; i++)
List<FocasPmcReadItem>? 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<int>? 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<FocasPmcReadItem> failed, uint status, IReadOnlyList<string> refs,
DataValueSnapshot[] res, DateTime stamp, ref bool[]? servedRef, string reason)
void FailMembers(IEnumerable<FocasPmcReadItem> 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).
/// </summary>
private DataValueSnapshot? TryReadFixedTree(string reference, DateTime now)
private DataValueSnapshot? TryReadFixedTree(
IReadOnlyDictionary<string, DeviceState> 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;
/// <summary>Gets the device configuration options.</summary>
public FocasDeviceOptions Options { get; } = options;
/// <summary>
/// <c>"{HostAddress}/"</c> — the prefix of this device's fixed-tree cache keys, built once rather than
/// per reference per poll (twice per poll before #740).
/// </summary>
public string FixedTreePrefix { get; } = options.HostAddress + "/";
/// <summary>Gets or sets the FOCAS client instance.</summary>
public IFocasClient? Client { get; set; }
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs
+115
View File
@@ -0,0 +1,115 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;
namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests;
/// <summary>
/// #740 / R07:P9 — <see cref="FocasDriver.ReadAsync"/> 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.
/// </summary>
[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";
/// <summary>
/// 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.
/// </summary>
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) { }
}
}
/// <summary>A stalled device no longer holds every other device's read hostage — and each device stays serialised.</summary>
[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<OrderingClient>().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);
}
/// <summary>Results land at their own indices whatever order the devices finish in.</summary>
[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);
}
}