perf(ablegacy): coalesce adjacent N/B/F/L scalar reads into one PCCC request

Review 07 P7 (rest): adjacent words on one data file were never coalesced
(one round trip per tag) although the array path proves the transport can read
a span. Within a device pass, scalar tags on one N/B/F/L file within 8
elements of each other (block <= 200 data bytes, under the 236-byte typed-read
limit) now ride one multi-element read and are decoded from their offsets with
exactly the CLR types the single-tag path returns (int / float / bool).

Only shapes the single path decodes identically are coalesced: indirect,
sub-element, array, file-number-less (S:/I:/O:) and width-mismatched tags keep
the single path. A block the device refuses (an authored element past the end
of the file, a native PLC-5 non-N file) falls back to per-tag reads for its
members and its shape is remembered; a comms fault fails the members together.

New driver-config key coalesceReads (default true) with an AdminUI switch
(Razor change: AbLegacyDriverForm, for the Wave 4 live check). Adds an
env-gated ab_server parity case (coalesced == per-tag values).

Refs #738
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 14:41:20 -04:00
parent e5527d0201
commit 530f4bdcd4
9 changed files with 563 additions and 6 deletions
@@ -48,6 +48,13 @@ public sealed class AbLegacyDriverOptions
/// <summary>Gets or sets the default timeout for read/write operations.</summary>
public TimeSpan Timeout { get; init; } = TimeSpan.FromSeconds(2);
/// <summary>
/// Whether scalar tags on adjacent elements of one N / B / F / L file are read as ONE multi-element PCCC read
/// (driver-config key <c>coalesceReads</c>, default <see langword="true"/>). A block the device refuses falls
/// back to per-tag reads on its own; set this false to read every tag singly.
/// </summary>
public bool CoalesceReads { get; init; } = true;
/// <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.
@@ -0,0 +1,242 @@
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy;
/// <summary>
/// Adjacent-element read coalescing (review 07 P7): scalar tags on the same N / B / F / L data file whose
/// elements are close together are read as ONE multi-element PCCC read (the same path array tags already use),
/// then each tag is decoded from its offset in the block. One round trip instead of one per tag.
/// </summary>
/// <remarks>
/// <para><b>Only shapes the single-tag path decodes identically are coalesced</b>: a word tag
/// (<c>Int</c>/<c>AnalogInt</c>) or a bit of a word (<c>Bit</c>, with or without <c>/n</c>) on an N or B file; a
/// <c>Float</c> on an F file; a <c>Long</c> or a bit of a long on an L file. Indirect addresses, sub-elements
/// (<c>T4:0.ACC</c>), arrays, the file-number-less system files (<c>S:</c>, <c>I:</c>, <c>O:</c>) and any tag
/// authored with a type that does not match its file's element width keep the single-tag path, so a
/// mis-authored tag still gets exactly the answer it got before.</para>
/// <para><b>The span is safe to read.</b> A block runs from its lowest to its highest member, so every element
/// it reads lies between two authored elements of the same file — inside the file whenever the authored tags
/// are. A device REFUSAL of a block (an authored element past the end of the file, a PLC-5 native non-N file)
/// falls back to single reads for that block's members, so one bad tag never blackens its neighbours, and the
/// block shape is remembered so it is not retried every poll. A comms fault fails the members together — they
/// share the connection — exactly as their single reads would have.</para>
/// <para>Turned off with the driver-config key <c>coalesceReads: false</c>.</para>
/// </remarks>
public sealed partial class AbLegacyDriver
{
/// <summary>Largest block, in data bytes. Below the 236-byte PCCC typed-read limit, so libplctag never needs
/// to fragment a coalesced read (the native runtime would chunk it anyway).</summary>
internal const int MaxCoalescedBytes = 200;
/// <summary>Largest run of UNAUTHORED elements a block may read across to join two members.</summary>
internal const int MaxCoalesceGap = 8;
/// <summary>One unit of a device pass: a single tag, or a coalesced block of several.</summary>
private sealed record ReadUnit(ReadItem? Single, CoalescedBlock? Block);
/// <summary>A coalesced block: <paramref name="Count"/> elements of one file from <paramref name="Start"/>.</summary>
private sealed record CoalescedBlock(
string FileLetter, int FileNumber, int Start, int Count, AbLegacyDataType ElementType, List<ReadItem> Members)
{
/// <summary>The runtime-cache key; distinct from any RawPath (RawPaths never contain '|').</summary>
public string Key => $"|block|{FileLetter}{FileNumber}:{Start}+{Count}";
/// <summary>The block's base address in libplctag form.</summary>
public string StartName => $"{FileLetter}{FileNumber}:{Start}";
}
/// <summary>
/// The element type a scalar tag is decoded from when it rides a coalesced block, or <see langword="null"/>
/// when the tag keeps the single-tag path.
/// </summary>
private static AbLegacyDataType? CoalescedElementType(ReadItem item)
{
var a = item.Parsed;
if (a.IsIndirect || a.SubElement is not null || a.FileNumber is null || IsArrayTag(item.Def)) return null;
return (a.FileLetter, item.Def.DataType) switch
{
("N" or "B", AbLegacyDataType.Int or AbLegacyDataType.AnalogInt or AbLegacyDataType.Bit) => AbLegacyDataType.Int,
("F", AbLegacyDataType.Float) => AbLegacyDataType.Float,
("L", AbLegacyDataType.Long) => AbLegacyDataType.Long,
("L", AbLegacyDataType.Bit) when a.BitIndex is not null => AbLegacyDataType.Long,
_ => null,
};
}
/// <summary>
/// Plans a device pass: coalescable tags grouped per file into runs of nearby elements (a run of two or more
/// tags becomes a block), everything else a single unit. Block shapes the device refused before stay single.
/// </summary>
private List<ReadUnit> PlanReadUnits(DeviceState device, List<ReadItem> items)
{
if (!_options.CoalesceReads || items.Count < 2) return [.. items.Select(i => new ReadUnit(i, null))];
var units = new List<ReadUnit>(items.Count);
var groups = new Dictionary<(string Letter, int File, AbLegacyDataType Type), List<ReadItem>>();
foreach (var item in items)
{
if (CoalescedElementType(item) is not { } type)
{
units.Add(new ReadUnit(item, null));
continue;
}
var key = (item.Parsed.FileLetter, item.Parsed.FileNumber!.Value, type);
if (!groups.TryGetValue(key, out var list)) groups[key] = list = [];
list.Add(item);
}
foreach (var ((letter, file, type), members) in groups)
{
var width = type == AbLegacyDataType.Int ? 2 : 4;
var maxElements = MaxCoalescedBytes / width;
var ordered = members.OrderBy(m => m.Parsed.WordNumber).ToList();
var run = new List<ReadItem> { ordered[0] };
for (var i = 1; i <= ordered.Count; i++)
{
var next = i < ordered.Count ? ordered[i] : null;
var start = run[0].Parsed.WordNumber;
if (next is not null
&& next.Parsed.WordNumber - run[^1].Parsed.WordNumber <= MaxCoalesceGap + 1
&& next.Parsed.WordNumber - start + 1 <= maxElements)
{
run.Add(next);
continue;
}
var count = run[^1].Parsed.WordNumber - start + 1;
var block = new CoalescedBlock(letter, file, start, count, type, run);
if (run.Count < 2 || device.RefusedBlocks.ContainsKey(block.Key))
units.AddRange(run.Select(r => new ReadUnit(r, null)));
else
units.Add(new ReadUnit(null, block));
if (next is not null) run = [next];
}
}
return units;
}
/// <summary>What a block read concluded; <see cref="Fallback"/> = read the members singly instead.</summary>
private enum BlockOutcome { Done, Fallback }
/// <summary>
/// Reads one coalesced block and decodes each member from its offset. A device refusal or a local refusal
/// returns <see cref="BlockOutcome.Fallback"/> (and remembers the shape) without writing any member's result.
/// </summary>
private async Task<(BlockOutcome Outcome, OpVerdict Verdict, string? Message)> ReadBlockAsync(
DeviceState device, CoalescedBlock block, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
{
void FailAll(uint status)
{
foreach (var m in block.Members) results[m.Index] = new DataValueSnapshot(null, status, null, now);
}
(BlockOutcome, OpVerdict, string?) Refused(string why)
{
device.RefusedBlocks.TryAdd(block.Key, 0);
_logger.LogInformation(
"AbLegacy coalesced read {Block} refused ({Reason}); its {Count} tags read singly from now on. Driver={DriverInstanceId} Device={DeviceHostAddress}",
block.StartName + "+" + block.Count, why, block.Members.Count, _driverInstanceId, device.Options.HostAddress);
return (BlockOutcome.Fallback, OpVerdict.Neutral, null);
}
try
{
var createParams = new AbLegacyTagCreateParams(
Gateway: device.ParsedAddress.Gateway,
Port: device.ParsedAddress.Port,
CipPath: device.EffectiveCipPath,
LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute,
TagName: block.StartName,
Timeout: _options.Timeout,
ElementCount: block.Count);
var runtime = await CreateSingleFlightAsync(
device, device.Runtimes, block.Key, () => _tagFactory.Create(createParams), ct).ConfigureAwait(false);
int status;
Array? data = null;
var opLock = device.GetRuntimeLock(block.Key);
await opLock.WaitAsync(ct).ConfigureAwait(false);
try
{
await runtime.ReadAsync(ct).ConfigureAwait(false);
status = runtime.GetStatus();
if (status == 0) data = runtime.DecodeArray(block.ElementType, block.Count) as Array;
}
finally
{
opLock.Release();
}
if (status != 0)
{
if (IsDeviceRefusal((AbStatus)status) || (AbStatus)status == AbStatus.Unsupported)
{
EvictRuntime(device, block.Key);
return Refused($"device status {(AbStatus)status}");
}
EvictRuntime(device, block.Key);
FailAll(AbLegacyStatusMapper.MapStatus(status));
return (BlockOutcome.Done, OpVerdict.Fault, $"transport status {status} reading {block.StartName}+{block.Count}");
}
if (data is null || data.Length < block.Count)
return Refused("the block decoded short");
foreach (var m in block.Members)
results[m.Index] = new DataValueSnapshot(DecodeMember(block, m, data), AbLegacyStatusMapper.Good, now, now);
device.FirstNonZeroStatusLogged = false;
return (BlockOutcome.Done, OpVerdict.Answered, null);
}
catch (OperationCanceledException) { throw; }
catch (ConnectionThrottledException tex)
{
FailAll(AbLegacyStatusMapper.BadCommunicationError);
return (BlockOutcome.Done, OpVerdict.Throttled, tex.Message);
}
catch (AbTransportException ex) when (IsLocalRefusal(ex) || IsDeviceRefusal(ex.Status))
{
EvictRuntime(device, block.Key);
return Refused(ex.Message);
}
catch (NotSupportedException nse)
{
EvictRuntime(device, block.Key);
return Refused(nse.Message);
}
catch (Exception ex)
{
EvictRuntime(device, block.Key);
FailAll(AbLegacyStatusMapper.BadCommunicationError);
return (BlockOutcome.Done, OpVerdict.Fault, ex.Message);
}
}
/// <summary>
/// Decodes one member from the block, returning exactly the CLR type and value the single-tag path's
/// <see cref="IAbLegacyTagRuntime.DecodeValue"/> would (<c>int</c> for a word or long, <c>float</c>, <c>bool</c>).
/// </summary>
private static object DecodeMember(CoalescedBlock block, ReadItem m, Array data)
{
var offset = m.Parsed.WordNumber - block.Start;
var bit = m.Parsed.BitIndex;
switch (data)
{
case short[] words:
{
var w = words[offset];
return m.Def.DataType == AbLegacyDataType.Bit
? ((ushort)w >> (bit ?? 0) & 1) != 0
: (int)w;
}
case int[] longs:
{
var l = longs[offset];
return m.Def.DataType == AbLegacyDataType.Bit ? ((uint)l >> bit!.Value & 1) != 0 : l;
}
case float[] floats:
return floats[offset];
default:
throw new InvalidOperationException($"Unexpected block buffer {data.GetType().Name}.");
}
}
}
@@ -12,7 +12,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy;
/// <see cref="IDriver"/> only at PR 1 time; read / write / discovery / subscribe / probe /
/// host-resolver capabilities ship in PRs 2 and 3.
/// </summary>
public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable,
public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable,
IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table.</summary>
@@ -565,21 +565,42 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
DeviceState device, List<ReadItem> items, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
{
var outcome = new DeviceReadOutcome();
foreach (var item in items)
var work = new Queue<ReadUnit>(PlanReadUnits(device, items));
while (work.TryDequeue(out var unit))
{
var members = unit.Block?.Members ?? [unit.Single!];
if (device.IsThrottled(DateTime.UtcNow))
{
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now);
foreach (var m in members)
results[m.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now);
outcome.FaultMessage ??= $"device {device.Options.HostAddress} is in its reconnect backoff window";
continue;
}
var (verdict, message) = await ReadOneAsync(device, item, results, now, ct).ConfigureAwait(false);
OpVerdict verdict;
string? message;
if (unit.Block is { } block)
{
(var blockOutcome, verdict, message) = await ReadBlockAsync(device, block, results, now, ct).ConfigureAwait(false);
if (blockOutcome == BlockOutcome.Fallback)
{
// Refused as a block: read its members singly, now, so each gets its own answer.
var rest = work.ToList();
work.Clear();
foreach (var m in block.Members) work.Enqueue(new ReadUnit(m, null));
foreach (var r in rest) work.Enqueue(r);
continue;
}
}
else
{
(verdict, message) = await ReadOneAsync(device, unit.Single!, results, now, ct).ConfigureAwait(false);
}
switch (verdict)
{
case OpVerdict.Answered:
outcome.AnyAnswered = true;
if (results[item.Index].StatusCode == AbLegacyStatusMapper.Good) outcome.AnyGood = true;
if (members.Any(m => results[m.Index].StatusCode == AbLegacyStatusMapper.Good)) outcome.AnyGood = true;
device.RecordAnswered();
break;
case OpVerdict.Fault:
@@ -1685,6 +1706,10 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover
/// </summary>
public bool FirstNonZeroStatusLogged { get; set; }
/// <summary>Coalesced block shapes the device refused; their members read singly from then on (bounded by the
/// distinct subscription shapes, released with the state).</summary>
public System.Collections.Concurrent.ConcurrentDictionary<string, byte> RefusedBlocks { get; } = new(StringComparer.Ordinal);
/// <summary>Tag operations already warned about for a deterministic refusal (once-per-tag-and-operation log).</summary>
public System.Collections.Concurrent.ConcurrentDictionary<string, byte> RefusalLogged { get; } = new(StringComparer.Ordinal);
@@ -97,6 +97,7 @@ public static class AbLegacyDriverFactoryExtensions
ProbeAddress = dto.Probe?.ProbeAddress ?? "S:0",
},
Timeout = PositiveTimeoutOrDefault(dto.TimeoutMs, 2_000),
CoalesceReads = dto.CoalesceReads ?? true,
// Absent ⇒ null ("use the default" = Libplctag, resolved by AbLegacyTransportResolver);
// a bad value still fails exactly as any other bad enum field does.
Transport = string.IsNullOrWhiteSpace(dto.Transport)
@@ -157,6 +158,9 @@ public static class AbLegacyDriverFactoryExtensions
/// </summary>
public string? Transport { get; init; }
/// <summary>Gets or sets whether adjacent scalar reads are coalesced; absent means <see langword="true"/>.</summary>
public bool? CoalesceReads { get; init; }
/// <summary>
/// Gets or sets the list of devices to connect to.
/// </summary>
@@ -28,6 +28,13 @@
</InputSelect>
<div class="form-text">Which protocol stack talks to the controller. Default omits the setting (libplctag). Native is opt-in: wire-proven only for N-file reads/writes and F8 reads. Takes effect on the next deploy.</div>
</div>
<div class="col-md-6">
<div class="form-check form-switch mt-4">
<InputCheckbox @bind-Value="_form.CoalesceReads" @bind-Value:after="EmitAsync" class="form-check-input" id="ablegacyCoalesceReads" />
<label class="form-check-label" for="ablegacyCoalesceReads">Coalesce adjacent reads</label>
</div>
<div class="form-text">Read nearby scalar tags on one N/B/F/L file as a single PCCC request. A block the PLC refuses falls back to per-tag reads automatically. Default on.</div>
</div>
</div>
</div>
</section>
@@ -97,7 +104,7 @@
/// <summary>JSON paths this form edits; everything else in the stored config is preserved.</summary>
internal static readonly string[] OwnedPaths =
[
"transport", "probeTimeoutSeconds",
"transport", "probeTimeoutSeconds", "coalesceReads",
"probe.enabled", "probe.probeAddress",
];
@@ -163,6 +170,7 @@
// PLC-5 / LogixPccc exclusion still applies). Never re-emit a concrete default here — that would pin
// every edited driver to it.
public AbLegacyTransport? Transport { get; set; }
public bool CoalesceReads { get; set; } = true;
// Probe
public bool ProbeEnabled { get; set; } = true;
@@ -200,6 +208,7 @@
_loadedProbeTimeout = o.Probe.Timeout,
TimeoutSeconds = (int)o.Timeout.TotalSeconds,
Transport = o.Transport,
CoalesceReads = o.CoalesceReads,
ProbeEnabled = o.Probe.Enabled,
ProbeIntervalSeconds = (int)o.Probe.Interval.TotalSeconds,
ProbeTimeoutSeconds = (int)o.Probe.Timeout.TotalSeconds,
@@ -220,6 +229,7 @@
},
Timeout = DriverConfigMs.Save(TimeoutSeconds, _loadedTimeout),
Transport = Transport,
CoalesceReads = CoalesceReads,
ProbeTimeoutSeconds = AdminProbeTimeoutSeconds,
};
}