Async but its not anything standard (not TCP etc)

kj4tmp · January 25, 2025, 7:40am

I racking my brain over here trying to implement an asynchronous implementation for raw sockets.

At a high level:

Server sends data to clients. Each data has unique 64 bit header identifying the transaction with the client.
Clients respond with the same 64 bit header.
Need an async interface on the server to run multiple transactions concurrently.
Have to use a single raw socket on the server.

This kind of feels like I am implementing my own TCP stack in terms of complexity. I currently have a synchronous API (server can perform one transaction at a time with clients), but I am trying to move to async API so I can do more than one transaction at once.

Do I need to hand-roll my own event loop?

I am thinking I could have some type of hash table to may the 64 bit transaction header to callbacks, but it feels like I may be re-inventing stuff that might already exist.
Can I accomplish this with libxev? Would it even be worth it?

I really don’t know where to even start here.

gonzo · January 25, 2025, 8:52am

When you say “raw sockets”, are you referring to transporting IP (not TCP) packets back and forth? That is really low-level, and I don’t think you will find much support for those already in place – maybe libpcap could be a starting point.

If you mean that you want to use datagram sockets, you will be using the UDP protocol (again, not TCP). In this case the “event loop philosophy” applies: you either use one of the well-known portable event loop implementations out there (libevent, libuv, libxev, maybe others) or you roll out your own by calling the OS primitives (select, poll, epoll, kqueue, completion ports, etc) – and deal with whatever portability requirements you may have.

Good luck!

kj4tmp · January 25, 2025, 8:58am

I’m referring to layer 2 networking: packet(7) - Linux manual page

I use npcap for windows, which provides a similar API to Linux.

gonzo · January 25, 2025, 9:15am

Yeah, I’m afraid you will have to rollout everything yourself.

If you do find a library that helps with this, could you please share it here?

From the mists of time, I do remember once using ACE, a C++ library to implement what amounted to ping. This was eons ago. The library is still supported, but I guess it would be considered old-style by now.

gonzo · January 25, 2025, 9:20am

And what do you know, that old thing is still there. For whatever it is worth.

dee0xeed · January 25, 2025, 9:58am

“Concurrency” and “asynchronicity/synchronicity” are orthogonal things.
You can do things concurrently without (true) asynchronous input/output,
i.e. using readiness notifications / reactor pattern (POSIX poll(), Linux specific epoll, FreeBSD specific kqueue). If you want (truly) asynchronous input/output, i.e completion notifications / proactor pattern, then IOCP on Windows, io_uring on Linux.

kj4tmp · January 26, 2025, 7:27am

So I guess I have decided to implement a sort of “Transaction Pool”. Which I guess you could call it an obect pool or event queue type thing.

I’m thinking of using a hash map. Key design considerations are:

I only want to allocate once on creation of the pool. Really trying to minimize allocations as much as possible.
I only need maybe worst case 50 “transactions” tracked in the pool at a time.

Do you have a recommendations on standard library data structures / example that I can learn from?

kj4tmp · January 26, 2025, 7:30am

It looks like libxev uses a linked list style event queue to be zero allocation?

github.com

mitchellh/libxev/blob/main/src/queue.zig

const std = @import("std");
const assert = std.debug.assert;

/// An intrusive queue implementation. The type T must have a field
/// "next" of type `?*T`.
///
/// For those unaware, an intrusive variant of a data structure is one in which
/// the data type in the list has the pointer to the next element, rather
/// than a higher level "node" or "container" type. The primary benefit
/// of this (and the reason we implement this) is that it defers all memory
/// management to the caller: the data structure implementation doesn't need
/// to allocate "nodes" to contain each element. Instead, the caller provides
/// the element and how its allocated is up to them.
pub fn Intrusive(comptime T: type) type {
    return struct {
        const Self = @This();

        /// Head is the front of the queue and tail is the back of the queue.
        head: ?*T = null,
        tail: ?*T = null,

This file has been truncated. show original

Looks like tiger beetle uses a linked list too

github.com

tigerbeetle/tigerbeetle/blob/main/src/fifo.zig

const std = @import("std");
const assert = std.debug.assert;

const constants = @import("./constants.zig");

/// An intrusive first in/first out linked list.
/// The element type T must have a field called "next" of type ?*T
pub fn FIFOType(comptime T: type) type {
    return struct {
        const FIFO = @This();

        in: ?*T = null,
        out: ?*T = null,
        count: u64 = 0,

        // This should only be null if you're sure we'll never want to monitor `count`.
        name: ?[]const u8,

        // If the number of elements is large, the constants.verify check in push() can be too
        // expensive. Allow the user to gate it. Could also be a comptime param?

This file has been truncated. show original

And looks like I have already accidentally implemented the proactor pattern…

github.com

kj4tmp/gatorcat/blob/main/src/lib/Port.zig

const std = @import("std");
const assert = std.debug.assert;

const nic = @import("nic.zig");
const telegram = @import("telegram.zig");
const wire = @import("wire.zig");

const Port = @This();

recv_frames_status_mutex: std.Thread.Mutex = .{},
recv_frames: [max_frames]*telegram.EtherCATFrame = undefined,
recv_frames_status: [max_frames]FrameStatus = [_]FrameStatus{FrameStatus.available} ** max_frames,
last_used_idx: u8 = 0,
link_layer: nic.LinkLayer,
settings: Settings,

pub const Settings = struct {
    source_mac_address: u48 = 0xffff_ffff_ffff,
    dest_mac_address: u48 = 0xABCD_EF12_3456,
};

This file has been truncated. show original