io_uring Ağ
Sıfır Kopya Yüksek Performans Ağ

00 io_uring ağ motivasyonu — epoll karşılaştırması

epoll, yıllardır yüksek performanslı Linux ağ sunucularının omurgasıdır. Ancak her hazır fd için ayrı sistem çağrısı yapılması, kullanıcı-kernel bellek kopyaları ve tek-shot operasyonlar önemli overhead kaynakları olmaya devam eder. io_uring bu sınırları kökünden aşar.

epoll modelinin syscall maliyeti

Klasik epoll tabanlı bir echo sunucusunda tek bir istek-yanıt döngüsü şu sistem çağrılarını gerektirir: epoll_wait() ile hazır fd'yi öğren, recv() ile veriyi al, send() ile yanıtı gönder. Her sistem çağrısı bağlam geçişi (context switch) maliyeti taşır. Yüksek bağlantı sayısında bu maliyet birikerek CPU bütçesinin önemli bir bölümünü sistem çağrı altyapısına harcar.

epoll tabanlı echo:
  epoll_wait()   → kernel/user geçişi 1
  recv(fd, ...)  → kernel/user geçişi 2 + bellek kopyası
  send(fd, ...)  → kernel/user geçişi 3 + bellek kopyası
  Toplam: 3 syscall + 2 kopya / istek

io_uring tabanlı echo (SQPOLL + zero-copy):
  SQE'yi ring'e yaz → syscall YOK (SQPOLL açıksa)
  CQE'yi ring'den oku → syscall YOK
  send_zc          → bellek kopyası YOK
  Toplam: 0 syscall + 0 kopya / istek (ideal durumda)

Performans karşılaştırma özeti

Temel io_uring ağ operasyonları

01 SQ/CQ ring tekrarı — io_uring_submit, io_uring_wait_cqe

io_uring çift yönlü ring buffer mimarisini hatırlamak, ağ operasyonlarının nasıl birbirine zincirlendiğini anlamak için gereklidir. SQE üretim, SQE gönderim ve CQE tüketim üç bağımsız süreçtir.

Ring buffer yapısı

io_uring kurulumunda kernel iki ring ve bir SQE dizisi ayırır. Submission Queue (SQ) ring, SQE dizisine indeks tutar; baş (tail) user tarafından ilerletilir. Completion Queue (CQ) ring, tamamlanan operasyonların sonuçlarını tutar; baş user, kuyruk kernel tarafından güncellenir. Her iki ring de mmap() ile kullanıcı alanına eşlenir; erişim sistem çağrısız gerçekleşir.

User-space                       Kernel
  SQE'yi sq.sqes[tail]'a yaz
  sq.tail++ (atomik)
  io_uring_submit() [opsiyonel]  → kernel SQE'leri okur
                                 → operasyonları başlatır
                                 → tamamlanınca cq.cqes[cq_tail]'a yazar
                                 → cq.tail++ (atomik)
  io_uring_wait_cqe() / peek
  CQE'yi cq.cqes[cq_head]'dan oku
  cq_head++ (atomik)             ← kernel: CQE okundu

liburing API özeti

#include <liburing.h>

struct io_uring ring;

/* Kurulum: 256 giriş kapasiteli ring */
io_uring_queue_init(256, &ring, 0);

/* SQE al (ring'den bir slot al) */
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);

/* Operasyonu konfigüre et */
io_uring_prep_recv(sqe, client_fd, buf, buf_len, 0);
sqe->user_data = (uint64_t)conn;  /* tag: hangi bağlantı? */

/* Kernel'a gönder */
io_uring_submit(&ring);

/* Tamamlanma bekle */
struct io_uring_cqe *cqe;
io_uring_wait_cqe(&ring, &cqe);   /* bloklayıcı */
/* veya: io_uring_peek_cqe — bloklayıcı değil */

/* Sonucu işle */
if (cqe->res < 0)
    fprintf(stderr, "recv failed: %s\n", strerror(-cqe->res));
else
    process_data(buf, cqe->res);

/* CQE'yi serbest bırak */
io_uring_cqe_seen(&ring, cqe);

/* Temizlik */
io_uring_queue_exit(&ring);

Toplu gönderim ve tüketim

