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); + } +}