fix(GWC-30): reuse the frame reader's length-prefix scratch buffer
ReadAsync allocated a fresh 4-byte array per inbound frame; the GWC-08 pass pooled the payload buffer but left the prefix. Replaced with a per-instance scratch field — the reader is single-consumer by construction (one read loop per WorkerClient, handshake reads complete before the loop starts), so a per-instance buffer is safe and the non-reentrancy that makes it safe is now stated on the class. Pooling four bytes via ArrayPool would cost more than the allocation it saves. Tests: WorkerFrameProtocolTests.ReadAsync_WithMultipleFramesOnOneReader_ ParsesEveryFrame reads five frames of differing payload length through one reader, so a stale prefix carried between calls would misparse.
This commit is contained in:
@@ -5,11 +5,24 @@ using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
|||||||
|
|
||||||
namespace ZB.MOM.WW.MxGateway.Server.Workers;
|
namespace ZB.MOM.WW.MxGateway.Server.Workers;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Reads length-prefixed WorkerEnvelope protobuf frames from a stream.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// <see cref="ReadAsync"/> is not reentrant: the reader keeps a per-instance length-prefix scratch
|
||||||
|
/// buffer, so exactly one consumer may be inside a read at a time. That matches how the reader is
|
||||||
|
/// used — a single read loop per <c>WorkerClient</c>, with handshake reads completing before the
|
||||||
|
/// loop starts.
|
||||||
|
/// </remarks>
|
||||||
public sealed class WorkerFrameReader
|
public sealed class WorkerFrameReader
|
||||||
{
|
{
|
||||||
private readonly WorkerFrameProtocolOptions _options;
|
private readonly WorkerFrameProtocolOptions _options;
|
||||||
private readonly Stream _stream;
|
private readonly Stream _stream;
|
||||||
|
|
||||||
|
// Reused across frames rather than allocated per read (GWC-30). Safe because ReadAsync is
|
||||||
|
// single-consumer by construction; the prefix is fully overwritten by every read.
|
||||||
|
private readonly byte[] _lengthPrefix = new byte[sizeof(uint)];
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of <see cref="WorkerFrameReader"/>.
|
/// Initializes a new instance of <see cref="WorkerFrameReader"/>.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -30,10 +43,9 @@ public sealed class WorkerFrameReader
|
|||||||
/// <returns>Parsed worker envelope.</returns>
|
/// <returns>Parsed worker envelope.</returns>
|
||||||
public async ValueTask<WorkerEnvelope> ReadAsync(CancellationToken cancellationToken = default)
|
public async ValueTask<WorkerEnvelope> ReadAsync(CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
byte[] lengthPrefix = new byte[sizeof(uint)];
|
await ReadExactlyOrThrowAsync(_lengthPrefix, cancellationToken).ConfigureAwait(false);
|
||||||
await ReadExactlyOrThrowAsync(lengthPrefix, cancellationToken).ConfigureAwait(false);
|
|
||||||
|
|
||||||
uint payloadLength = BinaryPrimitives.ReadUInt32LittleEndian(lengthPrefix);
|
uint payloadLength = BinaryPrimitives.ReadUInt32LittleEndian(_lengthPrefix);
|
||||||
if (payloadLength == 0)
|
if (payloadLength == 0)
|
||||||
{
|
{
|
||||||
throw new WorkerFrameProtocolException(
|
throw new WorkerFrameProtocolException(
|
||||||
|
|||||||
@@ -55,6 +55,41 @@ public sealed class WorkerFrameProtocolTests
|
|||||||
Assert.Equal(original, parsed);
|
Assert.Equal(original, parsed);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// One reader instance reads many frames in sequence. The reader reuses a single length-prefix
|
||||||
|
/// scratch buffer across calls (GWC-30), so varying the payload length frame to frame proves the
|
||||||
|
/// reused buffer is fully overwritten each time rather than carrying a stale prefix forward.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task ReadAsync_WithMultipleFramesOnOneReader_ParsesEveryFrame()
|
||||||
|
{
|
||||||
|
const int frameCount = 5;
|
||||||
|
WorkerFrameProtocolOptions options = new(SessionId);
|
||||||
|
await using MemoryStream stream = new();
|
||||||
|
WorkerFrameWriter writer = new(stream, options);
|
||||||
|
|
||||||
|
List<WorkerEnvelope> originals = [];
|
||||||
|
for (int index = 1; index <= frameCount; index++)
|
||||||
|
{
|
||||||
|
WorkerEnvelope envelope = CreateEnvelope();
|
||||||
|
envelope.Sequence = (ulong)index;
|
||||||
|
|
||||||
|
// Differing payload lengths so a stale length prefix would misparse rather than pass.
|
||||||
|
envelope.WorkerHello.WorkerVersion = new string('v', index * 37);
|
||||||
|
originals.Add(envelope);
|
||||||
|
await writer.WriteAsync(envelope);
|
||||||
|
}
|
||||||
|
|
||||||
|
stream.Position = 0;
|
||||||
|
WorkerFrameReader reader = new(stream, options);
|
||||||
|
foreach (WorkerEnvelope original in originals)
|
||||||
|
{
|
||||||
|
WorkerEnvelope parsed = await reader.ReadAsync();
|
||||||
|
Assert.Equal(original, parsed);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Verifies that reading a frame with partial reads reassembles the frame correctly.</summary>
|
/// <summary>Verifies that reading a frame with partial reads reassembles the frame correctly.</summary>
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
Reference in New Issue
Block a user