perf(s7): span leftovers batch through multi-var read PDUs with per-chunk fallback
This commit is contained in:
Joseph Doherty
@@ -3,6 +3,7 @@ using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using S7.Net;
|
||||
using S7NetDataType = global::S7.Net.DataType;
|
||||
using S7DataItem = global::S7.Net.Types.DataItem;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
@@ -514,11 +515,19 @@ public sealed class S7Driver
|
||||
/// <c>ReadBytesAsync</c>, its members decoded by slicing the returned block — this is
|
||||
/// what turns one wire request per tag into one per contiguous block. Whatever the
|
||||
/// planner declines (Timer/Counter, whose offsets are device numbers) plus any span not
|
||||
/// worth issuing (see <see cref="PrefersPerTagRead"/>) then goes through the unchanged
|
||||
/// worth issuing (see <see cref="PrefersPerTagRead"/>) falls to the leftover list.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Those leftovers get a SECOND grouping pass: the ones a
|
||||
/// <see cref="S7DataItem"/> expresses cleanly — the narrow scalars, which is what a
|
||||
/// demoted single-member span is — are batched into multi-var requests by
|
||||
/// <see cref="ReadLeftoversByMultiVarAsync"/>, which crosses <c>(area, DB)</c>
|
||||
/// boundaries a block read cannot. Whatever remains (Timer/Counter, unparseable
|
||||
/// addresses, and any chunk whose multi-var request faulted) goes through the unchanged
|
||||
/// per-tag <see cref="ReadOneAsync"/> path.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Setting <c>SpanReadMaxGapBytes</c> below zero skips the first pass entirely, which
|
||||
/// Setting <c>SpanReadMaxGapBytes</c> below zero skips BOTH passes, which
|
||||
/// restores the pre-grouping behaviour exactly. Either way the result array is written
|
||||
/// by original index, so the caller sees its own ordering regardless of how the batch
|
||||
/// was partitioned, and both passes share <see cref="RecordReadFailure"/> so a fault
|
||||
@@ -617,6 +626,18 @@ public sealed class S7Driver
|
||||
// Sorted so a batch is served in reference order regardless of how it was partitioned —
|
||||
// the per-tag tail is otherwise ordered by the planner's grouping, not by the caller's.
|
||||
perTag.Sort();
|
||||
|
||||
// Second grouping pass: the leftovers a DataItem can express ride ONE multi-var request
|
||||
// per chunk instead of a wire request each. Deliberately gated on the SAME switch as the
|
||||
// span pass — SpanReadMaxGapBytes < 0 is documented to restore the pre-grouping wire
|
||||
// shape exactly, and batching the (then complete) per-tag list would break that. What
|
||||
// comes back is the tags still owed an individual read, still in reference order.
|
||||
if (_options.SpanReadMaxGapBytes >= 0)
|
||||
{
|
||||
perTag = await ReadLeftoversByMultiVarAsync(
|
||||
plc, resolved, perTag, results, now, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
foreach (var i in perTag)
|
||||
{
|
||||
try
|
||||
@@ -647,6 +668,205 @@ public sealed class S7Driver
|
||||
return results;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Items one multi-var request may carry. <b>The caller owns chunking — S7.Net does NOT
|
||||
/// split a multi-var read</b>, it throws when the request or the projected response exceeds
|
||||
/// the negotiated PDU (<c>AssertPduSizeForRead</c>).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The request costs a 19-byte header plus <b>12 bytes per item</b>; the response costs a
|
||||
/// 14-byte header plus 4 bytes of item header plus the item's own data. Against the common
|
||||
/// 240-byte PDU that permits 18 items request-side and — for the 4-byte-or-narrower scalars
|
||||
/// this pass batches — 28 response-side. 12 sits well inside both, leaving headroom for a
|
||||
/// CPU that negotiates a smaller PDU than 240, which is the failure this constant exists to
|
||||
/// avoid: an oversized request is a hard throw, not a paged read. It is deliberately NOT a
|
||||
/// config key — the safe value is a property of the protocol, not of a deployment.
|
||||
/// </remarks>
|
||||
private const int MultiVarReadMaxItems = 12;
|
||||
|
||||
/// <summary>
|
||||
/// Fewest batchable leftovers worth issuing a multi-var request for. ONE leftover is one
|
||||
/// wire request either way, so batching it would change the observable wire shape of every
|
||||
/// single-tag read for zero benefit — the same reasoning
|
||||
/// <see cref="PrefersPerTagRead"/> applies to a lone narrow span.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The gate is on the PASS, not on each chunk: a batch of 13 still splits into 12 + 1, and
|
||||
/// that trailing single is left in the multi-var path. It costs one wire request either way,
|
||||
/// so re-routing it would buy nothing and would make the chunk loop's index bookkeeping
|
||||
/// answer a second question.
|
||||
/// </remarks>
|
||||
private const int MultiVarReadMinItems = 2;
|
||||
|
||||
/// <summary>
|
||||
/// Serves the span planner's leftovers that a <see cref="S7DataItem"/> can express by
|
||||
/// batching them into multi-var requests, and returns the indexes still owed an individual
|
||||
/// <see cref="ReadOneAsync"/> — in ascending reference order.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Only a narrow scalar is batched.</b> A <see cref="S7DataItem"/> carrying the same
|
||||
/// area / DB / offset / bit / <c>VarType</c> that S7.Net's own address parser derives
|
||||
/// from the tag's address string, with <c>Count == 1</c>, is decoded by the SAME
|
||||
/// <c>ParseBytes</c> call the per-address read path uses — so its boxed value is
|
||||
/// byte-for-byte and type-for-type what <see cref="ReadOneAsync"/> would have produced,
|
||||
/// and <see cref="ReinterpretRawValue"/> applies to it unchanged.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Timer and Counter deliberately stay on the per-tag path</b> even though S7.Net's
|
||||
/// <c>VarType</c> names them. Their per-tag path does not go through <c>ParseBytes</c>
|
||||
/// at all — it block-reads the 2-byte word and decodes it in
|
||||
/// <see cref="DecodeScalarBlock"/>, which surfaces a Counter as <c>int</c> where
|
||||
/// S7.Net's own <c>Counter.FromByteArray</c> returns <c>ushort</c>, and which carries a
|
||||
/// documented BCD follow-up for legacy C-area counters. Batching them would fork that
|
||||
/// decode into a second site that must be kept in step by hand. Correctness beats the
|
||||
/// round-trip. Arrays, wide/buffer scalars and unparseable addresses are likewise left
|
||||
/// alone — the first two already read a block apiece, and the last has no address to
|
||||
/// build an item from.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>A multi-var fault is never a tag's fault.</b> The whole chunk falls back to the
|
||||
/// per-tag path, which is what decides the tag's status — batching must not be able to
|
||||
/// stamp a tag Bad that reading it one at a time would have served, so
|
||||
/// <see cref="RecordReadFailure"/> is deliberately NOT called for the request itself.
|
||||
/// S7.Net offers no per-item error surface to be finer-grained with: a
|
||||
/// <see cref="S7DataItem"/> carries only its <c>Value</c>, and the response parser
|
||||
/// throws on the FIRST item the CPU refused (before populating any later item), so
|
||||
/// per-chunk fallback is the finest granularity the library actually admits. Per-ITEM
|
||||
/// decoding, by contrast, is guarded individually — exactly as the span path guards its
|
||||
/// members — because a mis-authored tag must not spoil the neighbours it shared a
|
||||
/// request with, and it fails identically on the per-tag path anyway.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="plc">The live connection (caller holds <see cref="_gate"/>).</param>
|
||||
/// <param name="resolved">The batch's resolved tags, indexed by reference position.</param>
|
||||
/// <param name="leftovers">Indexes the span pass left for the per-tag path, ascending.</param>
|
||||
/// <param name="results">The batch's result array, written in place at each tag's original index.</param>
|
||||
/// <param name="now">The batch timestamp, shared by every tag in the batch.</param>
|
||||
/// <param name="ct">Cancellation token for the reads.</param>
|
||||
/// <returns>The indexes still owed an individual read, ascending.</returns>
|
||||
private async Task<List<int>> ReadLeftoversByMultiVarAsync(
|
||||
IS7Plc plc,
|
||||
S7TagDefinition?[] resolved,
|
||||
List<int> leftovers,
|
||||
DataValueSnapshot[] results,
|
||||
DateTime now,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var stillPerTag = new List<int>();
|
||||
List<(int Index, S7TagDefinition Tag, S7ParsedAddress Addr)>? batchable = null;
|
||||
|
||||
foreach (var i in leftovers)
|
||||
{
|
||||
var tag = resolved[i]!;
|
||||
if (tag.ArrayCount is >= 1
|
||||
|| !TryGetParsedAddress(tag, out var addr)
|
||||
|| IsBufferType(tag, addr))
|
||||
{
|
||||
stillPerTag.Add(i);
|
||||
continue;
|
||||
}
|
||||
(batchable ??= []).Add((i, tag, addr));
|
||||
}
|
||||
|
||||
// Nothing to gain: leave the (at most one) batchable tag exactly where it was.
|
||||
if (batchable is null || batchable.Count < MultiVarReadMinItems)
|
||||
{
|
||||
if (batchable is not null)
|
||||
foreach (var (index, _, _) in batchable) stillPerTag.Add(index);
|
||||
stillPerTag.Sort();
|
||||
return stillPerTag;
|
||||
}
|
||||
|
||||
for (var start = 0; start < batchable.Count; start += MultiVarReadMaxItems)
|
||||
{
|
||||
var chunk = batchable.GetRange(start, Math.Min(MultiVarReadMaxItems, batchable.Count - start));
|
||||
|
||||
IReadOnlyList<S7DataItem> answered;
|
||||
try
|
||||
{
|
||||
var items = new List<S7DataItem>(chunk.Count);
|
||||
foreach (var (_, _, addr) in chunk) items.Add(ToDataItem(addr));
|
||||
|
||||
answered = await plc.ReadMultipleVarsAsync(items, ct).ConfigureAwait(false);
|
||||
|
||||
// The contract is "the same items back, in the supplied order". Anything else is
|
||||
// unmappable to the batch's indexes, so treat it as a failed chunk rather than
|
||||
// guessing which tag got which value.
|
||||
if (answered is null || answered.Count != chunk.Count)
|
||||
{
|
||||
throw new System.IO.InvalidDataException(
|
||||
$"S7.Net returned {answered?.Count.ToString() ?? "no"} multi-var item(s) " +
|
||||
$"for a {chunk.Count}-item request");
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Same reasoning as the per-tag and span paths: a cancelled read abandons a half-read
|
||||
// response on the single gated stream, so the handle must be reopened (STAB-15a).
|
||||
_plcDead = true;
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogDebug(ex,
|
||||
"S7 driver {DriverInstanceId}: multi-var read of {ItemCount} leftover tag(s) failed; " +
|
||||
"falling back to per-tag reads for that chunk", _driverInstanceId, chunk.Count);
|
||||
foreach (var (index, _, _) in chunk) stillPerTag.Add(index);
|
||||
continue;
|
||||
}
|
||||
|
||||
for (var k = 0; k < chunk.Count; k++)
|
||||
{
|
||||
var (index, tag, addr) = chunk[k];
|
||||
try
|
||||
{
|
||||
var raw = answered[k].Value
|
||||
?? throw new System.IO.InvalidDataException(
|
||||
$"S7.Net returned null multi-var value for '{tag.Address}'");
|
||||
results[index] = new DataValueSnapshot(ReinterpretRawValue(tag, addr, raw), 0u, now, now);
|
||||
_health = new DriverHealth(DriverState.Healthy, now, null);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Decoding is pure and synchronous — no cancellation can surface here, so there
|
||||
// is deliberately no OperationCanceledException arm to shadow the request one.
|
||||
results[index] = new DataValueSnapshot(null, RecordReadFailure(ex), null, now);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The fallback chunks were appended out of order behind the non-batchable tags.
|
||||
stillPerTag.Sort();
|
||||
return stillPerTag;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds the multi-var request item for one narrow scalar — the exact
|
||||
/// <c>(DataType, DB, StartByteAdr, BitAdr, VarType, Count)</c> tuple S7.Net's own address
|
||||
/// parser derives from the equivalent address string, so the item reads and decodes
|
||||
/// identically to <see cref="ReadOneAsync"/>'s per-address read.
|
||||
/// </summary>
|
||||
/// <param name="addr">The tag's parsed address; its <see cref="S7Size"/> picks the var type.</param>
|
||||
/// <returns>The request item, with no value yet.</returns>
|
||||
private static S7DataItem ToDataItem(S7ParsedAddress addr) => new()
|
||||
{
|
||||
DataType = ToS7NetArea(addr.Area),
|
||||
DB = addr.DbNumber,
|
||||
StartByteAdr = addr.ByteOffset,
|
||||
// Non-Bit sizes parse to a bit offset of 0, which S7.Net ignores for those var types.
|
||||
BitAdr = (byte)addr.BitOffset,
|
||||
VarType = addr.Size switch
|
||||
{
|
||||
S7Size.Bit => VarType.Bit,
|
||||
S7Size.Byte => VarType.Byte,
|
||||
S7Size.Word => VarType.Word,
|
||||
S7Size.DWord => VarType.DWord,
|
||||
_ => throw new NotSupportedException($"S7 size {addr.Size} has no multi-var VarType mapping"),
|
||||
},
|
||||
Count = 1,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// True when a planned span is better served by the per-tag path: it holds exactly ONE
|
||||
/// narrow scalar, so issuing it as a block read saves no round-trip and buys nothing.
|
||||
|
||||
Reference in New Issue
Block a user