Sockets on a Pipeline: Low-Allocation Networking with Beskar.Memory

Back to the metal: returning to sockets

For the past several months, my focus has been locked onto text parsing. Building Beskar.Markdown was an exercise in extreme micro-optimization: using ReadOnlySpan<char> to slice headers without heap allocations, designing index-based value node structures to replace recursive object trees, and creating direct, stream-like HTML renderers to avoid string concatenation.

It was deeply satisfying work, but after spending so much time parsing text, I felt the familiar pull of my first love.

Sockets.

Ever since I was a little kid writing basic chat apps and raw socket loopbacks in my bedroom, networking has had a special hold on me. The sheer joy of watching raw bytes move across a network interface, managing connection state, and negotiating handshake semantics remains unmatched.

So, I am coming back to networking.

But this time, I want to do it right. I want to build a transport layer (Beskar.Networking) that can comfortably process over 680,000 messages per second on a single loopback connection. To get there, we have to leave the classic NetworkStream behind and embrace two powerhouse concepts in modern .NET 10: System.IO.Pipelines and the custom memory pooling engine of Beskar.Memory.

Why standard socket programming is allocation-heavy

A naive TCP server usually relies on a thread-per-connection pattern or tasks that read from a stream:

public async Task HandleConnectionAsync(TcpClient client)
{
    var stream = client.GetStream();
    var buffer = new byte[4096];

    while (true)
    {
        var read = await stream.ReadAsync(buffer, 0, buffer.Length);
        if (read == 0) break;

        // Process incoming bytes...
    }
}

This looks clean, but under load, it introduces severe bottlenecks:

1. Buffer allocations: If you allocate a new buffer per connection or per read, you inundate the Garbage Collector (GC) with short-lived Gen0 objects.
2. Buffer pinning: Passing heap-allocated byte arrays to OS socket calls requires the runtime to "pin" them, preventing the GC from relocating them during compacting cycles and leading to heap fragmentation.
3. Queueing latency: Managing thread scheduling for reading and writing on the same stream leads to synchronization overhead and context switching.

To solve this, Beskar.Networking decouples raw input/output (I/O) from application-level processing using a duplex pipeline architecture.

The pipeline architecture

Instead of working with raw sockets directly, the core of Beskar.Networking models a connection as an IDuplexPipe consisting of a separate PipeReader (the input stream) and PipeWriter (the output stream).

            +----------------+   Raw Bytes   +----------------+
  +-------->|   OS Socket    |-------------->| SocketReceiver |
  |         +----------------+               +----------------+
  |                                                  |
  | SendAsync                                        | Fill / Flush
  |                                                  v
+--------------+                             +----------------+
| SocketSender |                             |   Input Pipe   |
+--------------+                             +----------------+
  ^                                                  |
  | Read / Advance                                   | ReadAsync
  |                                                  v
+----------------+   WriteAsync              +----------------+
|  Output Pipe   |<--------------------------|  App Logic     |
+----------------+                           +----------------+

By decoupling reading and writing into isolated asynchronous tasks (ProcessReceiveAsync and ProcessSendAsync), we ensure that slow application processing never blocks the OS socket receive buffers.

Here is the central wrapper class, SocketConnection, which manages these two pipelines:

using System.Buffers;
using System.IO.Pipelines;
using System.Net.Sockets;
using Beskar.Networking.Abstractions.Interfaces.Pools;

namespace Beskar.Networking.Transports.Common.Sockets;

