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

perf(modbus): pooled request ADUs, a reused MBAP header and single-buffer RTU deframing

Browse Source 
R07:L5 (Modbus half, former F2): ModbusTcpTransport builds the MBAP ADU in a
pooled buffer and reads the reply header into one per-transport buffer (sends are
single-flight under the gate). ModbusRtuOverTcpTransport writes its ADU into a
pooled buffer (ModbusRtuFraming.WriteAdu, byte-identical to BuildAdu), and
ReadResponsePduAsync reads and validates the whole frame in one pooled buffer
instead of five arrays. Only the returned PDU is allocated. The exchange stays one
async method: splitting it cost a state-machine box that ate the saving. Scratch
A/B per op: RTU deframe 352 -> 184 B, RTU send 640 -> 408 B, TCP send 864 -> 800 B;
timings within noise; Driver.Modbus.Tests 2 s before and after. NormalizeWordOrder
lives in ModbusDriver (not a transport) and is left alone.

Refs #747
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 17:28:53 -04:00
parent 004c09fa13
commit 7e958cd750
4 changed files with 163 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuFraming.cs
+69 -49
View File
@@ -1,3 +1,4 @@
using System.Buffers;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
/// <summary>
@@ -55,12 +56,33 @@ public static class ModbusRtuFraming
/// <returns>A new array: <c>[unitId, ...pdu, crcLo, crcHi]</c>.</returns>
public static byte[] BuildAdu(byte unitId, ReadOnlySpan<byte> pdu)
{
var frame = new byte[1 + pdu.Length];
frame[0] = unitId;
pdu.CopyTo(frame.AsSpan(1));
return ModbusCrc.Append(frame);
var frame = new byte[AduLength(pdu.Length)];
WriteAdu(unitId, pdu, frame);
return frame;
}
/// <summary>The length of the RTU ADU for a PDU of <paramref name="pduLength"/> bytes (unit id + PDU + CRC).</summary>
/// <param name="pduLength">The bare PDU length.</param>
/// <returns>The ADU length.</returns>
internal static int AduLength(int pduLength) => 1 + pduLength + 2;
/// <summary>Writes the RTU ADU into <paramref name="destination"/> (exactly <see cref="AduLength"/> bytes).</summary>
/// <param name="unitId">The RTU slave/unit address.</param>
/// <param name="pdu">The bare PDU.</param>
/// <param name="destination">The ADU buffer, exactly <see cref="AduLength"/>(<paramref name="pdu"/>.Length) long.</param>
internal static void WriteAdu(byte unitId, ReadOnlySpan<byte> pdu, Span<byte> destination)
{
destination[0] = unitId;
pdu.CopyTo(destination[1..]);
var body = destination[..(1 + pdu.Length)];
var crc = ModbusCrc.Compute(body);
destination[1 + pdu.Length] = (byte)(crc & 0xFF); // CRC low byte first
destination[2 + pdu.Length] = (byte)(crc >> 8);
}
/// <summary>The largest RTU response frame: addr + fc + byteCount + 255 data bytes + CRC.</summary>
private const int MaxResponseFrame = 3 + 255 + 2;
/// <summary>
/// Reads a single RTU response frame from <paramref name="stream"/> and returns its bare PDU
/// (<c>[fc, ...data]</c>), with the leading unit-id and trailing CRC stripped.
@@ -82,58 +104,56 @@ public static class ModbusRtuFraming
public static async Task<byte[]> ReadResponsePduAsync(
Stream stream, byte expectedUnit, CancellationToken ct)
{
// Header: addr(1) + fc(1). The function code selects how many more bytes to read.
var header = new byte[2];
await ReadExactlyAsync(stream, header, ct).ConfigureAwait(false);
var fc = header[1];
// R07:L5: the whole frame is read into ONE pooled buffer and validated in place; only the returned PDU is
// allocated (it used to be five arrays: header, byteCount, rest, the reassembled frame and the PDU).
var frame = ArrayPool<byte>.Shared.Rent(MaxResponseFrame);
try
{
// Header: addr(1) + fc(1). The function code selects how many more bytes to read.
await ReadExactlyAsync(stream, frame.AsMemory(0, 2), ct).ConfigureAwait(false);
var fc = frame[1];
int trailing; // bytes after addr+fc, up to and including the 2 CRC bytes.
if ((fc & 0x80) != 0)
{
// Exception frame: excCode(1) + CRC(2).
trailing = 1 + 2;
}
else if (fc is 0x01 or 0x02 or 0x03 or 0x04 or 0x17)
{
// Read response (and FC23 Read/Write Multiple, whose reply is a read): byteCount(1) then
// byteCount data bytes + CRC(2).
var bc = new byte[1];
await ReadExactlyAsync(stream, bc, ct).ConfigureAwait(false);
var byteCount = bc[0];
var rest = new byte[byteCount + 2];
await ReadExactlyAsync(stream, rest, ct).ConfigureAwait(false);
return ValidateAndStrip(expectedUnit, header, bc, rest);
}
else
{
// Fixed 4-byte echo: correct for the write FCs this driver emits (05/06/0F/10). FC23 is
// variable-length and handled above; any NEW variable-length response FC must be added there
// or it is mis-sized here and surfaces as a desync.
trailing = 4 + 2;
}
int length; // the whole frame, addr through CRC
var already = 2; // bytes of the frame read so far
if ((fc & 0x80) != 0)
{
// Exception frame: excCode(1) + CRC(2).
length = 2 + 1 + 2;
}
else if (fc is 0x01 or 0x02 or 0x03 or 0x04 or 0x17)
{
// Read response (and FC23 Read/Write Multiple, whose reply is a read): byteCount(1) then
// byteCount data bytes + CRC(2).
await ReadExactlyAsync(stream, frame.AsMemory(2, 1), ct).ConfigureAwait(false);
length = 3 + frame[2] + 2;
already = 3;
}
else
{
// Fixed 4-byte echo: correct for the write FCs this driver emits (05/06/0F/10). FC23 is
// variable-length and handled above; any NEW variable-length response FC must be added there
// or it is mis-sized here and surfaces as a desync.
length = 2 + 4 + 2;
}
var tail = new byte[trailing];
await ReadExactlyAsync(stream, tail, ct).ConfigureAwait(false);
return ValidateAndStrip(expectedUnit, header, ReadOnlySpan<byte>.Empty, tail);
await ReadExactlyAsync(stream, frame.AsMemory(already, length - already), ct).ConfigureAwait(false);
return ValidateAndStrip(expectedUnit, frame.AsSpan(0, length));
}
finally
{
ArrayPool<byte>.Shared.Return(frame);
}
}
/// <summary>
/// Assembles the full frame from its pieces, validates the trailing CRC over everything
/// except the CRC bytes, then validates the response address against
/// <paramref name="expectedUnit"/> and strips <c>addr</c> + <c>CRC</c> to return the bare PDU.
/// Validates the trailing CRC over everything except the CRC bytes, then validates the response address
/// against <paramref name="expectedUnit"/> and strips <c>addr</c> + <c>CRC</c> to return the bare PDU.
/// A CRC-valid exception PDU throws <see cref="ModbusException"/>.
/// </summary>
private static byte[] ValidateAndStrip(
byte expectedUnit, ReadOnlySpan<byte> header, ReadOnlySpan<byte> middle, ReadOnlySpan<byte> tail)
private static byte[] ValidateAndStrip(byte expectedUnit, ReadOnlySpan<byte> frame)
{
// Reassemble the whole frame: header(addr,fc) + middle(optional byteCount) + tail.
var frame = new byte[header.Length + middle.Length + tail.Length];
header.CopyTo(frame);
middle.CopyTo(frame.AsSpan(header.Length));
tail.CopyTo(frame.AsSpan(header.Length + middle.Length));
// CRC covers everything except the trailing 2 CRC bytes.
var body = frame.AsSpan(0, frame.Length - 2);
var body = frame[..^2];
var expected = ModbusCrc.Compute(body);
var actual = (ushort)(frame[^2] | (frame[^1] << 8));
if (actual != expected)
@@ -172,12 +192,12 @@ public static class ModbusRtuFraming
/// <see cref="ModbusTransportDesyncException"/> so a length-less RTU frame that arrives
/// short is classified as a desync rather than a bare <see cref="EndOfStreamException"/>.
/// </summary>
private static async Task ReadExactlyAsync(Stream stream, byte[] buf, CancellationToken ct)
private static async Task ReadExactlyAsync(Stream stream, Memory<byte> buf, CancellationToken ct)
{
var read = 0;
while (read < buf.Length)
{
var n = await stream.ReadAsync(buf.AsMemory(read), ct).ConfigureAwait(false);
var n = await stream.ReadAsync(buf[read..], ct).ConfigureAwait(false);
if (n == 0)
throw new ModbusTransportDesyncException(
ModbusTransportDesyncException.DesyncReason.TruncatedFrame,
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs
+12 -3
View File
@@ -1,3 +1,4 @@
using System.Buffers;
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
@@ -150,14 +151,17 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport
var stream = CurrentStream;
if (stream is null) throw new InvalidOperationException("Transport not connected");
// RTU ADU: [unitId] + PDU + CRC — no MBAP header, no TxId.
var adu = ModbusRtuFraming.BuildAdu(unitId, pdu);
// RTU ADU: [unitId] + PDU + CRC — no MBAP header, no TxId. Built in a pooled buffer (R07:L5); the same bytes
// ModbusRtuFraming.BuildAdu produces (pinned by a test).
var aduLength = ModbusRtuFraming.AduLength(pdu.Length);
var adu = ArrayPool<byte>.Shared.Rent(aduLength);
ModbusRtuFraming.WriteAdu(unitId, pdu, adu.AsSpan(0, aduLength));
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(_timeout);
try
{
await stream.WriteAsync(adu.AsMemory(), cts.Token).ConfigureAwait(false);
await stream.WriteAsync(adu.AsMemory(0, aduLength), cts.Token).ConfigureAwait(false);
await stream.FlushAsync(cts.Token).ConfigureAwait(false);
// Framing owns the length-less RTU deframe + CRC + unit-address validation. A CRC-valid
@@ -181,6 +185,11 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport
ModbusTransportDesyncException.DesyncReason.Timeout,
$"Modbus per-op response timeout after {_timeout.TotalMilliseconds:0}ms");
}
finally
{
// The write has completed, or was cancelled and the stream torn down, so nothing still reads it.
ArrayPool<byte>.Shared.Return(adu);
}
}
/// <summary>Asynchronously disposes the transport and underlying socket resources.</summary>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs
+20 -5
View File
@@ -1,3 +1,4 @@
using System.Buffers;
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
@@ -59,6 +60,10 @@ public sealed class ModbusTcpTransport : IModbusTransport
/// <summary>The socket lifecycle this transport composes (test seam for its clock).</summary>
internal ModbusSocketLifecycle Lifecycle => _life;
// R07:L5: the 7-byte MBAP header of the reply being read. SendOnceAsync runs only under _gate (single-flight), so
// one buffer per transport serves every transaction (it used to be a fresh array per reply).
private readonly byte[] _header = new byte[7];
/// <inheritdoc />
public Task ConnectAsync(CancellationToken ct) => _life.ConnectAsync(ct);
@@ -139,11 +144,16 @@ public sealed class ModbusTcpTransport : IModbusTransport
if (stream is null) throw new InvalidOperationException("Transport not connected");
var txId = ++_nextTx;
// MBAP: [TxId(2)][Proto=0(2)][Length(2)][UnitId(1)] + PDU
var adu = new byte[7 + pdu.Length];
// MBAP: [TxId(2)][Proto=0(2)][Length(2)][UnitId(1)] + PDU. Built in a pooled buffer (R07:L5): the request
// never escapes, and the write has completed (or been cancelled and the socket torn down) before the finally
// hands it back. Only the reply PDU, which the caller keeps, is allocated.
// (One async method on purpose: a second awaited helper costs a state-machine box per call, which ate the saving.)
var aduLength = 7 + pdu.Length;
var adu = ArrayPool<byte>.Shared.Rent(aduLength);
adu[0] = (byte)(txId >> 8);
adu[1] = (byte)(txId & 0xFF);
// protocol id already zero
adu[2] = 0; // protocol id
adu[3] = 0;
var len = (ushort)(1 + pdu.Length); // unit id + pdu
adu[4] = (byte)(len >> 8);
adu[5] = (byte)(len & 0xFF);
@@ -154,10 +164,10 @@ public sealed class ModbusTcpTransport : IModbusTransport
cts.CancelAfter(_timeout);
try
{
await stream.WriteAsync(adu.AsMemory(), cts.Token).ConfigureAwait(false);
await stream.WriteAsync(adu.AsMemory(0, aduLength), cts.Token).ConfigureAwait(false);
await stream.FlushAsync(cts.Token).ConfigureAwait(false);
var header = new byte[7];
var header = _header;
await ReadExactlyAsync(stream, header, cts.Token).ConfigureAwait(false);
var respTxId = (ushort)((header[0] << 8) | header[1]);
if (respTxId != txId)
@@ -211,6 +221,11 @@ public sealed class ModbusTcpTransport : IModbusTransport
ModbusTransportDesyncException.DesyncReason.Timeout,
$"Modbus per-op response timeout after {_timeout.TotalMilliseconds:0}ms");
}
finally
{
// The write has completed, or was cancelled and the socket torn down, so nothing still reads it.
ArrayPool<byte>.Shared.Return(adu);
}
}
private static async Task ReadExactlyAsync(Stream s, byte[] buf, CancellationToken ct)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusPooledFramingTests.cs
+62
View File
@@ -0,0 +1,62 @@
using Shouldly;
using Xunit;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests;
/// <summary>
/// R07:L5 (Modbus half): the RTU ADU is now written into a pooled buffer and the response is deframed in one pooled
/// buffer. The wire bytes and the decoded PDUs must be exactly what the allocating code produced, and a reused pooled
/// buffer must never leak a previous frame's bytes into the next result.
/// </summary>
[Trait("Category", "Unit")]
public sealed class ModbusPooledFramingTests
{
/// <summary>The pooled ADU writer matches the reference construction (unit + PDU, CRC appended low byte first).</summary>
/// <param name="hex">The PDU as hex.</param>
[Theory]
[InlineData("0300000001")]
[InlineData("10000A0002040001000200")]
[InlineData("06")]
public void WriteAdu_matches_the_reference_ADU(string hex)
{
var pdu = Convert.FromHexString(hex);
var reference = ModbusCrc.Append([0x11, .. pdu]);
var pooled = new byte[ModbusRtuFraming.AduLength(pdu.Length) + 7]; // oversized, like a rented array
ModbusRtuFraming.WriteAdu(0x11, pdu, pooled.AsSpan(0, ModbusRtuFraming.AduLength(pdu.Length)));
pooled.AsSpan(0, reference.Length).ToArray().ShouldBe(reference);
ModbusRtuFraming.BuildAdu(0x11, pdu).ShouldBe(reference);
}
/// <summary>
/// A long read reply followed by a short echo and an exception frame, each deframed through the pooled buffer:
/// every result is exactly its own PDU (no stale tail from the previous, longer frame).
/// </summary>
[Fact]
public async Task Consecutive_deframes_through_the_pooled_buffer_do_not_leak_bytes()
{
var ct = TestContext.Current.CancellationToken;
byte[] longRead = [0x03, 200, .. Enumerable.Range(0, 200).Select(i => (byte)(i + 1))];
byte[] echo = [0x06, 0x00, 0x01, 0x00, 0x2A];
(await ModbusRtuFraming.ReadResponsePduAsync(new MemoryStream(ModbusRtuFraming.BuildAdu(1, longRead)), 1, ct))
.ShouldBe(longRead);
(await ModbusRtuFraming.ReadResponsePduAsync(new MemoryStream(ModbusRtuFraming.BuildAdu(1, echo)), 1, ct))
.ShouldBe(echo);
var ex = await Should.ThrowAsync<ModbusException>(() =>
ModbusRtuFraming.ReadResponsePduAsync(new MemoryStream(ModbusRtuFraming.BuildAdu(1, [0x83, 0x02])), 1, ct));
ex.ExceptionCode.ShouldBe((byte)0x02);
(await ModbusRtuFraming.ReadResponsePduAsync(new MemoryStream(ModbusRtuFraming.BuildAdu(1, [0x03, 2, 9, 8])), 1, ct))
.ShouldBe(new byte[] { 0x03, 2, 9, 8 });
}
/// <summary>A byte-count-255 read (the largest RTU reply) fits the pooled frame.</summary>
[Fact]
public async Task The_largest_read_reply_deframes()
{
byte[] max = [0x04, 255, .. new byte[255]];
(await ModbusRtuFraming.ReadResponsePduAsync(new MemoryStream(ModbusRtuFraming.BuildAdu(1, max)), 1,
TestContext.Current.CancellationToken)).ShouldBe(max);
}
}