/* Birden fazla SQE gönder: tek syscall */
struct io_uring_sqe *sqe;
for (int i = 0; i < n_ready; i++) {
    sqe = io_uring_get_sqe(&ring);
    io_uring_prep_send(sqe, fds[i], bufs[i], lens[i], 0);
    sqe->user_data = (uint64_t)i;
}
io_uring_submit(&ring);  /* tüm N SQE tek syscall ile */

/* Birden fazla CQE tüket */
unsigned head;
struct io_uring_cqe *cqe;
int processed = 0;
io_uring_for_each_cqe(&ring, head, cqe) {
    handle_completion(cqe);
    processed++;
}
io_uring_cq_advance(&ring, processed);  /* toplu serbest bırak */

io_uring_params ve flag'ler

02 IORING_OP_RECV ve IORING_OP_SEND — temel ağ operasyonları

RECV ve SEND operasyonları io_uring'in en temel ağ ilkeleridir. recv(2) ve send(2) sistem çağrılarının async karşılığı olarak çalışırlar; ancak bağlı oldukları soketin non-blocking olması gerekmez.

RECV operasyonu

#include <liburing.h>
#include <sys/socket.h>

#define BUF_SIZE 4096

static void submit_recv(struct io_uring *ring, int fd,
                         char *buf, struct conn_state *conn)
{
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
    if (!sqe) {
        /* Ring dolu — flush et ve tekrar dene */
        io_uring_submit(ring);
        sqe = io_uring_get_sqe(ring);
    }

    io_uring_prep_recv(sqe, fd, buf, BUF_SIZE, 0);

    /* user_data ile bağlantı context'ini etiketle */
    io_uring_sqe_set_data(sqe, conn);

    /* IOSQE_FIXED_FILE: registered file kullan (daha hızlı) */
    /* sqe->flags |= IOSQE_FIXED_FILE; */
}

static void handle_recv_cqe(struct io_uring_cqe *cqe)
{
    struct conn_state *conn = io_uring_cqe_get_data(cqe);
    int bytes = cqe->res;

    if (bytes == 0) {
        /* Bağlantı kapatıldı */
        close_connection(conn);
        return;
    }
    if (bytes < 0) {
        if (bytes == -ENOBUFS) {
            /* Buffer ring doldu — multishot için yeniden dene */
            resubmit_recv(conn);
        } else {
            fprintf(stderr, "recv error: %s\n", strerror(-bytes));
            close_connection(conn);
        }
        return;
    }

    process_data(conn, conn->recv_buf, bytes);

    /* Bir sonraki recv'i hemen gönder */
    submit_recv(conn->ring, conn->fd, conn->recv_buf, conn);
}

SEND operasyonu

static void submit_send(struct io_uring *ring, int fd,
                         const char *data, size_t len,
                         struct conn_state *conn)
{
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);

    io_uring_prep_send(sqe, fd, data, len, MSG_NOSIGNAL);
    io_uring_sqe_set_data(sqe, conn);

    /* Kısmi gönderim durumunda otomatik yeniden dene */
    sqe->flags |= IOSQE_IO_LINK;  /* sonraki SQE ile zincirle */
}

static void handle_send_cqe(struct io_uring_cqe *cqe)
{
    struct conn_state *conn = io_uring_cqe_get_data(cqe);
    int sent = cqe->res;

    if (sent < 0) {
        fprintf(stderr, "send error: %s\n", strerror(-sent));
        close_connection(conn);
        return;
    }

    conn->sent_bytes += sent;
    if (conn->sent_bytes < conn->total_bytes) {
        /* Kısmi gönderim — kalanı gönder */
        submit_send(conn->ring, conn->fd,
                    conn->send_buf + conn->sent_bytes,
                    conn->total_bytes - conn->sent_bytes,
                    conn);
        io_uring_submit(conn->ring);
    }
}

IORING_OP_RECVMSG ve IORING_OP_SENDMSG

recvmsg ve sendmsg operasyonları, scatter-gather I/O (iovec dizisi) ve kontrol mesajı (cmsg) desteği için kullanılır. UDP socket'lerde çok kullanışlıdır: tek SQE ile kaynak adres bilgisi alınabilir.

/* UDP recv: kaynak adresi de al */
struct msghdr msg = {
    .msg_name    = &src_addr,
    .msg_namelen = sizeof(src_addr),
    .msg_iov     = &iov,
    .msg_iovlen  = 1,
};
iov.iov_base = buf;
iov.iov_len  = BUF_SIZE;

struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
io_uring_prep_recvmsg(sqe, udp_fd, &msg, 0);
io_uring_sqe_set_data(sqe, conn);

03 Multishot recv — IORING_RECV_MULTISHOT ve buffer ring

Multishot recv, tek bir SQE ile kernel'ın sürekli olarak veri almasını sağlar. Her veri gelişinde bir CQE üretilir; SQE yeniden gönderilmek zorunda kalınmaz. Buffer ring ile birleşince hem SQE hem de bellek yönetimi otomatikleşir.

Multishot recv kavramı

Normal recv modunda: recv → CQE → yeni recv SQE → recv → CQE döngüsü SQE gönderim maliyeti taşır. Multishot modunda: tek recv SQE → N×CQE — her veri paketinde yeni SQE gerekmez. Soket kapatılana veya hata oluşana kadar SQE aktif kalır.

/* Multishot recv SQE hazırla */
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);

io_uring_prep_recv_multishot(sqe, client_fd, NULL, 0, 0);
/* Not: buf=NULL, len=0 — buffer ring'den otomatik alınır */

sqe->buf_group = BUF_GRP_ID;    /* hangi buffer grubundan al */
sqe->flags    |= IOSQE_BUFFER_SELECT;
io_uring_sqe_set_data(sqe, conn);

/* CQE'de: cqe->flags & IORING_CQE_F_MORE != 0 ise multishot devam ediyor */
static void handle_multishot_recv(struct io_uring_cqe *cqe)
{
    struct conn_state *conn = io_uring_cqe_get_data(cqe);
    int bytes = cqe->res;

    if (!(cqe->flags & IORING_CQE_F_MORE)) {
        /* Multishot sona erdi — yeniden başlat */
        resubmit_multishot_recv(conn);
    }

    if (bytes <= 0) return;

    /* Buffer ring'den hangi buffer kullanıldı? */
    int buf_id = cqe->flags >> IORING_CQE_BUFFER_SHIFT;
    char *buf  = get_buffer(buf_id);

    process_data(conn, buf, bytes);
    return_buffer(buf_id);  /* buffer'ı gruba iade et */
}

Buffer ring kurulumu

#define BUF_GRP_ID   0
#define BUF_COUNT    64
#define BUF_SIZE     4096

struct io_uring_buf_ring *buf_ring;
void *bufs_mem;

static int setup_buffer_ring(struct io_uring *ring)
{
    /* Ring belleği ayır: hizalı olmalı */
    size_t ring_sz = sizeof(struct io_uring_buf_ring)
                   + BUF_COUNT * sizeof(struct io_uring_buf);
    posix_memalign((void **)&buf_ring, 4096, ring_sz);

    /* Buffer belleği ayır */
    bufs_mem = mmap(NULL, BUF_COUNT * BUF_SIZE,
                    PROT_READ | PROT_WRITE,
                    MAP_ANONYMOUS | MAP_PRIVATE, -1, 0);

    /* io_uring'e buffer ring'i kaydet */
    struct io_uring_buf_reg reg = {
        .ring_addr    = (uint64_t)buf_ring,
        .ring_entries = BUF_COUNT,
        .bgid         = BUF_GRP_ID,
    };
    int ret = io_uring_register_buf_ring(ring, &reg, 0);
    if (ret) return ret;

    /* Her buffer'ı gruba ekle */
    io_uring_buf_ring_init(buf_ring);
    for (int i = 0; i < BUF_COUNT; i++) {
        void *buf_addr = (char *)bufs_mem + i * BUF_SIZE;
        io_uring_buf_ring_add(buf_ring, buf_addr, BUF_SIZE,
                              i, io_uring_buf_ring_mask(BUF_COUNT), i);
    }
    io_uring_buf_ring_advance(buf_ring, BUF_COUNT);
    return 0;
}

Buffer iade döngüsü

Buffer ring'deki buffer'lar tüketildikten sonra geri iade edilmelidir. İade yapılmaması halinde kernel yeni recv için buffer bulamaz ve ENOBUFS döner. Veriyi işledikten hemen sonra buffer'ı iade etmek en güvenli yaklaşımdır.

static void return_buffer(struct io_uring *ring, int buf_id)
{
    void *buf_addr = (char *)bufs_mem + buf_id * BUF_SIZE;
    int mask = io_uring_buf_ring_mask(BUF_COUNT);

    io_uring_buf_ring_add(buf_ring, buf_addr, BUF_SIZE,
                          buf_id, mask, 0);
    io_uring_buf_ring_advance(buf_ring, 1);
}

