echo.h

/*
 * libecho — kernel-bypass I/O engine on io_uring
 *
 * Async API, SQPOLL, registered buffer pool,
 * linked operations, mixed epoll + io_uring event loop.
 *
 * SPDX-License-Identifier: MIT
 */

#ifndef ECHO_H
#define ECHO_H

#define _POSIX_C_SOURCE 200809L

#include <linux/io_uring.h>
#include <stddef.h>
#include <stdint.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/syscall.h>
#include <sys/uio.h>
#include <time.h>
#include <unistd.h>

/* ── barrier ───────────────────────────────────────────────────────────── */
#if defined(__x86_64__) || defined(__i386__)
#define le_smp_rmb() __asm__ __volatile__("lfence" ::: "memory")
#define le_smp_wmb() __asm__ __volatile__("sfence" ::: "memory")
#elif defined(__aarch64__)
#define le_smp_rmb() __asm__ __volatile__("dmb ishld" ::: "memory")
#define le_smp_wmb() __asm__ __volatile__("dmb ishst" ::: "memory")
#else
#define le_smp_rmb() __sync_synchronize()
#define le_smp_wmb() __sync_synchronize()
#endif
#define le_read_barrier()  le_smp_rmb()
#define le_write_barrier() le_smp_wmb()

/* ── callback ──────────────────────────────────────────────────────────── */
struct le_ring;
typedef void (*le_cb)(struct le_ring *, struct io_uring_cqe *, void *);

struct le_task {
	le_cb   cb;
	void   *udata;
};

/* ── ring ──────────────────────────────────────────────────────────────── */
struct le_ring {
	int                fd;
	unsigned           sq_entries, cq_entries, flags;

	void              *sq_ring;
	unsigned          *sq_head, *sq_tail, *sq_mask, *sq_flags, *sq_dropped;
	__u32             *sq_array;
	struct io_uring_sqe *sqes;

	void              *cq_ring;
	unsigned          *cq_head, *cq_tail, *cq_mask, *cq_flags, *cq_overflow;
	struct io_uring_cqe *cqes;

	size_t             sq_ring_sz, cq_ring_sz, sqes_sz;
	unsigned           sqe_tail, cqe_head;

	struct le_task     *tasks;
	unsigned            max_tasks;
	int                 event_fd, running;
};

/* ── buffer pool ───────────────────────────────────────────────────────── */
struct le_bufpool {
	struct iovec  *iovecs;
	void         **bufs;
	size_t         buf_sz;
	unsigned       nr;
	int           *free_list;
	unsigned       free_head;
};

/* ══════════════════════════════════════════════════════════════════════════ */
/* ring init                                                               */
/* ══════════════════════════════════════════════════════════════════════════ */

int  le_ring_init(struct le_ring *, unsigned entries, unsigned flags);
int  le_ring_init_async(struct le_ring *, unsigned entries,
    unsigned flags, unsigned max_tasks);
void le_ring_destroy(struct le_ring *);
static inline int le_ring_fd(struct le_ring *r) { return r->fd; }

/* ══════════════════════════════════════════════════════════════════════════ */
/* SQE acquisition                                                        */
/* ══════════════════════════════════════════════════════════════════════════ */

static inline struct io_uring_sqe *
le_get_sqe(struct le_ring *ring)
{
	unsigned head, next, tail, mask;
	struct io_uring_sqe *sqe;

	mask = *ring->sq_mask;
	head = __atomic_load_n(ring->sq_head, __ATOMIC_ACQUIRE);
	tail = ring->sqe_tail;
	next = tail + 1;

	if (next - head > ring->sq_entries)
		return NULL;

	ring->sq_array[tail & mask] = tail & mask;
	ring->sqe_tail = next;

	sqe = &ring->sqes[tail & mask];
	__builtin_memset(sqe, 0, sizeof(*sqe));
	return sqe;
}

static inline struct io_uring_sqe *
le_get_sqes(struct le_ring *ring, unsigned n)
{
	unsigned head, next, tail, mask, i;
	struct io_uring_sqe *sqe;

	if (n == 0 || n > ring->sq_entries) return NULL;

	mask = *ring->sq_mask;
	head = __atomic_load_n(ring->sq_head, __ATOMIC_ACQUIRE);
	tail = ring->sqe_tail;
	next = tail + n;

	if (next - head > ring->sq_entries) return NULL;

	for (i = 0; i < n; i++)
		ring->sq_array[(tail + i) & mask] = (tail + i) & mask;
	ring->sqe_tail = next;

	sqe = &ring->sqes[tail & mask];
	__builtin_memset(sqe, 0, sizeof(*sqe) * n);
	return sqe;
}

static inline void
le_sqe_zero(struct io_uring_sqe *sqe)
{ __builtin_memset(sqe, 0, sizeof(*sqe)); }

