From baae59ce3b96cb6816c282dbb47d0c31a40b09db Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 14:58:09 -0400 Subject: [PATCH] perf(gateway): pool+coalesce worker pipe frame I/O (GWC-08, IPC-13, IPC-14) Gateway server-side named-pipe frame path, wire format byte-identical: - Writer (GWC-08/IPC-13): serialize once into a single ArrayPool-rented buffer holding the 4-byte LE length prefix + payload, then one WriteAsync. Removes the second serialization pass (ToByteArray re-ran CalculateSize), the separate prefix array + second stream write, and per-frame heap allocation. - Reader (IPC-14): rent the payload buffer from ArrayPool instead of new byte[] per frame; read/parse bounded to the real length, return in finally. Matches the worker side which already pools. - Correctness: length tracked separately from (larger) pool capacity; every rented buffer returned exactly once in finally incl. the malformed-protobuf path; ParseFrom copies into the message so the envelope never aliases the returned buffer. Cross-checked LE-prefix framing agrees across gateway+worker writer/reader. Server build clean (0 warnings); WorkerFrameProtocol tests 10/10 incl. a new large-payload-near-cap round-trip (forces pool buffer > frame). The 2 failing WorkerClient/FakeWorkerHarness tests are the known macOS UDS 104-char path limit, not a regression. Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW --- .../Workers/WorkerFrameReader.cs | 33 ++++++++++++++----- .../Workers/WorkerFrameWriter.cs | 24 +++++++++++--- .../Workers/WorkerFrameProtocolTests.cs | 26 +++++++++++++++ 3 files changed, 70 insertions(+), 13 deletions(-) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameReader.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameReader.cs index 076d5e9..95cbe13 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameReader.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameReader.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Buffers.Binary; using Google.Protobuf; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -47,20 +48,34 @@ public sealed class WorkerFrameReader $"Worker frame payload length {payloadLength} exceeds the configured maximum of {_options.MaxMessageBytes} bytes."); } - byte[] payload = new byte[payloadLength]; - await ReadExactlyOrThrowAsync(payload, cancellationToken).ConfigureAwait(false); - + // Rent the payload buffer from the shared pool rather than allocating a fresh byte[] per + // frame; large event frames near the cap would otherwise allocate an LOH buffer each time + // (IPC-14). ParseFrom copies whatever it needs into the parsed message, so the rented buffer + // can be returned as soon as parsing completes without the envelope aliasing it. The rented + // buffer may be larger than requested, so the read and parse are bounded to length. + int length = checked((int)payloadLength); + byte[] payload = ArrayPool.Shared.Rent(length); WorkerEnvelope envelope; try { - envelope = WorkerEnvelope.Parser.ParseFrom(payload); + await ReadExactlyOrThrowAsync(new Memory(payload, 0, length), cancellationToken) + .ConfigureAwait(false); + + try + { + envelope = WorkerEnvelope.Parser.ParseFrom(payload, 0, length); + } + catch (InvalidProtocolBufferException exception) + { + throw new WorkerFrameProtocolException( + WorkerFrameProtocolErrorCode.InvalidEnvelope, + "Worker frame payload is not a valid WorkerEnvelope protobuf message.", + exception); + } } - catch (InvalidProtocolBufferException exception) + finally { - throw new WorkerFrameProtocolException( - WorkerFrameProtocolErrorCode.InvalidEnvelope, - "Worker frame payload is not a valid WorkerEnvelope protobuf message.", - exception); + ArrayPool.Shared.Return(payload); } WorkerEnvelopeValidator.Validate(envelope, _options); diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameWriter.cs index 728a611..124e58f 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerFrameWriter.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Buffers.Binary; using Google.Protobuf; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -53,10 +54,25 @@ public sealed class WorkerFrameWriter $"Worker envelope payload length {payloadLength} exceeds the configured maximum of {_options.MaxMessageBytes} bytes."); } - byte[] lengthPrefix = new byte[sizeof(uint)]; - BinaryPrimitives.WriteUInt32LittleEndian(lengthPrefix, (uint)payloadLength); + // Serialize once into a single pooled buffer that carries the 4-byte little-endian length + // prefix followed by the payload, then issue one stream write. This avoids a second + // serialization pass (ToByteArray re-runs CalculateSize), a separate prefix array and its + // own write, and any per-frame heap allocation (GWC-08, IPC-13). The rented buffer may be + // larger than requested, so only the first frameLength bytes are ever written. + int frameLength = sizeof(uint) + payloadLength; + byte[] frame = ArrayPool.Shared.Rent(frameLength); + try + { + BinaryPrimitives.WriteUInt32LittleEndian(frame, (uint)payloadLength); + envelope.WriteTo(new Span(frame, sizeof(uint), payloadLength)); - await _stream.WriteAsync(lengthPrefix, cancellationToken).ConfigureAwait(false); - await _stream.WriteAsync(envelope.ToByteArray(), cancellationToken).ConfigureAwait(false); + await _stream + .WriteAsync(new ReadOnlyMemory(frame, 0, frameLength), cancellationToken) + .ConfigureAwait(false); + } + finally + { + ArrayPool.Shared.Return(frame); + } } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerFrameProtocolTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerFrameProtocolTests.cs index ed26f6d..910e77c 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerFrameProtocolTests.cs @@ -29,6 +29,32 @@ public sealed class WorkerFrameProtocolTests Assert.Equal(original, parsed); } + /// Verifies that a large payload near the message cap round-trips through the pooled write/read path. + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAndReadAsync_WithLargePayloadNearMax_RoundTripsFrame() + { + // A payload comfortably larger than an ArrayPool bucket boundary so the rented buffer is + // larger than the exact frame length, exercising the length-bounded read and parse. + const int maxMessageBytes = 1024 * 1024; + WorkerFrameProtocolOptions options = + new(SessionId, GatewayContractInfo.WorkerProtocolVersion, maxMessageBytes); + await using MemoryStream stream = new(); + + WorkerEnvelope original = CreateEnvelope(); + original.WorkerHello.WorkerVersion = new string('x', maxMessageBytes - 4096); + Assert.InRange(original.CalculateSize(), 1, maxMessageBytes); + + WorkerFrameWriter writer = new(stream, options); + await writer.WriteAsync(original); + stream.Position = 0; + + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope parsed = await reader.ReadAsync(); + + Assert.Equal(original, parsed); + } + /// Verifies that reading a frame with partial reads reassembles the frame correctly. /// A task that represents the asynchronous operation. [Fact]