04 SQPOLL — kernel taraflı polling ve CPU pin

SQPOLL modunda kernel, SQ ring'i sürekli izleyen ayrı bir kernel thread başlatır. Kullanıcı SQE yazdıktan sonra io_uring_submit() çağırmak zorunda kalmaz; kernel thread bunu tespit edip operasyonu doğrudan başlatır. Sıfır syscall ile maksimum throughput elde edilir.

SQPOLL kurulumu

#include <liburing.h>

int setup_sqpoll_ring(struct io_uring *ring, int sq_thread_cpu)
{
    struct io_uring_params params = {
        .flags = IORING_SETUP_SQPOLL
               | IORING_SETUP_SQ_AFF,  /* CPU sabitleme */
        .sq_thread_cpu  = sq_thread_cpu,
        .sq_thread_idle = 2000,  /* ms — idle sonrası thread uyur */
    };

    int ret = io_uring_queue_init_params(4096, ring, &params);
    if (ret) {
        perror("io_uring_queue_init_params");
        /* EPERM: CAP_SYS_NICE veya root gerekir */
        return ret;
    }

    printf("SQPOLL aktif, kernel thread CPU=%d\n", sq_thread_cpu);
    return 0;
}

SQPOLL ile SQE gönderimi

/* SQPOLL aktif: io_uring_submit() çağrısı gerekmez */
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
io_uring_prep_recv(sqe, fd, buf, len, 0);
io_uring_sqe_set_data(sqe, conn);
/* sq_tail güncelleme: liburing bunu io_uring_get_sqe içinde yapar */
/* Kernel thread sq_tail değişimini atomik okur ve işler */

/* Ancak: kernel thread uyumuşsa (sq_thread_idle geçtiyse)
   wake-up için submit çağrısı gerekebilir */
if (io_uring_sq_need_wakeup(ring))
    io_uring_submit(ring);  /* sadece wake-up için — hızlı */

SQPOLL güvenlik gereksinimleri

SQPOLL CPU mimarisi

SQPOLL en yüksek faydayı özel CPU'larda verir. Uygulama thread'i ve SQPOLL thread'i aynı NUMA node'undaki iki farklı CPU'ya sabitlenirse L3 cache paylaşımı ile ring buffer erişim gecikmesi minimuma iner. Gömülü ARM çok çekirdekli sistemlerde (Cortex-A72 gibi) CPU 0 işletim sistemi görevlerine, CPU 1 SQPOLL thread'ine, CPU 2–3 uygulama thread'lerine ayrılabilir.

/* SQPOLL thread'ini CPU 1'e sabitle */
struct io_uring_params params = {
    .flags          = IORING_SETUP_SQPOLL | IORING_SETUP_SQ_AFF,
    .sq_thread_cpu  = 1,
    .sq_thread_idle = 10000,  /* 10 sn idle → uyku */
};

/* Uygulama thread'ini CPU 2'ye sabitle */
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(2, &cpuset);
sched_setaffinity(0, sizeof(cpuset), &cpuset);

05 Zero-copy send — IORING_OP_SEND_ZC ve notification CQE

Zero-copy send, kernel'ın gönderilecek veriyi kullanıcı buffer'ından kopyalamak yerine doğrudan DMA ile ağ kartına göndermesini sağlar. Büyük veri transferlerinde CPU kopyalama overhead'ini sıfıra indirger.

Geleneksel send vs zero-copy send

Geleneksel send():
  User buffer → kernel socket buffer (kopya 1) → NIC DMA buffer (kopya 2) → ağ

Zero-copy send (SEND_ZC):
  User buffer → NIC DMA buffer (kopya 0) → ağ
  Kernel user buffer'ını pin eder (page pin), DMA tamamlandıktan sonra serbest bırakır

Two-phase completion modeli

Zero-copy send tamamlanması iki aşamalıdır:

1. Birinci CQE: SQE gönderim işlemi tamamlandı — cqe->res gönderilen byte sayısı, IORING_CQE_F_MORE flag set
2. İkinci CQE (notification): DMA tamamlandı, user buffer artık güvenle kullanılabilir — IORING_CQE_F_NOTIF flag set, cqe->res == 0

