perf(ab-transport): pool CipWriter, reassembly and frame-header buffers
R07:L5 (AB half): CipWriter rents its working buffer from ArrayPool and returns it in ToArray (which now completes the writer; later writes throw). A single-reply Logix read is one copy of the data (no List<byte> + ToArray); a fragmented read reassembles in a pooled buffer. EncapsulationClient reads the frame header into a per-session buffer (reads only happen under the session gate) instead of a new array per reply. Measured per op (scratch A/B, Release): frame encode 784 -> 200 B, Logix read 1994 -> 1655 B, 3-fragment read 11.9 -> 8.2 KB, EIP round trip 2.75 -> 2.15 KB; encode time 41-46 -> 28-39 ms per 500k, other timings within noise; Ab.Transport.Tests 3 s before and after. Modbus transports are left to a follow-up (its lane is still in review). Refs #747 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
@@ -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<byte[]> 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;
|
||||
}
|
||||
|
||||
@@ -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<byte>(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<byte> 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<byte>.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<byte>.Shared.Return(data);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Appends <paramref name="chunk"/> to a pooled buffer, growing it (rent bigger, copy, return) as needed.</summary>
|
||||
private static void Append(ref byte[] buffer, ref int count, ReadOnlySpan<byte> chunk)
|
||||
{
|
||||
if (count + chunk.Length > buffer.Length)
|
||||
{
|
||||
var bigger = ArrayPool<byte>.Shared.Rent(Math.Max(buffer.Length * 2, count + chunk.Length));
|
||||
buffer.AsSpan(0, count).CopyTo(bigger);
|
||||
ArrayPool<byte>.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)";
|
||||
|
||||
@@ -1,17 +1,38 @@
|
||||
using System.Buffers;
|
||||
using System.Buffers.Binary;
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Wire;
|
||||
|
||||
/// <summary>Growable little-endian writer for CIP / EIP frames.</summary>
|
||||
/// <remarks>
|
||||
/// R07:L5: the working buffer is rented from <see cref="ArrayPool{T}.Shared"/> (a 256-byte array per request used
|
||||
/// to be allocated and thrown away) and handed back by <see cref="ToArray"/>, which therefore COMPLETES the writer:
|
||||
/// the result is the caller's own array, a second <see cref="ToArray"/> returns another copy of it, and any further
|
||||
/// write throws <see cref="InvalidOperationException"/> (the buffer belongs to the pool again). A writer abandoned
|
||||
/// before <see cref="ToArray"/> (a builder that threw) simply leaves its buffer to the GC.
|
||||
/// </remarks>
|
||||
public sealed class CipWriter
|
||||
{
|
||||
private byte[] _buf = new byte[256];
|
||||
private const int InitialCapacity = 256;
|
||||
private byte[]? _buf = ArrayPool<byte>.Shared.Rent(InitialCapacity);
|
||||
private byte[]? _result;
|
||||
|
||||
/// <summary>Gets the number of bytes written so far.</summary>
|
||||
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<byte> 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<byte>.Shared.Rent(Math.Max(buf.Length * 2, Length + n));
|
||||
buf.AsSpan(0, Length).CopyTo(bigger);
|
||||
ArrayPool<byte>.Shared.Return(buf);
|
||||
_buf = buf = bigger;
|
||||
}
|
||||
var s = buf.AsSpan(Length, n);
|
||||
Length += n;
|
||||
return s;
|
||||
}
|
||||
@@ -45,14 +66,27 @@ public sealed class CipWriter
|
||||
/// <param name="v">The replacement value.</param>
|
||||
/// <exception cref="ArgumentOutOfRangeException">The two bytes at <paramref name="offset"/> are not already
|
||||
/// written (an offset in the spare buffer capacity would otherwise be patched silently and lost).</exception>
|
||||
/// <exception cref="InvalidOperationException">The writer was completed by <see cref="ToArray"/>.</exception>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>Copies the written bytes to a new array.</summary>
|
||||
/// <returns>The bytes written so far.</returns>
|
||||
public byte[] ToArray() => _buf.AsSpan(0, Length).ToArray();
|
||||
/// <summary>
|
||||
/// Copies the written bytes to a new array and returns the working buffer to the pool, completing the writer
|
||||
/// (see the remarks on <see cref="CipWriter"/>).
|
||||
/// </summary>
|
||||
/// <returns>The bytes written, in an array the caller owns.</returns>
|
||||
public byte[] ToArray()
|
||||
{
|
||||
if (_result is { } done) return (byte[])done.Clone();
|
||||
var buf = Buffer;
|
||||
_result = buf.AsSpan(0, Length).ToArray();
|
||||
_buf = null;
|
||||
ArrayPool<byte>.Shared.Return(buf);
|
||||
return _result;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -25,3 +25,60 @@ public sealed class CipWriterTests
|
||||
w.ToArray().ShouldBe(new byte[] { 0, 0, 7 }); // nothing was patched
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// R07:L5: <see cref="CipWriter"/> works in a buffer rented from <see cref="System.Buffers.ArrayPool{T}"/> and hands
|
||||
/// it back on <see cref="CipWriter.ToArray"/>. 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.
|
||||
/// </summary>
|
||||
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<InvalidOperationException>(() => w.U8(2));
|
||||
Should.Throw<InvalidOperationException>(() => 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 });
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user