Zero-Allocation Event Processing in .NET 10: Building 100k+ RPS Stream Processors with System.IO.Pipelines & Bounded Channels

// The classic, allocation-heavy socket reader anti-pattern
byte[] buffer = new byte[4096];
while ((bytesRead = await networkStream.ReadAsync(buffer, 0, buffer.Length)) > 0)
{
    // 1. Allocates a new byte array slice on every read
    byte[] payload = new byte[bytesRead];
    Buffer.BlockCopy(buffer, 0, payload, 0, bytesRead);
    
    // 2. Allocates a UTF-8 string on the heap for framing
    string rawMessage = Encoding.UTF8.GetString(payload);
    
    // 3. Allocates substring arrays during delimiter splitting
    string[] tokens = rawMessage.Split('\n');
    
    // 4. Passes string tokens downstream via unbounded Task queues
    _ = Task.Run(() => ProcessMessage(tokens));
}

graph TD
    A[Incoming TCP / Kafka Socket] -->|Zero-Copy Socket Receive| B[PipeWriter: Pooled Memory Blocks]
    B -->|Commit Written Bytes| C[Pipe: Linked ReadOnlySequence Memory]
    C -->|ReadAsync Zero Alloc| D[PipeReader]
    D -->|Zero-Copy Slice| E["SequenceReader<byte>\n(Frame Boundary Parser)"]
    E -->|AdvanceTo Consumed/Examined| D
    E -->|Emit Unmanaged Value Struct| F["System.Threading.Channels.Channel<T>\n(Bounded Backpressure Queue)"]
    F -->|WaitToReadAsync / ValueTask| G[Worker Pool: Async Consumer Processors]
    G -->|Batch Write| H[Downstream Database / Cache / Sink]

using System;
using System.Runtime.InteropServices;

namespace Centralogic.Streaming.Core;

/// <summary>
/// Fully blittable, unmanaged telemetry event struct.
/// Guaranteed 0 heap allocations and zero GC pressure.
/// </summary>
[StructLayout(LayoutKind.Sequential, Pack = 1)]
public readonly record struct TelemetryEvent(
    long EventId,
    int DeviceId,
    short EventType,
    long TimestampUtc,
    double MetricValue
)
{
    public const int BinaryFrameSize = sizeof(long) + sizeof(int) + sizeof(short) + sizeof(long) + sizeof(double); // 30 bytes
    public const byte FrameMagicByte = 0xAA;
}

using System;
using System.Buffers;
using System.Buffers.Binary;
using System.IO.Pipelines;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;

namespace Centralogic.Streaming.Pipelines;

public sealed class ZeroAllocEventProcessor
{
    private readonly PipeReader _reader;
    private readonly ChannelWriter<TelemetryEvent> _channelWriter;
    private readonly ILogger<ZeroAllocEventProcessor> _logger;

    public ZeroAllocEventProcessor(
        PipeReader reader,
        ChannelWriter<TelemetryEvent> channelWriter,
        ILogger<ZeroAllocEventProcessor> logger)
    {
        _reader = reader;
        _channelWriter = channelWriter;
        _logger = logger;
    }

    public async ValueTask ProcessPipeAsync(CancellationToken cancellationToken = default)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            ReadResult result = await _reader.ReadAsync(cancellationToken);
            ReadOnlySequence<byte> buffer = result.Buffer;

            try
            {
                // Parse all complete frames available in the current sequence
                ParseFrames(ref buffer);
            }
            finally
            {
                // Tell the PipeReader how much of the buffer was consumed and examined
                _reader.AdvanceTo(buffer.Start, buffer.End);
            }

            if (result.IsCompleted || result.IsCanceled)
            {
                break;
            }
        }

        _channelWriter.Complete();
    }

    private void ParseFrames(ref ReadOnlySequence<byte> buffer)
    {
        var reader = new SequenceReader<byte>(buffer);

        while (!reader.End)
        {
            // Look for magic byte marker indicating frame start
            if (!reader.TryAdvanceTo(TelemetryEvent.FrameMagicByte, advancePastDelimiter: true))
            {
                // Magic byte not found; entire buffer examined without match
                break;
            }

            // Check if remaining bytes in sequence satisfy the payload requirement
            if (reader.Remaining < TelemetryEvent.BinaryFrameSize)
            {
                // Incomplete frame; rewind 1 byte to include magic byte in next read
                reader.Rewind(1);
                break;
            }

            // Parse fields with zero allocations using big-endian binary primitives
            if (reader.TryReadBigEndian(out long eventId) &&
                reader.TryReadBigEndian(out int deviceId) &&
                reader.TryReadBigEndian(out short eventType) &&
                reader.TryReadBigEndian(out long timestampUtc) &&
                reader.TryReadBigEndian(out long rawMetricBits))
            {
                double metricValue = BitConverter.Int64BitsToDouble(rawMetricBits);
                var eventItem = new TelemetryEvent(eventId, deviceId, eventType, timestampUtc, metricValue);

                // Write to bounded backpressure channel without waiting if capacity exists
                if (!_channelWriter.TryWrite(eventItem))
                {
                    // Channel is full: execute non-blocking async write to enforce backpressure
                    FastAsyncWrite(eventItem);
                }
            }
        }

        // Slice buffer to discard parsed portions
        buffer = buffer.Slice(reader.Position);
    }

    private void FastAsyncWrite(TelemetryEvent eventItem)
    {
        ValueTask writeTask = _channelWriter.WriteAsync(eventItem);
        if (!writeTask.IsCompletedSuccessfully)
        {
            writeTask.AsTask().GetAwaiter().GetResult();
        }
    }
}

using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace Centralogic.Streaming.Workers;

public sealed class StreamConsumerService : BackgroundService
{
    private readonly ChannelReader<TelemetryEvent> _channelReader;
    private readonly ILogger<StreamConsumerService> _logger;

    public StreamConsumerService(
        ChannelReader<TelemetryEvent> channelReader,
        ILogger<StreamConsumerService> logger)
    {
        _channelReader = channelReader;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Stream Consumer Worker initialized with zero-allocation backpressure.");

        // Stream through channel reader without allocations
        while (await _channelReader.WaitToReadAsync(stoppingToken))
        {
            while (_channelReader.TryRead(out TelemetryEvent item))
            {
                // Process individual unmanaged event with 0 GC overhead
                ProcessTelemetryEvent(item);
            }
        }
    }

    private static void ProcessTelemetryEvent(in TelemetryEvent item)
    {
        // High-speed business logic: e.g., in-memory delta tracking or vectorized ring buffer
        if (item.MetricValue > 950.0)
        {
            // Trigger sub-millisecond alert threshold
        }
    }
}