perf(worker): pooled frame buffers — brings the net48 codec up to the gateway side's GWC-30 pattern

This commit is contained in:
Joseph Doherty
2026-08-15 13:28:29 -04:00
parent 13df92fbd8
commit f56798aeb9
2 changed files with 31 additions and 11 deletions
@@ -9,11 +9,21 @@ using ZB.MOM.WW.MxGateway.Contracts.Proto;
namespace ZB.MOM.WW.MxGateway.Worker.Ipc; namespace ZB.MOM.WW.MxGateway.Worker.Ipc;
/// <summary>Reads length-prefixed WorkerEnvelope protobuf frames from a stream.</summary> /// <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>WorkerPipeSession</c>, with the startup handshake read
/// 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>Initializes the reader with a stream and protocol options.</summary> /// <summary>Initializes the reader with a stream and protocol options.</summary>
/// <param name="stream">Stream to read frames from.</param> /// <param name="stream">Stream to read frames from.</param>
/// <param name="options">Protocol options for frame validation.</param> /// <param name="options">Protocol options for frame validation.</param>
@@ -30,10 +40,9 @@ public sealed class WorkerFrameReader
/// <returns>The validated <see cref="WorkerEnvelope"/> read from the stream.</returns> /// <returns>The validated <see cref="WorkerEnvelope"/> read from the stream.</returns>
public async Task<WorkerEnvelope> ReadAsync(CancellationToken cancellationToken = default) public async Task<WorkerEnvelope> ReadAsync(CancellationToken cancellationToken = default)
{ {
byte[] lengthPrefix = new byte[sizeof(uint)]; await ReadExactlyOrThrowAsync(_lengthPrefix, sizeof(uint), cancellationToken).ConfigureAwait(false);
await ReadExactlyOrThrowAsync(lengthPrefix, lengthPrefix.Length, cancellationToken).ConfigureAwait(false);
uint payloadLength = ReadUInt32LittleEndian(lengthPrefix); uint payloadLength = ReadUInt32LittleEndian(_lengthPrefix);
if (payloadLength == 0) if (payloadLength == 0)
{ {
throw new WorkerFrameProtocolException( throw new WorkerFrameProtocolException(
@@ -1,4 +1,5 @@
using System; using System;
using System.Buffers;
using System.Collections.Generic; using System.Collections.Generic;
using System.IO; using System.IO;
using System.Runtime.ExceptionServices; using System.Runtime.ExceptionServices;
@@ -459,17 +460,27 @@ public sealed class WorkerFrameWriter
_nextSequence = candidateSequence; _nextSequence = candidateSequence;
// Serialize once into a single buffer that carries the 4-byte length prefix followed by the // Serialize once into a single pooled buffer that carries the 4-byte length prefix followed
// payload, then issue one stream write. This avoids a second serialization pass, a separate // by the payload, then issue one stream write. This avoids a second serialization pass, a
// prefix array, and a separate prefix write. The flush is deferred to the end of the drained // separate prefix array, a separate prefix write, and any per-frame heap allocation — the
// batch (see DrainQueuedFramesAsync) so a burst of frames shares one flush. // gateway-side writer's GWC-30 shape, now matched on the net48 side. The rented buffer may
// be larger than requested, so only the first frameLength bytes are ever written. The buffer
// is returned only after this frame's write has completed; the batch flush deferred to the
// end of the drain (see DrainQueuedFramesAsync) does not read from it.
int frameLength = sizeof(uint) + payloadLength; int frameLength = sizeof(uint) + payloadLength;
byte[] frame = new byte[frameLength]; byte[] frame = ArrayPool<byte>.Shared.Rent(frameLength);
try
{
WriteUInt32LittleEndian(frame, (uint)payloadLength); WriteUInt32LittleEndian(frame, (uint)payloadLength);
envelope.WriteTo(new Span<byte>(frame, sizeof(uint), payloadLength)); envelope.WriteTo(new Span<byte>(frame, sizeof(uint), payloadLength));
await _stream.WriteAsync(frame, 0, frameLength, CancellationToken.None).ConfigureAwait(false); await _stream.WriteAsync(frame, 0, frameLength, CancellationToken.None).ConfigureAwait(false);
} }
finally
{
ArrayPool<byte>.Shared.Return(frame);
}
}
private static void WriteUInt32LittleEndian( private static void WriteUInt32LittleEndian(
byte[] buffer, byte[] buffer,