From 7e958cd750dba3ebb531a5bf2458f4f2f8dbf034 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 17:28:53 -0400 Subject: [PATCH] perf(modbus): pooled request ADUs, a reused MBAP header and single-buffer RTU deframing 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 --- .../ModbusRtuFraming.cs | 118 ++++++++++-------- .../ModbusRtuOverTcpTransport.cs | 15 ++- .../ModbusTcpTransport.cs | 25 +++- .../ModbusPooledFramingTests.cs | 62 +++++++++ 4 files changed, 163 insertions(+), 57 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusPooledFramingTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuFraming.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuFraming.cs index 9fd9c682..bec5055d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuFraming.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuFraming.cs @@ -1,3 +1,4 @@ +using System.Buffers; namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; /// @@ -55,12 +56,33 @@ public static class ModbusRtuFraming /// A new array: [unitId, ...pdu, crcLo, crcHi]. public static byte[] BuildAdu(byte unitId, ReadOnlySpan 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; } + /// The length of the RTU ADU for a PDU of bytes (unit id + PDU + CRC). + /// The bare PDU length. + /// The ADU length. + internal static int AduLength(int pduLength) => 1 + pduLength + 2; + + /// Writes the RTU ADU into (exactly bytes). + /// The RTU slave/unit address. + /// The bare PDU. + /// The ADU buffer, exactly (.Length) long. + internal static void WriteAdu(byte unitId, ReadOnlySpan pdu, Span 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); + } + + /// The largest RTU response frame: addr + fc + byteCount + 255 data bytes + CRC. + private const int MaxResponseFrame = 3 + 255 + 2; + /// /// Reads a single RTU response frame from and returns its bare PDU /// ([fc, ...data]), with the leading unit-id and trailing CRC stripped. @@ -82,58 +104,56 @@ public static class ModbusRtuFraming public static async Task 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.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.Empty, tail); + await ReadExactlyAsync(stream, frame.AsMemory(already, length - already), ct).ConfigureAwait(false); + return ValidateAndStrip(expectedUnit, frame.AsSpan(0, length)); + } + finally + { + ArrayPool.Shared.Return(frame); + } } /// - /// Assembles the full frame from its pieces, validates the trailing CRC over everything - /// except the CRC bytes, then validates the response address against - /// and strips addr + CRC to return the bare PDU. + /// Validates the trailing CRC over everything except the CRC bytes, then validates the response address + /// against and strips addr + CRC to return the bare PDU. /// A CRC-valid exception PDU throws . /// - private static byte[] ValidateAndStrip( - byte expectedUnit, ReadOnlySpan header, ReadOnlySpan middle, ReadOnlySpan tail) + private static byte[] ValidateAndStrip(byte expectedUnit, ReadOnlySpan 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 /// so a length-less RTU frame that arrives /// short is classified as a desync rather than a bare . /// - private static async Task ReadExactlyAsync(Stream stream, byte[] buf, CancellationToken ct) + private static async Task ReadExactlyAsync(Stream stream, Memory 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, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs index 7dcc5dff..e2f4047e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs @@ -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.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.Shared.Return(adu); + } } /// Asynchronously disposes the transport and underlying socket resources. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs index f21ce678..6889b1e2 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs @@ -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 /// The socket lifecycle this transport composes (test seam for its clock). 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]; + /// 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.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.Shared.Return(adu); + } } private static async Task ReadExactlyAsync(Stream s, byte[] buf, CancellationToken ct) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusPooledFramingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusPooledFramingTests.cs new file mode 100644 index 00000000..6cf0450b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusPooledFramingTests.cs @@ -0,0 +1,62 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// 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. +/// +[Trait("Category", "Unit")] +public sealed class ModbusPooledFramingTests +{ + /// The pooled ADU writer matches the reference construction (unit + PDU, CRC appended low byte first). + /// The PDU as hex. + [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); + } + + /// + /// 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). + /// + [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(() => + 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 }); + } + + /// A byte-count-255 read (the largest RTU reply) fits the pooled frame. + [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); + } +}