/* ══════════════════════════════════════════════════════════════════════════ */
/* operation prep (inline, zero overhead) — must be BEFORE async helpers  */
/* ══════════════════════════════════════════════════════════════════════════ */

static inline void
le_prep_rw(int op, struct io_uring_sqe *sqe, int fd,
    const void *addr, unsigned len, __u64 offset)
{
	sqe->opcode = (__u8)op; sqe->fd = fd; sqe->off = offset;
	sqe->addr = (__u64)(uintptr_t)addr; sqe->len = len; sqe->rw_flags = 0;
}

static inline void le_prep_read (struct io_uring_sqe *s, int fd, void *b, unsigned n, __u64 off)
{ le_prep_rw(IORING_OP_READ,  s, fd, b,         n, off); }
static inline void le_prep_write(struct io_uring_sqe *s, int fd, const void *b, unsigned n, __u64 off)
{ le_prep_rw(IORING_OP_WRITE, s, fd, (void *)b, n, off); }

static inline void le_prep_readv (struct io_uring_sqe *s, int fd, const struct iovec *i, unsigned nr, __u64 off)
{ le_prep_rw(IORING_OP_READV,  s, fd, i, nr, off); }
static inline void le_prep_writev(struct io_uring_sqe *s, int fd, const struct iovec *i, unsigned nr, __u64 off)
{ le_prep_rw(IORING_OP_WRITEV, s, fd, (const void *)i, nr, off); }

static inline void le_prep_read_fixed (struct io_uring_sqe *s, int fd, void *b, unsigned n, __u64 off, int idx)
{ le_prep_read(s, fd, b, n, off); s->buf_index = (__u16)idx; }
static inline void le_prep_write_fixed(struct io_uring_sqe *s, int fd, const void *b, unsigned n, __u64 off, int idx)
{ le_prep_write(s, fd, b, n, off); s->buf_index = (__u16)idx; }

static inline void le_prep_fsync  (struct io_uring_sqe *s, int fd, unsigned fl)
{ s->opcode = IORING_OP_FSYNC; s->fd = fd; s->fsync_flags = fl; }

static inline void le_prep_send(struct io_uring_sqe *s, int fd, const void *b, size_t n, int fl)
{ s->opcode = IORING_OP_SEND; s->fd = fd; s->addr = (__u64)(uintptr_t)b; s->len = (__u32)n; s->msg_flags = (__u32)fl; }

static inline void le_prep_recv(struct io_uring_sqe *s, int fd, void *b, size_t n, int fl)
{ s->opcode = IORING_OP_RECV; s->fd = fd; s->addr = (__u64)(uintptr_t)b; s->len = (__u32)n; s->msg_flags = (__u32)fl; }

static inline void le_prep_sendmsg(struct io_uring_sqe *s, int fd, const struct msghdr *m, unsigned fl)
{ s->opcode = IORING_OP_SENDMSG; s->fd = fd; s->addr = (__u64)(uintptr_t)m; s->len = 1; s->msg_flags = fl; }
static inline void le_prep_recvmsg(struct io_uring_sqe *s, int fd, struct msghdr *m, unsigned fl)
{ s->opcode = IORING_OP_RECVMSG; s->fd = fd; s->addr = (__u64)(uintptr_t)m; s->len = 1; s->msg_flags = fl; }
static inline void le_prep_recvmsg_multishot(struct io_uring_sqe *s, int fd, struct msghdr *m, unsigned fl)
{ le_prep_recvmsg(s, fd, m, fl); s->ioprio |= IORING_RECV_MULTISHOT; }

static inline void le_prep_accept(struct io_uring_sqe *s, int fd, struct sockaddr *a, socklen_t *al, int fl)
{ s->opcode = IORING_OP_ACCEPT; s->fd = fd; s->addr = (__u64)(uintptr_t)a; s->addr2 = (__u64)(uintptr_t)al; s->accept_flags = (__u32)fl; }
static inline void le_prep_accept_multishot(struct io_uring_sqe *s, int fd, struct sockaddr *a, socklen_t *al, int fl)
{ le_prep_accept(s, fd, a, al, fl); s->ioprio |= IORING_ACCEPT_MULTISHOT; }

static inline void le_prep_connect(struct io_uring_sqe *s, int fd, const struct sockaddr *a, socklen_t al)
{ s->opcode = IORING_OP_CONNECT; s->fd = fd; s->addr2 = (__u64)(uintptr_t)a; s->addr = (__u64)al; }
static inline void le_prep_close  (struct io_uring_sqe *s, int fd)
{ s->opcode = IORING_OP_CLOSE; s->fd = fd; }

