diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs index b746745e..8c911779 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Eip/EncapsulationClient.cs @@ -47,6 +47,10 @@ public sealed class EncapsulationClient : IAsyncDisposable private readonly SemaphoreSlim _gate = new(1, 1); private long _ctx; private ushort _seq; // connected sequence count; only touched under _gate + + // R07:L5: the 24-byte header of the frame being read. Frames are read only by RunExchangeAsync, which holds _gate + // for the whole exchange, so one buffer per session serves every read (it used to be a fresh array per reply). + private readonly byte[] _head = new byte[EncapsulationHeader.Size]; private volatile bool _faulted; private int _disposed; @@ -282,11 +286,12 @@ public sealed class EncapsulationClient : IAsyncDisposable private async Task ReadFrameAsync(CancellationToken ct) { - var head = new byte[EncapsulationHeader.Size]; - await _stream.ReadExactlyAsync(head, ct).ConfigureAwait(false); - var len = head[2] | (head[3] << 8); + // Only called under _gate (RunExchangeAsync), so the shared header buffer is never read concurrently. The frame + // itself is allocated exactly: the matched reply's CIP payload is returned as a slice of it and outlives the call. + await _stream.ReadExactlyAsync(_head, ct).ConfigureAwait(false); + var len = _head[2] | (_head[3] << 8); var frame = new byte[EncapsulationHeader.Size + len]; - head.CopyTo(frame, 0); + _head.CopyTo(frame, 0); await _stream.ReadExactlyAsync(frame.AsMemory(EncapsulationHeader.Size), ct).ConfigureAwait(false); return frame; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs index ba63cc38..85987cce 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Logix/LogixTagClient.cs @@ -1,3 +1,4 @@ +using System.Buffers; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Logix; @@ -52,34 +53,61 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann var reply = await _scheduler.SendAsync(LogixServices.ReadTagFragmented(path, elements, 0), ct, expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, 0)).ConfigureAwait(false); var (type, handle, first) = LogixServices.ParseReadReply(reply); // throws on any status but 0x00/0x06 - var data = new List(first.Length); - data.AddRange(first.Span); - var lastChunk = first.Length; - while (reply.GeneralStatus == StatusPartial) + // R07:L5: the common single-reply read is one copy of the reply data (it used to go through a List and a + // second ToArray); a fragmented read reassembles in a pooled buffer and copies out once at the end. + if (reply.GeneralStatus != StatusPartial) return (type, handle, first.ToArray()); + + var data = ArrayPool.Shared.Rent(Math.Max(first.Length * 4, 1024)); + var count = 0; + try { - if (lastChunk == 0) - throw new AbTransportException(AbStatus.BadReply, "Partial read reply carried no data; the fragment loop cannot progress."); - reply = await _scheduler.SendAsync( - LogixServices.ReadTagFragmented(path, elements, (uint)data.Count), ct, - expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, data.Count)).ConfigureAwait(false); - var (fragmentType, fragmentHandle, chunk) = LogixServices.ParseReadReply(reply); - if (fragmentType != type) - throw new AbTransportException(AbStatus.BadReply, - $"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}."); - // A mid-read change of the structure handle means the tag's type changed under us (a - // download / online edit); stitching the halves would yield a value of neither type. - if (fragmentHandle != handle) - throw new AbTransportException(AbStatus.BadReply, - $"Read fragment structure handle {FormatHandle(fragmentHandle)} differs from the first reply's {FormatHandle(handle)}; the tag changed mid-read."); - data.AddRange(chunk.Span); - lastChunk = chunk.Length; - if (data.Count > MaxReadBytes) - throw new AbTransportException(AbStatus.TooLarge, - $"Fragmented read exceeded {MaxReadBytes} bytes; the controller kept answering Partial."); - } + Append(ref data, ref count, first.Span); + var lastChunk = first.Length; - return (type, handle, data.ToArray()); + while (reply.GeneralStatus == StatusPartial) + { + if (lastChunk == 0) + throw new AbTransportException(AbStatus.BadReply, "Partial read reply carried no data; the fragment loop cannot progress."); + reply = await _scheduler.SendAsync( + LogixServices.ReadTagFragmented(path, elements, (uint)count), ct, + expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, count)).ConfigureAwait(false); + var (fragmentType, fragmentHandle, chunk) = LogixServices.ParseReadReply(reply); + if (fragmentType != type) + throw new AbTransportException(AbStatus.BadReply, + $"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}."); + // A mid-read change of the structure handle means the tag's type changed under us (a + // download / online edit); stitching the halves would yield a value of neither type. + if (fragmentHandle != handle) + throw new AbTransportException(AbStatus.BadReply, + $"Read fragment structure handle {FormatHandle(fragmentHandle)} differs from the first reply's {FormatHandle(handle)}; the tag changed mid-read."); + Append(ref data, ref count, chunk.Span); + lastChunk = chunk.Length; + if (count > MaxReadBytes) + throw new AbTransportException(AbStatus.TooLarge, + $"Fragmented read exceeded {MaxReadBytes} bytes; the controller kept answering Partial."); + } + + return (type, handle, data.AsSpan(0, count).ToArray()); + } + finally + { + ArrayPool.Shared.Return(data); + } + } + + /// Appends to a pooled buffer, growing it (rent bigger, copy, return) as needed. + private static void Append(ref byte[] buffer, ref int count, ReadOnlySpan chunk) + { + if (count + chunk.Length > buffer.Length) + { + var bigger = ArrayPool.Shared.Rent(Math.Max(buffer.Length * 2, count + chunk.Length)); + buffer.AsSpan(0, count).CopyTo(bigger); + ArrayPool.Shared.Return(buffer); + buffer = bigger; + } + chunk.CopyTo(buffer.AsSpan(count)); + count += chunk.Length; } private static string FormatHandle(ushort? handle) => handle is { } h ? $"0x{h:X4}" : "(none)"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Wire/CipWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Wire/CipWriter.cs index 8c86f186..94b77a79 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Wire/CipWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Wire/CipWriter.cs @@ -1,17 +1,38 @@ +using System.Buffers; using System.Buffers.Binary; namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Wire; /// Growable little-endian writer for CIP / EIP frames. +/// +/// R07:L5: the working buffer is rented from (a 256-byte array per request used +/// to be allocated and thrown away) and handed back by , which therefore COMPLETES the writer: +/// the result is the caller's own array, a second returns another copy of it, and any further +/// write throws (the buffer belongs to the pool again). A writer abandoned +/// before (a builder that threw) simply leaves its buffer to the GC. +/// public sealed class CipWriter { - private byte[] _buf = new byte[256]; + private const int InitialCapacity = 256; + private byte[]? _buf = ArrayPool.Shared.Rent(InitialCapacity); + private byte[]? _result; + /// Gets the number of bytes written so far. public int Length { get; private set; } + private byte[] Buffer => _buf ?? throw new InvalidOperationException( + "This CipWriter was completed by ToArray(); its buffer has been returned to the pool."); + private Span Take(int n) { - if (Length + n > _buf.Length) Array.Resize(ref _buf, Math.Max(_buf.Length * 2, Length + n)); - var s = _buf.AsSpan(Length, n); + var buf = Buffer; + if (Length + n > buf.Length) + { + var bigger = ArrayPool.Shared.Rent(Math.Max(buf.Length * 2, Length + n)); + buf.AsSpan(0, Length).CopyTo(bigger); + ArrayPool.Shared.Return(buf); + _buf = buf = bigger; + } + var s = buf.AsSpan(Length, n); Length += n; return s; } @@ -45,14 +66,27 @@ public sealed class CipWriter /// The replacement value. /// The two bytes at are not already /// written (an offset in the spare buffer capacity would otherwise be patched silently and lost). + /// The writer was completed by . public void PatchU16(int offset, ushort v) { + var buf = Buffer; if (offset < 0 || offset > Length - 2) throw new ArgumentOutOfRangeException(nameof(offset), offset, $"Cannot patch 2 bytes at {offset}: only {Length} bytes written."); - BinaryPrimitives.WriteUInt16LittleEndian(_buf.AsSpan(offset, 2), v); + BinaryPrimitives.WriteUInt16LittleEndian(buf.AsSpan(offset, 2), v); } - /// Copies the written bytes to a new array. - /// The bytes written so far. - public byte[] ToArray() => _buf.AsSpan(0, Length).ToArray(); + /// + /// Copies the written bytes to a new array and returns the working buffer to the pool, completing the writer + /// (see the remarks on ). + /// + /// The bytes written, in an array the caller owns. + public byte[] ToArray() + { + if (_result is { } done) return (byte[])done.Clone(); + var buf = Buffer; + _result = buf.AsSpan(0, Length).ToArray(); + _buf = null; + ArrayPool.Shared.Return(buf); + return _result; + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Wire/CipWriterTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Wire/CipWriterTests.cs index d196d294..cf4a1965 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Wire/CipWriterTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Wire/CipWriterTests.cs @@ -25,3 +25,60 @@ public sealed class CipWriterTests w.ToArray().ShouldBe(new byte[] { 0, 0, 7 }); // nothing was patched } } + +/// +/// R07:L5: works in a buffer rented from and hands +/// it back on . The returned array must be the caller's own (a later writer reusing +/// the pooled buffer cannot change it), growth must keep every byte, and a writer used after ToArray must refuse +/// rather than write into a buffer it no longer owns. +/// +public sealed class CipWriterPoolingTests +{ + [Fact] + public void Growth_past_the_initial_buffer_keeps_every_byte() + { + var w = new CipWriter(); + for (var i = 0; i < 1000; i++) w.U8((byte)i); + var bytes = w.ToArray(); + bytes.Length.ShouldBe(1000); + bytes.Select((b, i) => b == (byte)i).ShouldAllBe(ok => ok); + } + + [Fact] + public void A_returned_array_is_not_affected_by_a_later_writer_reusing_the_pooled_buffer() + { + var first = new CipWriter().U32(0xAABBCCDD).ToArray(); + for (var i = 0; i < 50; i++) new CipWriter().U32(0x11111111).U32(0x22222222).ToArray(); + first.ShouldBe(new byte[] { 0xDD, 0xCC, 0xBB, 0xAA }); + } + + [Fact] + public void ToArray_twice_returns_equal_independent_copies() + { + var w = new CipWriter().U16(0x0102); + var a = w.ToArray(); + var b = w.ToArray(); + b.ShouldBe(a); + b.ShouldNotBeSameAs(a); + } + + [Fact] + public void Writing_after_ToArray_is_refused() + { + var w = new CipWriter().U8(1); + _ = w.ToArray(); + Should.Throw(() => w.U8(2)); + Should.Throw(() => w.PatchU16(0, 1)); + w.Length.ShouldBe(1); + } + + [Fact] + public void Nested_writers_produce_independent_frames() + { + // The Execute PCCC / encapsulation builders build an inner writer while an outer one is live. + var outer = new CipWriter().U8(0xAA); + var inner = new CipWriter().U8(0x01).U8(0x02).ToArray(); + var frame = outer.Bytes(inner).U8(0xBB).ToArray(); + frame.ShouldBe(new byte[] { 0xAA, 0x01, 0x02, 0xBB }); + } +}