Pipes and Performance

Efficient and reliable data transmission is the backbone of seamless communication. At the heart of this process lies TCP (Transmission Control Protocol) networking — a fundamental technology that ensures data is sent and received accurately across the internet.

I never had to create a TCP server/client with .NET before, but I always wanted to learn. I was never compelled to do it seeing as the process was too complex and mistakes were easy to make.

A great reference for writing a TCP server is the Microsoft Pipelines documentation. Pipelines were created for two reasons:

Reduce the boilerplate code required to handle stream processing
Create a high-performance framework (using Memory/Span and rented buffers) for handling stream processing — read more on the .NET blog

What has this produced? The TechEmpower benchmarks put Kestrel (which is built on Pipelines) among the top performers:

TechEmpower plaintext benchmark results showing Kestrel performance

Usage

Using pipes is pretty straightforward — most of the code is inspired by David Fowler's example. I abstracted away some of the boilerplate and created a generic TMessage that is shared between server and client.

The Server — when started it will continuously listen for new connections. Processing the client only involves reading the pipe and iterating the buffer:

csharp

public class TestTcpSever<TMessage>
{
    private readonly IPAddress _ipAddress;
    private readonly int _port;
    private readonly IMessageSerialization<TMessage> _messageSerialization;

    internal TestTcpSever(IPAddress ipAddress,
        int port,
        IMessageSerialization<TMessage> messageSerialization)
    {
        _ipAddress = ipAddress;
        _port = port;
        _messageSerialization = messageSerialization;
    }

    public async Task StartAsync(Action<TMessage> onMessage)
    {
        var listener = new TcpListener(_ipAddress, _port);
        listener.Start();

        Console.WriteLine(
quot;Server listening on port {_port}");

        while (true)
        {
            TcpClient client = await listener.AcceptTcpClientAsync();
            var remote = client.Client.RemoteEndPoint.Serialize().ToString();
            var handle = client.Client.Handle.ToInt64();
            Console.WriteLine(
quot;new connection from client {remote} with handle {handle}");
            _ = ProcessClientAsync(client, onMessage);
        }
    }

    private async Task ProcessClientAsync(TcpClient client, Action<TMessage> onMessage)
    {
        NetworkStream stream = client.GetStream();
        PipeReader pipeReader = PipeReader.Create(stream);

        while (true)
        {
            ReadResult result = await pipeReader.ReadAsync();
            ReadOnlySequence<byte> buffer = result.Buffer;

            byte[] messageBytes = ArrayPool<byte>.Shared.Rent((int)buffer.Length);

            int lastIndex = 0;
            foreach (ReadOnlyMemory<byte> segment in buffer)
            {
                Array.Copy(segment.ToArray(), 0, messageBytes, lastIndex, segment.Length);
                lastIndex += segment.Length;
            }

            TMessage messageReceived = _messageSerialization.Deserialize(messageBytes);
            onMessage(messageReceived);

            pipeReader.AdvanceTo(buffer.End);

            if (result.IsCompleted)
                break;

            string received = 
quot;received at {DateTime.UtcNow.ToLongDateString()}";
            await stream.WriteAsync(
                Encoding.UTF8.GetBytes(received),
                0,
                Encoding.UTF8.GetBytes(received).Length);
        }

        pipeReader.Complete();
    }
}

The Client — mirrors the server shape, using PipeWriter to send and PipeReader to receive:

csharp

public class TestTcpClient<TMessage> : IDisposable, IAsyncDisposable
{
    private readonly NetworkStream _stream;
    private readonly TcpClient _client;
    private readonly IMessageSerialization<TMessage> _messageSerialization;

    internal TestTcpClient(TcpClient client,
        IMessageSerialization<TMessage> messageSerialization)
    {
        _client = client;
        _messageSerialization = messageSerialization;
        _stream = client.GetStream();
    }

    public void Dispose()
    {
        _stream.Dispose();
        _client.Dispose();
    }

    public async ValueTask DisposeAsync()
    {
        _client.Dispose();
        await _stream.DisposeAsync();
    }

    public async Task SendMessageAsync(TMessage message)
    {
        PipeWriter pipeWriter = PipeWriter.Create(_stream);
        byte[] messageBytes = _messageSerialization.Serialize(message);
        await pipeWriter.WriteAsync(messageBytes);
        await pipeWriter.FlushAsync();
    }

    public async Task ProcessIncoming(CancellationToken cancellationToken)
    {
        PipeReader pipeReader = PipeReader.Create(_stream);

        while (!cancellationToken.IsCancellationRequested)
        {
            ReadResult result = await pipeReader.ReadAsync();
            ReadOnlySequence<byte> buffer = result.Buffer;

            foreach (var segment in buffer)
            {
                byte[] messageBytes = segment.ToArray();
                string messageString = Encoding.UTF8.GetString(messageBytes);
                Console.WriteLine(
quot;Received message from server: {messageString}");
            }

            pipeReader.AdvanceTo(buffer.End);

            if (result.IsCompleted)
                break;
        }
    }
}

The Connection factory — ties server and client together under a shared TMessage type and address:

csharp

public class TestTcpConnection<TMessage>(
    IPAddress ipAddress,
    int port,
    IMessageSerialization<TMessage> serialization)
{
    public TestTcpSever<TMessage> CreateServer()
    {
        return new TestTcpSever<TMessage>(ipAddress, port, serialization);
    }

    public async Task<TestTcpClient<TMessage>> ConnectAsync()
    {
        TcpClient client = new TcpClient();
        CancellationTokenSource cts = new();

        await client.ConnectAsync(ipAddress, port, cts.Token);

        return new TestTcpClient<TMessage>(client, serialization);
    }
}

Using Our Server / Client

With the boilerplate in place, spinning up a server and connecting a client is minimal:

csharp

TestTcpConnection<TestMessage> connection = new TestTcpConnection<TestMessage>(
    IPAddress.Any,
    8080,
    new TestMessageSerialization());

TestTcpSever<TestMessage> server = connection.CreateServer();

await server.StartAsync((message) =>
{
    Console.WriteLine(
quot;Server: Received message {message}");
});

csharp

TestTcpConnection<TestMessage> connection = new TestTcpConnection<TestMessage>(
    IPAddress.Loopback,
    8080,
    new TestMessageSerialization());

TestTcpClient<TestMessage> client = await connection.ConnectAsync();

CancellationTokenSource cts = new();

client.ProcessIncoming(cts.Token);

while (!cts.IsCancellationRequested)
{
    Console.WriteLine("Send:");
    var input = Console.ReadLine();

    await client.SendMessageAsync(new TestMessage()
    {
        Content = input ?? "empty",
        Timestamp = DateTime.UtcNow
    });
}

SpinWait.SpinUntil(() => cts.IsCancellationRequested);

.NET TCP Networking With Pipes

Pipes and Performance

Usage

Using Our Server / Client