static inline void le_prep_timeout(struct io_uring_sqe *s, struct __kernel_timespec *ts, unsigned cnt, unsigned fl)
{ s->opcode = IORING_OP_TIMEOUT; s->fd = -1; s->off = (__u64)(uintptr_t)ts; s->len = 1; s->timeout_flags = fl; s->addr = (__u64)cnt; }
static inline void le_prep_link_timeout(struct io_uring_sqe *s, struct __kernel_timespec *ts, unsigned fl)
{ le_prep_timeout(s, ts, 0, fl); s->flags |= IOSQE_IO_LINK; }

static inline void le_prep_cancel (struct io_uring_sqe *s, void *ud, int fl)
{ s->opcode = IORING_OP_ASYNC_CANCEL; s->fd = -1; s->addr = (__u64)(uintptr_t)ud; s->cancel_flags = (__u32)fl; }
static inline void le_prep_poll_add   (struct io_uring_sqe *s, int fd, unsigned mask)
{ s->opcode = IORING_OP_POLL_ADD; s->fd = fd; s->poll_events = (__u16)mask; }
static inline void le_prep_poll_remove(struct io_uring_sqe *s, __u64 ud)
{ s->opcode = IORING_OP_POLL_REMOVE; s->fd = -1; s->addr = ud; }
static inline void le_prep_splice(struct io_uring_sqe *s, int fi, __u64 oi, int fo, __u64 oo, unsigned n, unsigned fl)
{ s->opcode = IORING_OP_SPLICE; s->fd = fo; s->splice_fd_in = fi; s->off = oo; s->addr = (__u64)oi; s->len = n; s->splice_flags = fl; }
static inline void le_prep_nop    (struct io_uring_sqe *s)
{ s->opcode = IORING_OP_NOP; }

/* ══════════════════════════════════════════════════════════════════════════ */
/* SQE / CQE helpers                                                      */
/* ══════════════════════════════════════════════════════════════════════════ */

static inline void le_sqe_set_flags(struct io_uring_sqe *s, unsigned fl)
{ s->flags |= (__u8)fl; }
static inline void le_sqe_set_data(struct io_uring_sqe *s, void *d)
{ s->user_data = (__u64)(uintptr_t)d; }
static inline void *le_cqe_get_data(const struct io_uring_cqe *c)
{ return (void *)(uintptr_t)c->user_data; }

/* ══════════════════════════════════════════════════════════════════════════ */
/* submission / completion                                                */
/* ══════════════════════════════════════════════════════════════════════════ */

int le_submit(struct le_ring *);
int le_submit_and_wait(struct le_ring *, unsigned wait_nr);
int le_flush(struct le_ring *);

static inline int le_peek_cqe(struct le_ring *ring, struct io_uring_cqe **out)
{
	unsigned h, t;
	le_read_barrier();
	h = __atomic_load_n(ring->cq_head, __ATOMIC_ACQUIRE);
	t = __atomic_load_n(ring->cq_tail, __ATOMIC_ACQUIRE);
	if (h == t) return -1;
	*out = &ring->cqes[h & *ring->cq_mask];
	return 0;
}

static inline void le_cqe_seen(struct le_ring *ring, struct io_uring_cqe *c)
{
	unsigned h;
	if (!c) return;
	h = __atomic_load_n(ring->cq_head, __ATOMIC_RELAXED);
	le_write_barrier();
	__atomic_store_n(ring->cq_head, h + 1, __ATOMIC_RELEASE);
}

int le_wait_cqe(struct le_ring *, struct io_uring_cqe **);
int le_wait_cqes(struct le_ring *, struct io_uring_cqe **, unsigned count);
int le_wait_cqe_timeout(struct le_ring *, struct __kernel_timespec *ts,
    struct io_uring_cqe **);

/* ══════════════════════════════════════════════════════════════════════════ */
/* async submission                                                       */
/* ══════════════════════════════════════════════════════════════════════════ */

int le_submit_task(struct le_ring *, struct io_uring_sqe *, le_cb, void *);
int le_process(struct le_ring *, unsigned min_wait);
int le_cancel(struct le_ring *, void *user_data);

static inline int
le_submit_nop_async(struct le_ring *r, le_cb cb, void *ud)
{ struct io_uring_sqe *s = le_get_sqe(r); if (!s) return -1;
  le_prep_nop(s); return le_submit_task(r, s, cb, ud); }

static inline int
le_submit_read_async(struct le_ring *r, int fd, void *b, unsigned n, __u64 off, le_cb cb, void *ud)
{ struct io_uring_sqe *s = le_get_sqe(r); if (!s) return -1;
  le_prep_read(s, fd, b, n, off); return le_submit_task(r, s, cb, ud); }

static inline int
le_submit_write_async(struct le_ring *r, int fd, const void *b, unsigned n, __u64 off, le_cb cb, void *ud)
{ struct io_uring_sqe *s = le_get_sqe(r); if (!s) return -1;
  le_prep_write(s, fd, b, n, off); return le_submit_task(r, s, cb, ud); }

