C# Networking Deep Dive With io_uring part 3 - Touching the bytes

In part 2 we introduced an asynchronous API when reading data from the wire, where only the number of received bytes was considered. On this part 3 let's extend it to receive the actual data and parse it.

As usual, the entire source code can be found at Minima

Before diving into the code let's understand which data structures make sense to consider.

Data will be pushed to the CQ shared ring buffers by the kernel whenever some data arrives from the wire, this can represent a partial, complete or more than one request. By request I mean it could be a HTTP/1.1, HTTP/2, gRPC, websocket, etc pretty much a request from any protocol. We are going to consider HTTP/1.1 for this project and articles. So, as I was saying whenever we call a ReadAsync, we receive a snapshot with a tail that should point us to a snapshot of CQE metadata, currently this metadata only includes the number of bytes from each CQE, we need to add a byte* pointing to where the data is.

public struct Item
{
    public byte* Ptr; // new
    public ushort Bid;
    public int Len;
    public bool HasBuffer;

    public ReadOnlySpan<byte> AsSpan() => new(Ptr, Len); // new
}

then on the reactor's dispatch receive branch

else if (kind == KindRecv)
{
    bool   hasBuf = (cqe.flags & IORING_CQE_F_BUFFER) != 0;
    ushort bid    = hasBuf ? (ushort)(cqe.flags >> IORING_CQE_BUFFER_SHIFT) : (ushort)0;

    if (!Connections.TryGetValue(fd, out var conn))
    {
        if (hasBuf) ReturnBuffer(bid);

        return;
    }

    byte* ptr = hasBuf ? _bufSlab + (nuint)bid * (nuint)BufferSize : null;
    conn.Complete(cqe.res, bid, hasBuf, ptr);

    if (!more && cqe.res > 0)
    {
         SubmitRecvMultishot(fd);
    }
}

Now for each received CQE we store the byte* where kernel stored the received data, each ReadAsync will return a snapshot that contains one or more Item, each Item contains the metadata for one CQE.

Now on the handler side we must consume this data. We don't want to be dealing with pointers though, that would force us to use unsafe everywhere we touch the received data.

We already have the ReadOnlySpan view of the data via the AsSpan() but spans are ref structs and can't be freely used anywhere, but why is that? Spans are used to create views over stack allocated data unlike its heap allocated counterpart Memory/ReadOnlyMemory, we can't directly use ReadOnlyMemory either because it can't be directly created from a byte* even though this byte* points at heap allocated data stored in each reactor's _bufSlab we initialize for each reactor.

I present you UnmanagedMemoryManager

public sealed unsafe class UnmanagedMemoryManager : MemoryManager<byte>
{
    private readonly byte* _ptr;
    private readonly int _length;

    public ushort BufferId { get; }

    public byte* Ptr => _ptr;

    public int Length => _length;

    public UnmanagedMemoryManager(byte* ptr, int length)
    {
        _ptr = ptr;
        _length = length;
    }

    public UnmanagedMemoryManager(byte* ptr, int length, ushort bufferId)
    {
        _ptr = ptr;
        _length = length;
        BufferId = bufferId;
    }

    public override Span<byte> GetSpan() => new Span<byte>(_ptr, _length);

    public override MemoryHandle Pin(int elementIndex = 0) => new MemoryHandle(_ptr + elementIndex);

    public override void Unpin() { }

    public void Free()
    {
        if (_ptr != null)
        { 
            NativeMemory.AlignedFree(_ptr); 
        }
    }

    protected override void Dispose(bool disposing) { }
}

UnmanagedMemoryManager is the bridge between safe and unsafe code, by inheriting from MemoryManager it can be exposed as Memory and plugged into the entire BCL ecosystem for free:

• PipeReader / PipeWriter
• Stream.ReadAsync(Memory) / WriteAsync(ReadOnlyMemory)
• ReadOnlySequence (built from ReadOnlyMemory segments)
• IBufferWriter
• Any async API that takes Memory

This is especially useful for ReadOnlySequence which is very handy when dealing with TCP fragmentation.

So, how does how handler look now?

{
    if (item.HasBuffer)
    {
        UnmanagedMemoryManager mem = item.AsMemoryManager();
        ReadOnlyMemory<byte> data = mem.Memory;// data is now usable with any BCL Memory<byte>/async API
        _ = data.Length;

         reactor.ReturnBuffer(mem.BufferId);
    }
    conn.QueueResponse(fd);
}

The possibilites are now endless, we can build a ReadOnlySequence from all the data to facilitate slicing across multiple segments, also in the case of incomplete requests we can again create a ReadOnlySequence, call another ReadAsync and add the received segments to the already existing ReadOnlySequence.