Kullanıcı, notification CQE gelmeden buffer'ı değiştirmemelidir; aksi hâlde DMA sürerken veri bozulur.

#include <linux/io_uring.h>
#include <liburing.h>

#define OP_SEND_ZC  0x10  /* user_data opcode tag */

static void submit_send_zc(struct io_uring *ring, int fd,
                             const char *data, size_t len,
                             struct conn_state *conn)
{
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);

    /* SEND_ZC: kernel 6.0+ */
    io_uring_prep_send_zc(sqe, fd, data, len, MSG_NOSIGNAL, 0);

    /* user_data üst bitlerine opcode etiket */
    io_uring_sqe_set_data64(sqe, (uint64_t)conn | OP_SEND_ZC);
}

static void handle_send_zc_cqe(struct io_uring_cqe *cqe)
{
    struct conn_state *conn = (struct conn_state *)(cqe->user_data & ~OP_SEND_ZC);

    if (cqe->flags & IORING_CQE_F_NOTIF) {
        /* Notification: DMA tamam, buffer serbest */
        conn->send_buf_in_use = false;
        return;
    }

    /* Birinci tamamlanma: gönderim sonucu */
    if (cqe->res < 0) {
        fprintf(stderr, "send_zc error: %s\n", strerror(-cqe->res));
        close_connection(conn);
        return;
    }

    conn->sent_bytes += cqe->res;
    /* cqe->flags & IORING_CQE_F_MORE: notification CQE bekliyor */
}

Registered buffer ile SEND_ZC

SEND_ZC en verimli şekilde önceden registered buffer'larla çalışır. Buffer önceden kernel'a registered edildiğinden her send'de yeniden pin işlemi gerekmez; DMA setup maliyeti azalır.

/* Buffer'ı kaydet */
struct iovec iov = { .iov_base = send_buf, .iov_len = BUF_SIZE };
io_uring_register_buffers(ring, &iov, 1);

/* SEND_ZC with fixed buffer */
struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
io_uring_prep_send_zc_fixed(sqe, fd, send_buf, data_len,
                              MSG_NOSIGNAL, 0,
                              0 /* buf_index */);

Zero-copy eşik değeri

Zero-copy'nin faydası veri boyutuyla doğru orantılıdır. Küçük paketlerde (<4 KB) page pin overhead'i kopyalama maliyetinden yüksek olabilir. Genel kural: 16 KB ve üzerinde veri gönderiminde zero-copy tercih edilmesi önerilir. Benchmark ile platforma özgü eşik bulunmalıdır.

06 Fixed buffers ve registered files

io_uring'in en önemli optimizasyonlarından biri, sık kullanılan buffer'ları ve fd'leri önceden kernel'a kaydetmektir. Her operasyonda tekrarlanan page table walk ve file table lookup maliyetleri tek seferlik registration ile amortize edilir.

IORING_REGISTER_BUFFERS

#define NUM_BUFS  16
#define BUF_SIZE  65536  /* 64 KB */

char      *bufs[NUM_BUFS];
struct iovec iovs[NUM_BUFS];

/* Buffer'ları ayır ve kaydet */
for (int i = 0; i < NUM_BUFS; i++) {
    posix_memalign((void **)&bufs[i], 4096, BUF_SIZE);
    iovs[i].iov_base = bufs[i];
    iovs[i].iov_len  = BUF_SIZE;
}

int ret = io_uring_register_buffers(&ring, iovs, NUM_BUFS);
if (ret) perror("io_uring_register_buffers");

/* Fixed buffer ile okuma */
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_read_fixed(sqe, fd, bufs[0], BUF_SIZE, 0,
                          0 /* buf_index: iovs[0] */);

/* Fixed buffer ile yazma */
sqe = io_uring_get_sqe(&ring);
io_uring_prep_write_fixed(sqe, fd, bufs[1], data_len, 0, 1);

/* Güncelleme: belirli buffer'ı güncelle (yeni alan ayarla) */
io_uring_register_buffers_update_tag(&ring, 0, &iovs[0], NULL, 1);

/* Kayıt sil */
io_uring_unregister_buffers(&ring);

IORING_REGISTER_FILES

#define MAX_CONNS  1024
int registered_fds[MAX_CONNS];

/* Başlangıçta tüm slotları -1 ile doldur (boş) */
memset(registered_fds, -1, sizeof(registered_fds));

