/* naut_echo — Phase 1 gate. * * A single-reactor io_uring echo server that proves the foundation works end to * end: multishot accept, recv/send driven entirely off the page-aligned buffer * pool with ZERO per-operation allocation in steady state. Throughput on * loopback should be limited by memory bandwidth / the single core, not by the * allocator or syscalls. * * It is intentionally one-in-flight-op-per-connection (recv -> send -> recv). * The real peer reactor (later phase) uses multishot recv + provided buffers * and pipelines; this is the minimal honest exercise of the primitives. * * usage: naut_echo [port] (default 9000) */ #include "naut/uring.h" #include "naut/net.h" #include "naut/buf.h" #include "naut/log.h" #include "naut/system.h" #include #include #include #include #include #include #include #define ECHO_BLOCK (128u * 1024u) #define ECHO_BUFS 4096u #define RING_ENTRIES 4096u /* user_data tagging: low 3 bits = op, high bits = conn* (16-byte aligned). */ enum { TAG_ACCEPT = 1, TAG_RECV = 2, TAG_SEND = 3 }; #define UD(p, tag) ((__u64)(uintptr_t)(p) | (unsigned)(tag)) #define UD_TAG(ud) ((unsigned)((ud) & 0x7u)) #define UD_PTR(ud) ((conn *)(uintptr_t)((ud) & ~(__u64)0x7u)) typedef struct conn { int fd; uint32_t sent; /* bytes of buf->len already written (partial sends) */ naut_buf *buf; bool awaiting_notif; bool recv_fixed; /* the in-flight recv used the fixed buffer */ } conn; static volatile sig_atomic_t g_stop = 0; static void on_signal(int s) { (void)s; g_stop = 1; } static naut_bufpool *g_pool; static _Atomic uint64_t g_bytes = 0, g_conns = 0, g_zc_copied = 0; static void arm_recv(naut_ring *owner, conn *c) { struct io_uring_sqe *sqe = io_uring_get_sqe(&owner->ring); c->recv_fixed = naut_ring_prep_recv(owner, sqe, c->fd, c->buf->data, c->buf->cap, 0); io_uring_sqe_set_data64(sqe, UD(c, TAG_RECV)); } static void arm_send(naut_ring *owner, conn *c) { struct io_uring_sqe *sqe = io_uring_get_sqe(&owner->ring); c->awaiting_notif = naut_ring_prep_send( owner, sqe, c->fd, c->buf->data + c->sent, c->buf->len - c->sent, MSG_NOSIGNAL, true); io_uring_sqe_set_data64(sqe, UD(c, TAG_SEND)); } static void conn_close(conn *c) { close(c->fd); naut_buf_put(c->buf); free(c); } int main(int argc, char **argv) { uint16_t port = (argc > 1) ? (uint16_t)atoi(argv[1]) : 9000; int cpu = getenv("NAUT_CPU") ? atoi(getenv("NAUT_CPU")) : -1; int numa_node = getenv("NAUT_NUMA_NODE") ? atoi(getenv("NAUT_NUMA_NODE")) : -1; bool sqpoll = getenv("NAUT_SQPOLL") != NULL; bool hugepages = getenv("NAUT_HUGEPAGES") != NULL; signal(SIGINT, on_signal); signal(SIGTERM, on_signal); signal(SIGPIPE, SIG_IGN); if (cpu >= 0 && naut_pin_current_thread(cpu) != NAUT_OK) NAUT_WARN("failed to pin reactor to CPU %d", cpu); naut_ring r; if (naut_ring_init_cpu(&r, RING_ENTRIES, sqpoll, cpu) != NAUT_OK) return 1; if (naut_ring_probe(&r) != NAUT_OK) { naut_ring_close(&r); return 1; } struct io_uring *ring = &r.ring; int lfd = naut_net_listen(port, 1024, true); if (lfd < 0) { naut_ring_close(&r); return 1; } g_pool = naut_bufpool_create_on_node( ECHO_BLOCK, ECHO_BUFS, hugepages, numa_node); if (!g_pool) { close(lfd); naut_ring_close(&r); return 1; } (void)naut_ring_register_bufpool(&r, g_pool); /* prime the multishot accept */ struct io_uring_sqe *sqe = io_uring_get_sqe(ring); io_uring_prep_multishot_accept(sqe, lfd, NULL, NULL, 0); io_uring_sqe_set_data64(sqe, UD(NULL, TAG_ACCEPT)); NAUT_INFO("echo listening on :%u", port); while (!g_stop) { int rc = io_uring_submit_and_wait(ring, 1); if (rc < 0 && rc != -EINTR) { NAUT_ERROR("submit_and_wait: %s", strerror(-rc)); break; } unsigned head, count = 0; struct io_uring_cqe *cqe; io_uring_for_each_cqe(ring, head, cqe) { count++; __u64 ud = cqe->user_data; int res = cqe->res; switch (UD_TAG(ud)) { case TAG_ACCEPT: { if (res < 0) { if (res != -ECANCELED) NAUT_WARN("accept: %s", strerror(-res)); } else { int cfd = res; naut_net_tune_peer(cfd); naut_buf *b = naut_buf_get(g_pool); if (!b) { NAUT_WARN("pool exhausted, dropping conn"); close(cfd); } else { conn *c = calloc(1, sizeof(*c)); c->fd = cfd; c->buf = b; atomic_fetch_add(&g_conns, 1); arm_recv(&r, c); } } /* re-arm if the kernel dropped the multishot registration */ if (!(cqe->flags & IORING_CQE_F_MORE)) { struct io_uring_sqe *s = io_uring_get_sqe(ring); io_uring_prep_multishot_accept(s, lfd, NULL, NULL, 0); io_uring_sqe_set_data64(s, UD(NULL, TAG_ACCEPT)); } break; } case TAG_RECV: { conn *c = UD_PTR(ud); if (res <= 0) { /* A fixed-buffer recv rejected with -EINVAL means this * kernel doesn't support IORING_RECVSEND_FIXED_BUF on plain * recv. Disable it ring-wide and retry THIS connection * unfixed. We key off the per-conn flag, not the ring flag, * so every connection that armed a fixed recv before the * flag flipped recovers too (otherwise all but the first * would be torn down). */ if (res == -EINVAL && c->recv_fixed) { if (r.recv_fixed) { NAUT_WARN("fixed-buffer recv unsupported at runtime; " "falling back to normal recv"); r.recv_fixed = false; } arm_recv(&r, c); break; } if (res < 0) NAUT_WARN("recv completion: %s", strerror(-res)); conn_close(c); break; } c->buf->len = (uint32_t)res; c->sent = 0; arm_send(&r, c); break; } case TAG_SEND: { conn *c = UD_PTR(ud); if (cqe->flags & IORING_CQE_F_NOTIF) { if (res & IORING_NOTIF_USAGE_ZC_COPIED) { uint64_t copied = atomic_fetch_add(&g_zc_copied, 1) + 1; if (copied == 8) { NAUT_WARN("SEND_ZC is copying on this transport; " "disabling it for this ring"); r.send_zc = false; } } c->awaiting_notif = false; if (c->sent < c->buf->len) arm_send(&r, c); else { c->buf->len = 0; arm_recv(&r, c); } break; } if (res <= 0) { conn_close(c); break; } c->sent += (uint32_t)res; atomic_fetch_add(&g_bytes, (uint64_t)res); if (!c->awaiting_notif) { if (c->sent < c->buf->len) arm_send(&r, c); else { c->buf->len = 0; arm_recv(&r, c); } } break; } default: NAUT_PANIC("bad user_data tag %u", UD_TAG(ud)); } } io_uring_cq_advance(ring, count); } NAUT_INFO("shutting down: %llu conns, %llu bytes echoed, %llu SEND_ZC copied notifications", (unsigned long long)atomic_load(&g_conns), (unsigned long long)atomic_load(&g_bytes), (unsigned long long)atomic_load(&g_zc_copied)); close(lfd); naut_ring_unregister_buffers(&r); naut_bufpool_destroy(g_pool); naut_ring_close(&r); return 0; }