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

perf(s7): span-grouped reads — one PDU per contiguous block instead of one per tag

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-16 04:15:37 -04:00
parent 63081d52d8
commit 39f67a0fce
6 changed files with 1188 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+256 -37
View File
@@ -500,6 +500,25 @@ public sealed class S7Driver
// ---- IReadable ----
/// <inheritdoc />
/// <remarks>
/// <para>
/// The batch is served in two passes under ONE gate acquisition. First
/// <see cref="S7ReadSpanPlanner"/> groups the resolved tags into contiguous
/// <c>(area, DB)</c> byte spans and each span is read with a single
/// <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
/// per-tag <see cref="ReadOneAsync"/> path.
/// </para>
/// <para>
/// Setting <c>SpanReadMaxGapBytes</c> below zero skips the first pass entirely, 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
/// stamps the same status whichever path observed it.
/// </para>
/// </remarks>
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
@@ -534,17 +553,69 @@ public sealed class S7Driver
return results;
}
// Resolve every reference first. An unresolvable one is a miss, stamped here and never
// offered to the planner — grouping must not be able to change what "unknown tag" means.
var resolved = new S7TagDefinition?[fullReferences.Count];
for (var i = 0; i < fullReferences.Count; i++)
{
var name = fullReferences[i];
if (!_resolver.TryResolve(name, out var tag))
{
if (_resolver.TryResolve(fullReferences[i], out var tag))
resolved[i] = tag;
else
results[i] = new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now);
continue;
}
// Indexes the per-tag path must serve: everything, when grouping is off; otherwise the
// planner's leftovers (Timer/Counter) plus any tag whose address won't parse.
var perTag = new List<int>();
if (_options.SpanReadMaxGapBytes >= 0)
{
List<(int OriginalIndex, S7TagDefinition Tag, S7ParsedAddress Addr)>? plannable = null;
for (var i = 0; i < fullReferences.Count; i++)
{
if (resolved[i] is not { } tag) continue;
if (!TryGetParsedAddress(tag, out var addr))
{
// Defensive: every tag pre-parses at init. A tag that somehow didn't goes
// to the per-tag path, where ReadOneAsync's Parse throws the FormatException
// the catch ladder has always mapped to BadCommunicationError.
perTag.Add(i);
continue;
}
(plannable ??= []).Add((i, tag, addr));
}
if (plannable is not null)
{
var plan = S7ReadSpanPlanner.Build(
plannable, _options.SpanReadMaxGapBytes, _options.SpanReadMaxSpanBytes);
foreach (var span in plan.Spans)
{
// A span that saves no round-trip is not worth changing the wire shape for.
if (PrefersPerTagRead(span))
{
perTag.Add(span.Members[0].OriginalIndex);
continue;
}
await ReadSpanAsync(plc, span, results, now, cancellationToken).ConfigureAwait(false);
}
perTag.AddRange(plan.LeftoverIndexes);
}
}
else
{
for (var i = 0; i < fullReferences.Count; i++)
if (resolved[i] is not null) perTag.Add(i);
}
// 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();
foreach (var i in perTag)
{
try
{
var value = await ReadOneAsync(plc, tag, cancellationToken).ConfigureAwait(false);
var value = await ReadOneAsync(plc, resolved[i]!, cancellationToken).ConfigureAwait(false);
results[i] = new DataValueSnapshot(value, 0u, now, now);
_health = new DriverHealth(DriverState.Healthy, now, null);
}
@@ -555,42 +626,14 @@ public sealed class S7Driver
// dead (we hold _gate) and propagate: the caller cancelled, nobody consumes the
// batch. Marking is unconditional on OCE (not routed through the type classifier):
// whether bytes moved is unknowable, and a false positive costs one cheap reopen
// versus permanent desync for a false negative.
// versus permanent desync for a false negative. Kept AT the call site rather
// than inside RecordReadFailure so the rethrow preserves the original stack.
_plcDead = true;
throw;
}
catch (NotSupportedException)
{
results[i] = new DataValueSnapshot(null, StatusBadNotSupported, null, now);
}
catch (PlcException pex) when (IsAccessDenied(pex))
{
// PUT/GET-disabled (S7-1200/1500) / access-protection — a permanent
// configuration fault, NOT a transient one. Blind retry is wasted effort,
// so map it to BadNotSupported and flag the driver as a config alert
// (Faulted) rather than Degraded — per driver-specs.md §5.
results[i] = new DataValueSnapshot(null, StatusBadNotSupported, null, now);
_health = new DriverHealth(DriverState.Faulted, _health.LastSuccessfulRead,
"S7 access denied — enable PUT/GET communication in TIA Portal " +
$"(Protection & Security) for this CPU. PLC reported: {pex.Message}");
}
catch (PlcException pex)
{
// A genuine device-layer fault (CPU error, hardware fault) — transient
// enough to keep retrying; report BadDeviceFailure and degrade health. A
// connection-fatal PlcException (socket drop / ConnectionError) additionally
// marks the handle dead so the next read reopens it.
MarkConnectionDeadIfFatal(pex);
results[i] = new DataValueSnapshot(null, StatusBadDeviceFailure, null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, pex.Message);
}
catch (Exception ex)
{
// Socket / timeout / transport fault — mark the handle dead when it's a
// connection loss so EnsureConnectedAsync reopens on the next call.
MarkConnectionDeadIfFatal(ex);
results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
results[i] = new DataValueSnapshot(null, RecordReadFailure(ex), null, now);
}
}
}
@@ -598,6 +641,182 @@ public sealed class S7Driver
return results;
}
/// <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.
/// </summary>
/// <remarks>
/// The point of grouping is to replace N wire requests with one. A lone narrow tag is one
/// wire request either way — but a block read reaches the CPU differently from S7.Net's
/// per-address string read, so converting it would change the observable wire shape of every
/// single-tag read on the server for zero benefit. A lone ARRAY or wide/buffer scalar is NOT
/// demoted: those already read a block on the per-tag path, so serving them as a
/// single-member span is byte-for-byte the request they always issued.
/// </remarks>
/// <param name="span">The planned span to judge.</param>
/// <returns><c>true</c> when the span's single member should go through <see cref="ReadOneAsync"/>.</returns>
private static bool PrefersPerTagRead(S7ReadSpanPlanner.Span span)
=> span.Members.Count == 1
&& span.Members[0].Tag.ArrayCount is not >= 1
&& !IsBufferType(span.Members[0].Tag, span.Members[0].Addr);
/// <summary>
/// Reads one planned span — a SINGLE <c>ReadBytesAsync</c> serving every member — and
/// decodes each member from its slice of the returned block.
/// </summary>
/// <remarks>
/// A span read is one wire request standing in for many per-tag requests, so a transport
/// fault that used to spoil one tag now spoils all of them: every member is stamped with
/// the SAME status <see cref="RecordReadFailure"/> would have produced for that exception on
/// the per-tag path, and health degrades identically. Decoding, by contrast, is per member
/// inside its own guard — a single mis-authored tag (declared type vs. address size) must
/// not spoil the neighbours that shared its block.
/// </remarks>
/// <param name="plc">The live connection (caller holds <see cref="_gate"/>).</param>
/// <param name="span">The planned block read.</param>
/// <param name="results">The batch's result array, written in place at each member'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 read.</param>
/// <returns>A task that completes once every member of the span has been stamped.</returns>
private async Task ReadSpanAsync(
IS7Plc plc, S7ReadSpanPlanner.Span span, DataValueSnapshot[] results, DateTime now, CancellationToken ct)
{
byte[] block;
try
{
block = await plc.ReadBytesAsync(
ToS7NetArea(span.Area), span.DbNumber, span.StartByte, span.LengthBytes, ct)
.ConfigureAwait(false)
?? throw new System.IO.InvalidDataException(
$"S7.Net returned null block for {span.Area} db{span.DbNumber} " +
$"@{span.StartByte}+{span.LengthBytes}");
}
catch (OperationCanceledException)
{
// Same reasoning as the per-tag path: a cancelled block read leaves the single gated
// stream at an untrustworthy position, so the handle must be reopened (STAB-15a).
_plcDead = true;
throw;
}
catch (Exception ex)
{
var status = RecordReadFailure(ex);
foreach (var member in span.Members)
results[member.OriginalIndex] = new DataValueSnapshot(null, status, null, now);
return;
}
foreach (var member in span.Members)
{
try
{
var value = DecodeMember(member, block);
results[member.OriginalIndex] = new DataValueSnapshot(value, 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 block-read one above.
results[member.OriginalIndex] = new DataValueSnapshot(null, RecordReadFailure(ex), null, now);
}
}
}
/// <summary>
/// Decodes one span member from its slice of the block, routing to exactly the decoder
/// <see cref="ReadOneAsync"/> would have chosen for that tag, in the same order.
/// </summary>
/// <param name="member">The member to decode, carrying its slice position and width.</param>
/// <param name="block">The block the span read.</param>
/// <returns>The decoded value, boxed identically to the per-tag path's.</returns>
private static object DecodeMember(S7ReadSpanPlanner.Member member, byte[] block)
{
// The two block decoders take byte[] (they hand the buffer to S7.Net's own
// S7String/DateTime/Timer/Counter FromByteArray helpers, which are byte[]-only), so a
// member that is not the whole block is copied out. Narrow scalars need no copy at all —
// DecodeNarrowScalar reads straight from a ReadOnlySpan positioned at the tag's own byte.
if (member.Tag.ArrayCount is >= 1)
return DecodeArrayBlock(member.Tag, member.Addr, SliceForMember(member, block));
if (IsBufferType(member.Tag, member.Addr))
return DecodeScalarBlock(member.Tag, member.Addr, SliceForMember(member, block));
return DecodeNarrowScalar(member.Tag, member.Addr, block.AsSpan(member.RelativeOffset, member.WidthBytes));
}
/// <summary>
/// The member's own bytes as a <c>byte[]</c> for the block decoders. Returns the block
/// itself — no copy — when the member IS the whole block, which is the single-member span
/// every array / wide scalar produces today.
/// </summary>
/// <param name="member">The member whose bytes are wanted.</param>
/// <param name="block">The block the span read.</param>
/// <returns>The member's bytes, starting at its own first byte.</returns>
private static byte[] SliceForMember(S7ReadSpanPlanner.Member member, byte[] block)
=> member.RelativeOffset == 0 && member.WidthBytes == block.Length
? block
: block[member.RelativeOffset..(member.RelativeOffset + member.WidthBytes)];
/// <summary>
/// The tag's parsed address: the table built at init, with an on-demand parse as the
/// defensive fallback for a tag that isn't in it. Mirrors <see cref="ReadOneAsync"/>'s
/// lookup, but never throws — an unparseable address returns <c>false</c> so the caller can
/// route the tag to the per-tag path, where the <see cref="FormatException"/> surfaces
/// through the ladder exactly as it always has.
/// </summary>
/// <param name="tag">The tag whose address is wanted.</param>
/// <param name="addr">The parsed address, valid only when this returns <c>true</c>.</param>
/// <returns><c>true</c> when the address is known or parses.</returns>
private bool TryGetParsedAddress(S7TagDefinition tag, out S7ParsedAddress addr)
=> _parsedByRawPath.TryGetValue(tag.Name, out addr) || S7AddressParser.TryParse(tag.Address, out addr);
/// <summary>
/// The read-failure classification ladder, shared by the per-tag path and the span path so
/// a failed span stamps every member with exactly the status the same exception would have
/// produced on a single tag. Applies the health / dead-handle side effects and returns the
/// status code the caller must stamp.
/// </summary>
/// <remarks>
/// <see cref="OperationCanceledException"/> is deliberately NOT handled here: it is control
/// flow rather than a per-tag status (it marks the handle dead and propagates), and
/// rethrowing it from a helper would rewrite the stack. Both call sites catch it first.
/// </remarks>
/// <param name="ex">The exception observed on a read.</param>
/// <returns>The OPC UA status code every affected tag must be stamped with.</returns>
private uint RecordReadFailure(Exception ex)
{
switch (ex)
{
case NotSupportedException:
return StatusBadNotSupported;
case PlcException pex when IsAccessDenied(pex):
// PUT/GET-disabled (S7-1200/1500) / access-protection — a permanent configuration
// fault, NOT a transient one. Blind retry is wasted effort, so map it to
// BadNotSupported and flag the driver as a config alert (Faulted) rather than
// Degraded — per driver-specs.md §5.
_health = new DriverHealth(DriverState.Faulted, _health.LastSuccessfulRead,
"S7 access denied — enable PUT/GET communication in TIA Portal " +
$"(Protection & Security) for this CPU. PLC reported: {pex.Message}");
return StatusBadNotSupported;
case PlcException pex:
// A genuine device-layer fault (CPU error, hardware fault) — transient enough to
// keep retrying; report BadDeviceFailure and degrade health. A connection-fatal
// PlcException (socket drop / ConnectionError) additionally marks the handle dead
// so the next read reopens it.
MarkConnectionDeadIfFatal(pex);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, pex.Message);
return StatusBadDeviceFailure;
default:
// Socket / timeout / transport fault — mark the handle dead when it's a connection
// loss so EnsureConnectedAsync reopens on the next call.
MarkConnectionDeadIfFatal(ex);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
return StatusBadCommunicationError;
}
}
private async Task<object> ReadOneAsync(IS7Plc plc, S7TagDefinition tag, CancellationToken ct)
{
// Every tag pre-parses its address at init (_parsedByRawPath); the on-demand
@@ -681,7 +900,7 @@ public sealed class S7Driver
/// <param name="tag">Tag definition carrying the <see cref="S7DataType"/>.</param>
/// <param name="addr">Parsed address — its <see cref="S7Area"/> routes Timer/Counter into the seam regardless of <see cref="S7DataType"/>.</param>
/// <returns><c>true</c> if the tag routes through the byte-buffer codec.</returns>
private static bool IsBufferType(S7TagDefinition tag, S7ParsedAddress addr) =>
internal static bool IsBufferType(S7TagDefinition tag, S7ParsedAddress addr) =>
addr.Area is S7Area.Timer or S7Area.Counter
|| tag.DataType is S7DataType.Int64 or S7DataType.UInt64 or S7DataType.Float64
or S7DataType.String or S7DataType.DateTime;