/* Kaydet */
io_uring_register_files(&ring, registered_fds, MAX_CONNS);

/* Yeni bağlantı geldiğinde: slota yerleştir */
int slot = alloc_fd_slot();
registered_fds[slot] = client_fd;
io_uring_register_files_update(&ring, slot, &client_fd, 1);

/* Registered file ile recv */
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_recv(sqe, slot, buf, BUF_SIZE, 0);
sqe->flags |= IOSQE_FIXED_FILE;  /* fixed file kullan */

/* Bağlantı kapandığında slotu temizle */
int minus_one = -1;
io_uring_register_files_update(&ring, slot, &minus_one, 1);
free_fd_slot(slot);
close(client_fd);

IORING_OP_SOCKET ve IORING_OP_CLOSE

/* Yeni soket oluşturma — async (kernel 5.19+) */
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_socket(sqe, AF_INET, SOCK_STREAM, 0, 0);
sqe->file_index = IORING_FILE_INDEX_ALLOC;  /* otomatik slot ata */

/* CQE'de: res = registered file indeksi */

/* fd kapatma — async */
sqe = io_uring_get_sqe(&ring);
io_uring_prep_close_direct(sqe, slot);  /* registered file'ı kapat ve serbest bırak */

Registered kaynak karşılaştırması

07 io_uring sunucu mimarisi — accept döngüsü ve bağlantı state machine

io_uring tabanlı yüksek performanslı sunucu, tek thread'de binlerce eş zamanlı bağlantıyı yönetebilir. Kilit nokta: her bağlantı için explicit state machine ve asla bloklamayan bir olay döngüsü.

Bağlantı state machine

CONN_ACCEPT
    |  accept_multishot SQE → CQE: yeni fd
    v
CONN_RECV (multishot)
    |  recv_multishot SQE → N×CQE: veri geldi
    |  (state değişmeden yeni veri gelmeye devam eder)
    v (iş emri dolduğunda)
CONN_SEND
    |  send SQE → CQE: veri gönderildi
    v
CONN_RECV  (döngü devam)
    |
    v (bağlantı kapandığında)
CONN_CLOSE
    |  close_direct SQE → CQE
    v