public sealed class SocketConnection
   : IDuplexPipe, IAsyncDisposable, IPooledObject
{
   private readonly SocketSender _sender;
   private readonly SocketReceiver _receiver;
   private Socket? _socket;

   private readonly Lock _shutdownLock = new();
   private bool _isDisposed;
   private bool _isAborted;

   public PipeReader Input => _receiver.Pipe.Reader;
   public PipeWriter Output => _sender.Pipe.Writer;

   public SocketConnection(PipeScheduler scheduler, MemoryPool<byte> bufferPool)
   {
      var pipeOptions = new PipeOptions(
         pool: bufferPool,
         readerScheduler: scheduler,
         writerScheduler: scheduler,
         useSynchronizationContext: false);

      _sender = new SocketSender(pipeOptions);
      _receiver = new SocketReceiver(pipeOptions);
   }

   public void Initialize(Socket socket)
   {
      _socket = socket;

      _sender.Initialize(this, socket);
      _receiver.Initialize(this, socket);
   }

   public void Start()
   {
      _sender.Start();
      _receiver.Start();
   }

   public void Abort(Exception? exception = null)
   {
      lock (_shutdownLock)
      {
         if (_isAborted || _isDisposed) return;
         _isAborted = true;
      }

      _sender.Stop();
      _receiver.Stop();

      if (_socket != null)
      {
         try
         {
            _socket.Close(timeout: 0);
         }
         catch { /* Expected */ }
      }
   }

   public async ValueTask StopAsync()
   {
      await _sender.StopAsync();
      await _receiver.StopAsync();

      if (!_isAborted && _socket != null)
      {
         try
         {
            _socket.Shutdown(SocketShutdown.Both);
         }
         catch { /* Expected */ }
         finally
         {
            _socket.Dispose();
         }
      }
   }

   public async ValueTask DisposeAsync()
   {
      lock (_shutdownLock)
      {
         if (_isDisposed) return;
         _isDisposed = true;
      }

      await StopAsync();
   }

   public bool TryResetState()
   {
      if (!_sender.TryResetState() || !_receiver.TryResetState())
      {
         return false;
      }

      _socket = null;
      _isDisposed = false;
      _isAborted = false;

      return true;
   }
}

Zero-allocation receiving with PinnedBlockMemoryPool

To read bytes from the socket without allocating memory, SocketReceiver rents a block from a custom memory pool provided by Beskar.Memory. This is the PinnedBlockMemoryPool.

Unlike System.Buffers.ArrayPool<byte>, which rents standard arrays that the GC must pin during native I/O transitions, a pinned memory pool allocates native, unmanaged memory blocks up-front. The PipeWriter rents these pinned blocks, allowing us to pass them directly to the socket's async receive method.

Here is the receive loop in SocketReceiver.cs:

private async Task ProcessReceiveAsync()
{
   var socket = _socket;
   if (socket == null) return;

   try
   {
      while (true)
      {
         // Rent memory from the pipe's pool (PinnedBlockMemoryPool)
         var memory = Pipe.Writer.GetMemory(MinAllocBufferSize);

         // Read directly into the pinned pool block
         var bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None, _cts.Token);
         if (bytesRead == 0)
         {
            break;
         }

         TraceLogger.LogNeutralInfo("SocketReceiver: Received {0} bytes from socket", bytesRead);

         // Tell the PipeWriter how much data was written
         Pipe.Writer.Advance(bytesRead);

         // Make the bytes visible to the application reader
         var result = await Pipe.Writer.FlushAsync(_cts.Token);
         if (result.IsCompleted || result.IsCanceled)
         {
            break;
         }
      }
   }
   catch (OperationCanceledException) { /* Expected */ }
   catch (Exception ex)
   {
      _connection?.Abort(ex);
   }
   finally
   {
      await Pipe.Writer.CompleteAsync();
   }
}

Because Pipe.Writer.GetMemory() returns a Memory<byte> pointing directly into a pre-allocated page of the PinnedBlockMemoryPool, the hot path of receiving data invokes zero allocations and causes zero heap fragmentation.

Asynchronous transmission with multi-segment buffer flattening

Sending data is the reverse process. The application writes bytes to the PipeWriter (SocketConnection.Output), which buffers them until they are flushed. On the other side, SocketSender handles the raw network transmission.

However, data written to a pipeline is stored in a ReadOnlySequence<byte>, which can span across multiple disconnected memory blocks (segments). Eagerly converting a multi-segment sequence into a single flat array via .ToArray() would allocate heap memory, destroying our low-allocation guarantees.