static inline int
le_submit_send_async(struct le_ring *r, int fd, const void *b, size_t n, int fl, le_cb cb, void *ud)
{ struct io_uring_sqe *s = le_get_sqe(r); if (!s) return -1;
  le_prep_send(s, fd, b, n, fl); return le_submit_task(r, s, cb, ud); }

static inline int
le_submit_recv_async(struct le_ring *r, int fd, void *b, size_t n, int fl, le_cb cb, void *ud)
{ struct io_uring_sqe *s = le_get_sqe(r); if (!s) return -1;
  le_prep_recv(s, fd, b, n, fl); return le_submit_task(r, s, cb, ud); }

static inline int
le_submit_close_async(struct le_ring *r, int fd, le_cb cb, void *ud)
{ struct io_uring_sqe *s = le_get_sqe(r); if (!s) return -1;
  le_prep_close(s, fd); return le_submit_task(r, s, cb, ud); }

/* ══════════════════════════════════════════════════════════════════════════ */
/* linked ops                                                             */
/* ══════════════════════════════════════════════════════════════════════════ */

static inline void
le_link_read_then_write(struct io_uring_sqe *r, int fi, void *b, unsigned n,
    __u64 oi, struct io_uring_sqe *w, int fo, __u64 oo)
{ le_prep_read(r, fi, b, n, oi); le_sqe_set_flags(r, IOSQE_IO_LINK);
  le_prep_write(w, fo, b, n, oo); }

static inline void
le_link_read_then_send(struct io_uring_sqe *r, int fi, void *b, unsigned n,
    __u64 oi, struct io_uring_sqe *w, int fd, int fl)
{ le_prep_read(r, fi, b, n, oi); le_sqe_set_flags(r, IOSQE_IO_LINK);
  le_prep_send(w, fd, b, n, fl); }

static inline void
le_link_recv_then_write(struct io_uring_sqe *r, int fd, void *b, size_t n,
    int rfl, struct io_uring_sqe *w, int fo, __u64 oo)
{ le_prep_recv(r, fd, b, n, rfl); le_sqe_set_flags(r, IOSQE_IO_LINK);
  le_prep_write(w, fo, b, n, oo); }

int le_submit_chain(struct le_ring *, struct io_uring_sqe **sqes,
    unsigned n, le_cb cb, void *udata);

/* ══════════════════════════════════════════════════════════════════════════ */
/* event loop — io_uring + epoll                                          */
/* ══════════════════════════════════════════════════════════════════════════ */

int  le_loop_init(struct le_ring *);
int  le_loop_run(struct le_ring *, int epfd, int timeout_ms);
void le_loop_stop(struct le_ring *);

/* ══════════════════════════════════════════════════════════════════════════ */
/* buffer pool                                                            */
/* ══════════════════════════════════════════════════════════════════════════ */

int  le_bufpool_init(struct le_ring *, struct le_bufpool *, unsigned nr, size_t sz);
void le_bufpool_destroy(struct le_ring *, struct le_bufpool *);
int  le_bufpool_get(struct le_bufpool *);
void le_bufpool_put(struct le_bufpool *, int idx);
static inline void *le_bufpool_buf(struct le_bufpool *p, int idx)
{ return p->bufs[idx]; }

/* ══════════════════════════════════════════════════════════════════════════ */
/* sync wrappers (blocking, single op)                                    */
/* ══════════════════════════════════════════════════════════════════════════ */

ssize_t le_read (struct le_ring *, int fd, void *buf, size_t count, __u64 off);
ssize_t le_write(struct le_ring *, int fd, const void *buf, size_t count, __u64 off);
ssize_t le_send (struct le_ring *, int fd, const void *buf, size_t len, int flags);
ssize_t le_recv (struct le_ring *, int fd, void *buf, size_t len, int flags);
int     le_accept (struct le_ring *, int fd, struct sockaddr *, socklen_t *, int);
int     le_connect(struct le_ring *, int fd, const struct sockaddr *, socklen_t);
int     le_close  (struct le_ring *, int fd);
int     le_fsync  (struct le_ring *, int fd);
int     le_nop    (struct le_ring *);

/* ══════════════════════════════════════════════════════════════════════════ */
/* registration                                                           */
/* ══════════════════════════════════════════════════════════════════════════ */

int le_register_buffers(struct le_ring *, const struct iovec *, unsigned);
int le_unregister_buffers(struct le_ring *);
int le_register_files(struct le_ring *, const int *, unsigned);
int le_unregister_files(struct le_ring *);
int le_register_eventfd(struct le_ring *, int);
int le_unregister_eventfd(struct le_ring *);

const char *le_strerror(int err);
const char *le_version(void);

#endif /* ECHO_H */