CONN_FREE (pool'a iade)

accept_multishot ile bağlantı kabul

struct conn_state {
    int          fd;           /* soket fd */
    int          fd_slot;      /* registered file slotu */
    char        *recv_buf;
    char        *send_buf;
    size_t       send_len;
    int          state;
    /* ... */
};

static void submit_accept_multishot(struct io_uring *ring,
                                     int listen_fd,
                                     struct sockaddr_in *addr,
                                     socklen_t *addrlen)
{
    struct io_uring_sqe *sqe = io_uring_get_sqe(ring);
    io_uring_prep_multishot_accept(sqe, listen_fd,
                                    (struct sockaddr *)addr,
                                    addrlen, 0);
    sqe->file_index = IORING_FILE_INDEX_ALLOC; /* otomatik registered */
    io_uring_sqe_set_data64(sqe, ACCEPT_TAG);
}

static void handle_accept_cqe(struct io_uring *ring,
                                struct io_uring_cqe *cqe)
{
    if (cqe->res < 0) {
        if (!(cqe->flags & IORING_CQE_F_MORE))
            resubmit_accept(ring);  /* multishot sona erdi */
        return;
    }

    int slot = cqe->res;  /* IORING_FILE_INDEX_ALLOC: registered slot */
    struct conn_state *conn = alloc_conn();
    conn->fd_slot = slot;

    submit_recv_multishot(ring, conn);
}

Ana olay döngüsü

int server_loop(struct io_uring *ring, int listen_fd)
{
    struct sockaddr_in client_addr;
    socklen_t          addrlen = sizeof(client_addr);

    /* İlk accept SQE */
    submit_accept_multishot(ring, listen_fd, &client_addr, &addrlen);
    io_uring_submit(ring);

    while (running) {
        struct io_uring_cqe *cqe;
        unsigned head;
        int processed = 0;

        /* En az 1 CQE bekle */
        io_uring_wait_cqe(ring, &cqe);

        /* Tüm hazır CQE'leri işle */
        io_uring_for_each_cqe(ring, head, cqe) {
            uint64_t tag = cqe->user_data;

            if (tag == ACCEPT_TAG) {
                handle_accept_cqe(ring, cqe);
            } else {
                struct conn_state *conn = (struct conn_state *)tag;
                dispatch_cqe(ring, conn, cqe);
            }
            processed++;
        }
        io_uring_cq_advance(ring, processed);

        /* SQE kuyruğunu flush et (SQPOLL değilse) */
        if (!sqpoll_mode)
            io_uring_submit(ring);
    }
    return 0;
}

Bağlantı bağlantılandırma: SQE zinciri (IOSQE_IO_LINK)

/* recv + send'i zincirle: recv tamamlanmadan send başlamaz */
struct io_uring_sqe *recv_sqe = io_uring_get_sqe(ring);
io_uring_prep_recv(recv_sqe, conn->fd_slot, conn->recv_buf, BUF_SIZE, 0);
recv_sqe->flags |= IOSQE_IO_LINK | IOSQE_FIXED_FILE;
io_uring_sqe_set_data(recv_sqe, conn);

struct io_uring_sqe *send_sqe = io_uring_get_sqe(ring);
io_uring_prep_send(send_sqe, conn->fd_slot, conn->send_buf, 0, 0);
/* len=0: recv tamamlandığında gerçek uzunluk CQE'den alınır */
send_sqe->flags |= IOSQE_FIXED_FILE;
io_uring_sqe_set_data(send_sqe, conn);

/* zincir: recv başarısız olursa send iptal edilir (CQE: -ECANCELED) */

08 Performans karşılaştırması ve gömülü sistemlerde kullanım

io_uring ağ performansı, doğru konfigürasyonla epoll tabanlı sunucuları önemli ölçüde geride bırakır. Gömülü sistemlerde ise kaynak kısıtları nedeniyle özellik seçimi ve kernel versiyon uyumu kritiktir.

Throughput karşılaştırması (ölçüm verileri)

Not: Değerler 4 çekirdek x86-64, kernel 6.1, 100 eş zamanlı bağlantı için gösterge niteliğindedir. Gerçek değerler donanıma, ağ kartına ve iş yüküne göre değişir.

Gömülü sistemlerde io_uring ağ

# Minimum kernel konfigürasyonu
CONFIG_IO_URING=y
CONFIG_NET=y
CONFIG_INET=y

# io_uring ağ için önerilen kernel versiyonu
# Kernel 5.19+: multishot recv, buffer ring tam destek
# Kernel 6.0+:  SEND_ZC, socket direct
# Kernel 6.1+:  LTS — gömülü için önerilen

# ARM Cortex-A53 (Raspberry Pi 3 gibi) için
# SQPOLL ile temkinli olun: idle CPU'yu tam kullanır
# sq_thread_idle değerini yükseğe ayarlayın: 50000 ms

Gömülü sistem konfigürasyon önerileri

liburing kurulumu ve cross-derleme

# Kaynak derle (Yocto recipe veya manuel):
git clone https://github.com/axboe/liburing.git
cd liburing

# Cross-derleme (ARM64):
./configure --prefix=/sysroot/usr \
            CC=aarch64-linux-gnu-gcc \
            AR=aarch64-linux-gnu-ar
make -j$(nproc)
make install

# Versiyon uyumu:
# liburing 2.3+ → kernel 5.19 özellikleri (multishot recv, buffer ring)
# liburing 2.5+ → kernel 6.0 özellikleri (SEND_ZC)

# Statik bağlama:
aarch64-linux-gnu-gcc -static myserver.c \
    -I /sysroot/usr/include \
    /sysroot/usr/lib/liburing.a \
    -o myserver_arm64

Hata ayıklama ve sorun giderme

# Kernel io_uring istatistikleri
cat /proc/sys/kernel/io_uring_disabled  # 0=açık, 1=kapalı, 2=sadece root
cat /proc/PID/fdinfo/FD_NO              # io_uring instance bilgisi

# strace ile SQE/CQE izleme
strace -e io_uring_setup,io_uring_enter,io_uring_register ./myserver

# perf ile hotspot analizi
perf stat -e io_uring:io_uring_submit_sqe,io_uring:io_uring_complete_task ./myserver

# CQ overflow tespiti
# /proc/PID/fdinfo'da cq_overflow alanını kontrol et
# Çözüm: CQ ring boyutunu artır (IORING_SETUP_CQSIZE ile SQ'dan büyük yap)