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
This commit is contained in:
Joseph Doherty
2026-07-09 14:58:09 -04:00
parent 34aede6767
commit baae59ce3b
3 changed files with 70 additions and 13 deletions
@@ -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<byte>.Shared.Rent(length);
WorkerEnvelope envelope;
try
{
envelope = WorkerEnvelope.Parser.ParseFrom(payload);
await ReadExactlyOrThrowAsync(new Memory<byte>(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<byte>.Shared.Return(payload);
}
WorkerEnvelopeValidator.Validate(envelope, _options);
@@ -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<byte>.Shared.Rent(frameLength);
try
{
BinaryPrimitives.WriteUInt32LittleEndian(frame, (uint)payloadLength);
envelope.WriteTo(new Span<byte>(frame, sizeof(uint), payloadLength));
await _stream.WriteAsync(lengthPrefix, cancellationToken).ConfigureAwait(false);
await _stream.WriteAsync(envelope.ToByteArray(), cancellationToken).ConfigureAwait(false);
await _stream
.WriteAsync(new ReadOnlyMemory<byte>(frame, 0, frameLength), cancellationToken)
.ConfigureAwait(false);
}
finally
{
ArrayPool<byte>.Shared.Return(frame);
}
}
}
@@ -29,6 +29,32 @@ public sealed class WorkerFrameProtocolTests
Assert.Equal(original, parsed);
}
/// <summary>Verifies that a large payload near the message cap round-trips through the pooled write/read path.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>Verifies that reading a frame with partial reads reassembles the frame correctly.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]