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

Auto: s7-b2 — block-read coalescing for contiguous DBs

Browse Source 
Closes #293
This commit is contained in:
Joseph Doherty
2026-04-25 21:23:06 -04:00
parent 5432c49364
commit 17faf76ea7
7 changed files with 976 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
src/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
View File

@@ -1,4 +1,5 @@
using System.Buffers.Binary;
using System.Collections.Generic;
using S7.Net;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -86,6 +87,31 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
private DriverHealth _health = new(DriverState.Unknown, null, null);
private bool _disposed;
// ---- Block-read coalescing diagnostics (PR-S7-B2) ----
//
// Counters surface through DriverHealth.Diagnostics so the driver-diagnostics
// RPC and integration tests can verify wire-level reduction without needing
// access to the underlying S7.Net PDU stream. Names match the
// "<DriverType>.<Counter>" convention adopted for the modbus and opcuaclient
// drivers — see decision #154.
private long _totalBlockReads; // Plc.ReadBytesAsync calls issued by the coalesced path
private long _totalMultiVarBatches; // Plc.ReadMultipleVarsAsync calls issued
private long _totalSingleReads; // per-tag ReadOneAsync fallbacks
/// <summary>
/// Total <c>Plc.ReadBytesAsync</c> calls the coalesced byte-range path issued.
/// Test-only entry point for the integration assertion that 50 contiguous DBWs
/// coalesce into exactly 1 byte-range read.
/// </summary>
internal long TotalBlockReads => Interlocked.Read(ref _totalBlockReads);
/// <summary>
/// Total <c>Plc.ReadMultipleVarsAsync</c> batches issued. For a fully-coalesced
/// contiguous workload this stays at 0 — every tag flows through the byte-range
/// path instead.
/// </summary>
internal long TotalMultiVarBatches => Interlocked.Read(ref _totalMultiVarBatches);
public string DriverInstanceId => driverInstanceId;
public string DriverType => "S7";
@@ -206,10 +232,11 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
try
{
// Phase 1: classify each request into (a) unknown / not-found, (b) packable
// scalar (Bool/Byte/Int16/UInt16/Int32/UInt32/Float32/Float64), or (c) needs
// per-tag fallback (arrays, strings, dates, 64-bit ints, UDT-fanout). Packable
// tags collect into 19-item batches sent via Plc.ReadMultipleVarsAsync; the
// rest stay on the legacy ReadOneAsync path.
// scalar (Bool/Byte/Int16/UInt16/Int32/UInt32/Float32/Float64) which can
// potentially coalesce into a byte-range read, or (c) per-tag fallback
// (arrays, strings, dates, 64-bit ints, UDT-fanout). Packable tags feed
// the block-coalescing planner first (PR-S7-B2); whatever survives as a
// singleton range falls through to the multi-var packer (PR-S7-B1).
var packableIndexes = new List<int>(fullReferences.Count);
var fallbackIndexes = new List<int>();
for (var i = 0; i < fullReferences.Count; i++)
@@ -225,15 +252,55 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
else fallbackIndexes.Add(i);
}
// Phase 2: bin-pack and dispatch the packable group via ReadMultipleVarsAsync.
// On a per-batch S7.Net failure the whole batch falls back to ReadOneAsync per
// tag — that way one bad item doesn't poison the rest of the batch and each
// tag still gets its own per-item StatusCode (BadDeviceFailure for PUT/GET
// refusal, BadCommunicationError for transport faults).
// Phase 2a: block-read coalescing — group same-area / same-DB packable
// tags into contiguous byte ranges (gap-merge threshold from
// S7DriverOptions.BlockCoalescingGapBytes, default 16). Multi-tag ranges
// dispatch via Plc.ReadBytesAsync; singleton ranges fall through to the
// multi-var packer below.
var singletons = new List<int>();
if (packableIndexes.Count > 0)
{
var specs = new List<S7BlockCoalescingPlanner.TagSpec>(packableIndexes.Count);
foreach (var idx in packableIndexes)
{
var tag = _tagsByName[fullReferences[idx]];
var addr = _parsedByName[fullReferences[idx]];
specs.Add(new S7BlockCoalescingPlanner.TagSpec(
CallerIndex: idx,
Area: addr.Area,
DbNumber: addr.DbNumber,
StartByte: addr.ByteOffset,
ByteCount: S7BlockCoalescingPlanner.ScalarByteCount(addr.Size),
OpaqueSize: false));
}
var ranges = S7BlockCoalescingPlanner.Plan(specs, _options.BlockCoalescingGapBytes);
foreach (var range in ranges)
{
if (range.Tags.Count == 1)
{
// Singleton — let the multi-var packer batch it with other
// singletons in the same ReadAsync call. Cheaper than its
// own one-tag ReadBytesAsync round-trip.
singletons.Add(range.Tags[0].CallerIndex);
}
else
{
await ReadCoalescedRangeAsync(plc, range, fullReferences, results, now, cancellationToken)
.ConfigureAwait(false);
}
}
}
// Phase 2b: bin-pack residual singletons through ReadMultipleVarsAsync.
// On a per-batch S7.Net failure the whole batch falls back to ReadOneAsync
// per tag — that way one bad item doesn't poison the rest of the batch
// and each tag still gets its own per-item StatusCode (BadDeviceFailure
// for PUT/GET refusal, BadCommunicationError for transport faults).
if (singletons.Count > 0)
{
var budget = S7ReadPacker.ItemBudget(S7ReadPacker.DefaultPduSize);
var batches = S7ReadPacker.BinPack(packableIndexes, budget);
var batches = S7ReadPacker.BinPack(singletons, budget);
foreach (var batch in batches)
{
await ReadBatchAsync(plc, batch, fullReferences, results, now, cancellationToken)
@@ -255,6 +322,108 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
return results;
}
/// <summary>
/// Issue one coalesced <c>Plc.ReadBytesAsync</c> covering
/// <paramref name="range"/> and slice the response per tag. On a transport
/// fault the whole range falls back to per-tag <see cref="ReadOneAsSnapshotAsync"/>
/// so a single bad slot doesn't poison N-1 good neighbours.
/// </summary>
private async Task ReadCoalescedRangeAsync(
global::S7.Net.Plc plc,
S7BlockCoalescingPlanner.BlockReadRange range,
IReadOnlyList<string> fullReferences,
DataValueSnapshot[] results,
DateTime now,
CancellationToken ct)
{
byte[]? buf;
try
{
Interlocked.Increment(ref _totalBlockReads);
buf = await plc.ReadBytesAsync(MapArea(range.Area), range.DbNumber, range.StartByte, range.ByteCount, ct)
.ConfigureAwait(false);
}
catch (Exception)
{
// Block read fault → fan out per-tag so a bad address in the block
// surfaces its own StatusCode and good neighbours can still retry
// through the per-tag fallback path.
foreach (var slice in range.Tags)
{
var tag = _tagsByName[fullReferences[slice.CallerIndex]];
results[slice.CallerIndex] = await ReadOneAsSnapshotAsync(plc, tag, now, ct).ConfigureAwait(false);
}
return;
}
if (buf is null || buf.Length != range.ByteCount)
{
// Short / truncated PDU — same fan-out semantics as a transport fault.
foreach (var slice in range.Tags)
{
results[slice.CallerIndex] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
}
return;
}
foreach (var slice in range.Tags)
{
var name = fullReferences[slice.CallerIndex];
var tag = _tagsByName[name];
var addr = _parsedByName[name];
try
{
var value = DecodeScalarFromBlock(buf, slice.OffsetInBlock, tag, addr);
results[slice.CallerIndex] = new DataValueSnapshot(value, 0u, now, now);
}
catch (Exception ex)
{
results[slice.CallerIndex] = new DataValueSnapshot(null, StatusBadInternalError, null, now);
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
}
_health = new DriverHealth(DriverState.Healthy, now, null, BuildDiagnostics());
}
/// <summary>
/// Decode one packable scalar from a coalesced byte buffer. Mirrors the
/// reinterpret table in <see cref="S7ReadPacker.DecodePackedValue"/> so the
/// coalesced and per-tag-batch paths produce identical .NET types for the
/// same wire bytes.
/// </summary>
private static object DecodeScalarFromBlock(byte[] buf, int offset, S7TagDefinition tag, S7ParsedAddress addr)
{
return (tag.DataType, addr.Size) switch
{
(S7DataType.Bool, S7Size.Bit) => ((buf[offset] >> addr.BitOffset) & 0x1) == 1,
(S7DataType.Byte, S7Size.Byte) => buf[offset],
(S7DataType.UInt16, S7Size.Word) => BinaryPrimitives.ReadUInt16BigEndian(buf.AsSpan(offset, 2)),
(S7DataType.Int16, S7Size.Word) => BinaryPrimitives.ReadInt16BigEndian(buf.AsSpan(offset, 2)),
(S7DataType.UInt32, S7Size.DWord) => BinaryPrimitives.ReadUInt32BigEndian(buf.AsSpan(offset, 4)),
(S7DataType.Int32, S7Size.DWord) => BinaryPrimitives.ReadInt32BigEndian(buf.AsSpan(offset, 4)),
(S7DataType.Float32, S7Size.DWord) =>
BitConverter.UInt32BitsToSingle(BinaryPrimitives.ReadUInt32BigEndian(buf.AsSpan(offset, 4))),
(S7DataType.Float64, S7Size.LWord) =>
BitConverter.UInt64BitsToDouble(BinaryPrimitives.ReadUInt64BigEndian(buf.AsSpan(offset, 8))),
_ => throw new System.IO.InvalidDataException(
$"S7 block-decode: tag '{tag.Name}' declared {tag.DataType} but address parsed Size={addr.Size}"),
};
}
/// <summary>
/// Snapshot of the wire-level coalescing counters surfaced through
/// <see cref="DriverHealth.Diagnostics"/>. Names follow the
/// <c>"&lt;DriverType&gt;.&lt;Counter&gt;"</c> convention so the driver-diagnostics
/// RPC can render them in the Admin UI alongside Modbus / OPC UA Client
/// metrics without a per-driver special-case.
/// </summary>
private IReadOnlyDictionary<string, double> BuildDiagnostics() => new Dictionary<string, double>
{
["S7.TotalBlockReads"] = Interlocked.Read(ref _totalBlockReads),
["S7.TotalMultiVarBatches"] = Interlocked.Read(ref _totalMultiVarBatches),
["S7.TotalSingleReads"] = Interlocked.Read(ref _totalSingleReads),
};
/// <summary>
/// Read one packed batch via <c>Plc.ReadMultipleVarsAsync</c>. On batch
/// success each <c>DataItem.Value</c> decodes into its tag's snapshot
@@ -279,6 +448,7 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
try
{
Interlocked.Increment(ref _totalMultiVarBatches);
var responses = await plc.ReadMultipleVarsAsync(items, ct).ConfigureAwait(false);
// S7.Net mutates the input list in place and also returns it; iterate by
// index against the input list so we are agnostic to either contract.
@@ -303,7 +473,7 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
}
_health = new DriverHealth(DriverState.Healthy, now, null);
_health = new DriverHealth(DriverState.Healthy, now, null, BuildDiagnostics());
}
catch (Exception)
{
@@ -329,6 +499,7 @@ public sealed class S7Driver(S7DriverOptions options, string driverInstanceId)
{
try
{
Interlocked.Increment(ref _totalSingleReads);
var value = await ReadOneAsync(plc, tag, ct).ConfigureAwait(false);
_health = new DriverHealth(DriverState.Healthy, now, null);
return new DataValueSnapshot(value, 0u, now, now);