diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs index 8a03900..041e879 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameReader.cs @@ -9,11 +9,21 @@ using ZB.MOM.WW.MxGateway.Contracts.Proto; namespace ZB.MOM.WW.MxGateway.Worker.Ipc; /// Reads length-prefixed WorkerEnvelope protobuf frames from a stream. +/// +/// 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 WorkerPipeSession, with the startup handshake read +/// completing before the loop starts. +/// public sealed class WorkerFrameReader { private readonly WorkerFrameProtocolOptions _options; 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)]; + /// Initializes the reader with a stream and protocol options. /// Stream to read frames from. /// Protocol options for frame validation. @@ -30,10 +40,9 @@ public sealed class WorkerFrameReader /// The validated read from the stream. public async Task ReadAsync(CancellationToken cancellationToken = default) { - byte[] lengthPrefix = new byte[sizeof(uint)]; - await ReadExactlyOrThrowAsync(lengthPrefix, lengthPrefix.Length, cancellationToken).ConfigureAwait(false); + await ReadExactlyOrThrowAsync(_lengthPrefix, sizeof(uint), cancellationToken).ConfigureAwait(false); - uint payloadLength = ReadUInt32LittleEndian(lengthPrefix); + uint payloadLength = ReadUInt32LittleEndian(_lengthPrefix); if (payloadLength == 0) { throw new WorkerFrameProtocolException( diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index 7fab801..cfb20e7 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; using System.Collections.Generic; using System.IO; using System.Runtime.ExceptionServices; @@ -459,16 +460,26 @@ public sealed class WorkerFrameWriter _nextSequence = candidateSequence; - // Serialize once into a single buffer that carries the 4-byte length prefix followed by the - // payload, then issue one stream write. This avoids a second serialization pass, a separate - // prefix array, and a separate prefix write. The flush is deferred to the end of the drained - // batch (see DrainQueuedFramesAsync) so a burst of frames shares one flush. + // Serialize once into a single pooled buffer that carries the 4-byte length prefix followed + // by the payload, then issue one stream write. This avoids a second serialization pass, a + // separate prefix array, a separate prefix write, and any per-frame heap allocation — the + // 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; - byte[] frame = new byte[frameLength]; - WriteUInt32LittleEndian(frame, (uint)payloadLength); - envelope.WriteTo(new Span(frame, sizeof(uint), payloadLength)); + byte[] frame = ArrayPool.Shared.Rent(frameLength); + try + { + WriteUInt32LittleEndian(frame, (uint)payloadLength); + envelope.WriteTo(new Span(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.Shared.Return(frame); + } } private static void WriteUInt32LittleEndian(