Instead, SocketSender inspects the buffer structure:

• If the buffer is a single segment, it sends it directly.
• If it spans multiple segments, it iterates through them, sending each segment sequentially without merging.

Here is the implementation in SocketSender.cs:

private async Task ProcessSendAsync()
{
   var socket = _socket;
   if (socket == null) return;

   try
   {
      while (true)
      {
         var result = await Pipe.Reader.ReadAsync(_cts.Token);
         var buffer = result.Buffer;

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

         if (!buffer.IsEmpty)
         {
            await SendBufferAsync(socket, buffer, _cts.Token);
         }

         // Mark the bytes as consumed so the pipeline can reuse the memory blocks
         Pipe.Reader.AdvanceTo(buffer.End);

         if (result.IsCompleted)
         {
            break;
         }
      }
   }
   catch (OperationCanceledException) { }
   catch (Exception ex)
   {
      _connection?.Abort(ex);
   }
   finally
   {
      await Pipe.Reader.CompleteAsync();
   }
}

private async ValueTask SendBufferAsync(Socket socket, ReadOnlySequence<byte> buffer, CancellationToken cancellationToken)
{
   if (buffer.IsSingleSegment)
   {
      await SendMemoryAsync(socket, buffer.First, cancellationToken);
   }
   else
   {
      // Send each segment individually to avoid flattening allocations
      foreach (var memory in buffer)
      {
         await SendMemoryAsync(socket, memory, cancellationToken);
      }
   }
}

private async ValueTask SendMemoryAsync(Socket socket, ReadOnlyMemory<byte> memory, CancellationToken cancellationToken)
{
   while (!memory.IsEmpty)
   {
      var bytesSent = await socket.SendAsync(memory, SocketFlags.None, cancellationToken);
      
      if (bytesSent == 0)
      {
         throw new SocketException((int)SocketError.ConnectionAborted);
      }

      TraceLogger.LogNeutralInfo("SocketSender: Transmitted {0} bytes to socket", bytesSent);
      
      memory = memory[bytesSent..];
   }
}

Eliminating connection wrapper allocations with Object Pooling

Although our byte buffers are pooled, instantiating a new SocketConnection, SocketReceiver, and SocketSender for every incoming connection would still trigger considerable heap allocation.

To bypass this, Beskar.Networking introduces a thread-safe connection pooling mechanism. Inside the TcpIoQueueRegistry, we maintain an AsyncDisposableObjectPool<SocketConnection>.

When a socket is accepted, we pull a pre-allocated SocketConnection wrapper from the pool, initialize it with the new socket handle, and push it to the accept loop. Once the session is closed, the connection is reset (TryResetState) and returned to the pool:

// Accept loop in TcpNetworkListener.cs
private async Task AcceptLoopAsync(Socket listenerSocket, CancellationToken token)
{
   while (!token.IsCancellationRequested)
   {
      var clientSocket = await listenerSocket.AcceptAsync(token);
      
      // Extract connection from TcpIoQueueRegistry's object pool and run handshake
      _ = Task.Run(() => HandshakeAndEnqueueAsync(clientSocket, token), token);
   }
}

By recycling the SocketConnection graph itself, the server can sustain a high rate of connection churn with near-zero GC impact.

Putting it all together: a simple client-server exchange

Here is how you use INetworkListener and INetworkClient to bind, connect, and stream data over our fast socket pipelines: (This is a very simple example that assumes no errors)

using System.Buffers;
using System.Net;
using Beskar.Networking.Transports.Tcp;

var endPoint = new IPEndPoint(IPAddress.Loopback, 8080);
var options = new TcpTransportOptions();

// 1. Bind the Listener
var listener = new TcpNetworkListener(endPoint, options);
await listener.BindAsync();

// 2. Connect the Client
var client = new TcpNetworkClient(options);
var connectResult = await client.ConnectAsync(endPoint);
var clientSession = connectResult.Success;

// 3. Accept the Server Session
var acceptResult = await listener.AcceptSessionAsync();
var serverSession = acceptResult.Success;

// 4. Open and Accept Bidirectional Streams
var clientStream = (await clientSession.AcceptStreamAsync()).Success;
var serverStream = (await serverSession.AcceptStreamAsync()).Success;

// 5. Client Writes, Server Reads
var message = "Hello from Beskar Pipelines!"u8;
await clientStream.Transport.Output.WriteAsync(message);
await clientStream.Transport.Output.FlushAsync();

var readResult = await serverStream.Transport.Input.ReadAsync();
var readBytes = readResult.Buffer.ToArray();
serverStream.Transport.Input.AdvanceTo(readResult.Buffer.End);

Console.WriteLine($"Server received: {System.Text.Encoding.UTF8.GetString(readBytes)}");

// Cleanup
await clientSession.DisposeAsync();
await serverSession.DisposeAsync();
await listener.UnbindAsync();

Unified Abstractions: WebSockets and QUIC under the Hood

While the TCP transport is the backbone of traditional raw socket workloads, Beskar.Networking was designed from day one to be transport-agnostic.

We don't want the application logic to care whether bytes are arriving over a raw TCP connection, wrapped inside WebSocket frames, or multiplexed over QUIC streams. To achieve this, the entire library centers around four core interfaces:

1. INetworkListener: Binds to local endpoints and accepts incoming sessions.
2. INetworkClient: Establishes outbound sessions.
3. INetworkSession: Represents the connection session. It indicates whether the transport supports multiplexing (like QUIC) or is a single-channel protocol (like TCP and WebSockets).
4. INetworkStream: Wraps the underlying pipeline (IDuplexPipe with Input and Output) to perform raw binary I/O.

Because these interfaces act as the universal compiler boundary, we can seamlessly support WebSockets and QUIC transports using the exact same application code.

WebSockets (WS) with Pipeline Buffering

WebSockets are inherently frame-based rather than stream-based. In Beskar.Networking.Transports.Ws, we write a custom frame parser that decodes WebSocket headers and channels the raw payload directly into our PipeWriter. The application still reads from INetworkStream.Transport.Input as a continuous stream of bytes. When writing, the pipeline automatically packets the outgoing data into WebSocket binary frames before pushing them to the socket.

QUIC (TLS) with Native Multiplexing

Unlike TCP and WebSockets, QUIC is a multiplexed transport supporting multiple independent bidirectional and unidirectional streams over a single connection. By implementing INetworkSession.AcceptStreamAsync and INetworkSession.OpenStreamAsync, we map QUIC's streams directly to individual INetworkStream pipelines. The outer INetworkSession wraps the underlying QuicConnection, while each individual stream is driven by its own isolated pipelines.

Because of this unified abstraction, swap-in protocols are trivial:

// To swap from TCP to WebSockets or QUIC, simply swap the Listener/Client implementation:
INetworkListener listener = options.UseQuic 
   ? new QuicNetworkListener(endPoint, quicOptions) 
   : new TcpNetworkListener(endPoint, tcpOptions);

await listener.BindAsync();
var sessionResult = await listener.AcceptSessionAsync();
var session = sessionResult.Success;

// The rest of the pipeline application logic is identical:
var stream = (await session.AcceptStreamAsync()).Success;
var readResult = await stream.Transport.Input.ReadAsync();

Conclusion & what is next

Returning to sockets with modern C# has been an absolute blast. It is incredible to see how far .NET has come since the days of spawning raw background threads and polling socket streams.

By marrying System.IO.Pipelines with the zero-allocation pinned pooling in Beskar.Memory, we have laid down a foundation for Beskar.Networking that is extremely fast, highly predictable, and gentle on the GC.

With TCP, WebSockets, and QUIC transports already working behind a unified interface, my focus now turns to implementing the MQTT broker and client protocol directly on top of this high-performance foundation.

Keep your allocations low and your pipelines flowing!

If you want to inspect the code or try them in your projects:

• GitHub: github.com/MarvinDrude/Beskar.Networking
• GitHub: github.com/MarvinDrude/Beskar.Memory