io_uring — যখন syscall নিজেই bottleneck, তখন ring buffer দিয়ে kernel-এর সাথে কথা বলা

অংশ ১ — liburing সংস্করণ

/* batch_read.c -- অনেকগুলো ফাইল io_uring দিয়ে batch করে পড়া
 * বিল্ড:  gcc -O2 batch_read.c -o batch_read -luring
 * চালান: ./batch_read /tmp/manyfiles/*.bin
 */
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <liburing.h>

#define BATCH   256
#define BUFSZ   4096

struct job {          /* প্রতিটা কাজের প্রেক্ষাপট -- pointer-টাই user_data */
    int   fd;
    char *buf;
};

int main(int argc, char **argv) {
    if (argc < 2) { fprintf(stderr, "ব্যবহার: %s file...\n", argv[0]); return 2; }

    struct io_uring ring;
    int rc = io_uring_queue_init(BATCH, &ring, 0);
    if (rc < 0) {
        fprintf(stderr, "io_uring অনুপলব্ধ (%s) -- সাধারণ read fallback\n",
                strerror(-rc));
        long t = 0; char b[BUFSZ];
        for (int i = 1; i < argc; i++) {
            int fd = open(argv[i], O_RDONLY);
            if (fd < 0) continue;
            ssize_t n = read(fd, b, BUFSZ);
            if (n > 0) t += n;
            close(fd);
        }
        printf("মোট %ld বাইট (fallback পথ)\n", t);
        return 0;
    }

    long total = 0, done = 0;
    int i = 1;

    while (i < argc) {
        int queued = 0;

        /* ---- ধাপ ১: SQ ring ভরাও, প্রতিটা SQE শুধু memory-তে লেখা ---- */
        while (i < argc && queued < BATCH) {
            int fd = open(argv[i], O_RDONLY);
            i++;
            if (fd < 0) continue;

            struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
            if (!sqe) { close(fd); break; }          /* ring পূর্ণ */

            struct job *j = malloc(sizeof *j);
            j->fd  = fd;
            j->buf = malloc(BUFSZ);

            io_uring_prep_read(sqe, fd, j->buf, BUFSZ, 0);
            io_uring_sqe_set_data(sqe, j);           /* user_data = আমাদের pointer */
            queued++;
        }
        if (queued == 0) break;

        /* ---- ধাপ ২: একটাই syscall -- queued-টা কাজ জমা দাও ---- */
        rc = io_uring_submit(&ring);
        if (rc < 0) { fprintf(stderr, "submit: %s\n", strerror(-rc)); break; }

        /* ---- ধাপ ৩: completion তুলে নাও ---- */
        for (int k = 0; k < queued; k++) {
            struct io_uring_cqe *cqe;
            rc = io_uring_wait_cqe(&ring, &cqe);
            if (rc < 0) { fprintf(stderr, "wait_cqe: %s\n", strerror(-rc)); break; }

            struct job *j = io_uring_cqe_get_data(cqe);
            if (cqe->res < 0)
                fprintf(stderr, "read ব্যর্থ: %s\n", strerror(-cqe->res));
            else
                total += cqe->res;                    /* res = কত বাইট পড়া হলো */
            done++;

            close(j->fd);
            free(j->buf);
            free(j);
            io_uring_cqe_seen(&ring, cqe);            /* CQ head++ -- slot ছেড়ে দাও */
        }
    }

    printf("মোট %ld বাইট, %ld completion\n", total, done);
    io_uring_queue_exit(&ring);
    return 0;
}

চালানো:

gcc -O2 batch_read.c -o batch_read -luring
./batch_read /tmp/manyfiles/*.bin

মোট 20480000 বাইট, 5000 completion

অংশ ২ — ঠিক একই জিনিস, কোনো লাইব্রেরি ছাড়া

liburing যা করে তা কোনো জাদু নয় — শুধু mmap আর index arithmetic। নিচের সংস্করণটা পড়লে ring-এর প্রকৃত মেকানিক্স চোখের সামনে খুলে যায়। এটাই সেই কোড যেটা io_uring_get_sqe আর io_uring_submit আসলে চালায়।

/* raw_uring.c -- liburing ছাড়া, শুধু syscall + mmap
 * বিল্ড:  gcc -O2 raw_uring.c -o raw_uring
 */
#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/mman.h>
#include <sys/syscall.h>
#include <linux/io_uring.h>

#define QD 64

/* liburing নেই, তাই syscall wrapper নিজে লিখতে হবে */
static int sys_io_uring_setup(unsigned e, struct io_uring_params *p) {
    return (int)syscall(__NR_io_uring_setup, e, p);
}
static int sys_io_uring_enter(int fd, unsigned to_submit,
                              unsigned min_complete, unsigned flags) {
    return (int)syscall(__NR_io_uring_enter, fd, to_submit,
                        min_complete, flags, NULL, 0);
}

struct ring {
    int fd;
    /* SQ */
    unsigned *sq_head, *sq_tail, *sq_mask, *sq_array;
    struct io_uring_sqe *sqes;
    /* CQ */
    unsigned *cq_head, *cq_tail, *cq_mask;
    struct io_uring_cqe *cqes;
};

static int ring_init(struct ring *r, unsigned entries) {
    struct io_uring_params p;
    memset(&p, 0, sizeof p);

    r->fd = sys_io_uring_setup(entries, &p);
    if (r->fd < 0) return -errno;

    /* --- SQ ring mmap --- */
    size_t sq_sz = p.sq_off.array + p.sq_entries * sizeof(unsigned);
    void *sq = mmap(NULL, sq_sz, PROT_READ | PROT_WRITE,
                    MAP_SHARED | MAP_POPULATE, r->fd, IORING_OFF_SQ_RING);
    if (sq == MAP_FAILED) return -errno;

    r->sq_head  = (unsigned *)((char *)sq + p.sq_off.head);
    r->sq_tail  = (unsigned *)((char *)sq + p.sq_off.tail);
    r->sq_mask  = (unsigned *)((char *)sq + p.sq_off.ring_mask);
    r->sq_array = (unsigned *)((char *)sq + p.sq_off.array);

    /* --- SQE array আলাদা mmap (আলাদা offset!) --- */
    r->sqes = mmap(NULL, p.sq_entries * sizeof(struct io_uring_sqe),
                   PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE,
                   r->fd, IORING_OFF_SQES);
    if (r->sqes == MAP_FAILED) return -errno;

    /* --- CQ ring mmap --- */
    size_t cq_sz = p.cq_off.cqes + p.cq_entries * sizeof(struct io_uring_cqe);
    void *cq = mmap(NULL, cq_sz, PROT_READ | PROT_WRITE,
                    MAP_SHARED | MAP_POPULATE, r->fd, IORING_OFF_CQ_RING);
    if (cq == MAP_FAILED) return -errno;

    r->cq_head = (unsigned *)((char *)cq + p.cq_off.head);
    r->cq_tail = (unsigned *)((char *)cq + p.cq_off.tail);
    r->cq_mask = (unsigned *)((char *)cq + p.cq_off.ring_mask);
    r->cqes    = (struct io_uring_cqe *)((char *)cq + p.cq_off.cqes);
    return 0;
}

int main(int argc, char **argv) {
    if (argc < 2) { fprintf(stderr, "ব্যবহার: %s file...\n", argv[0]); return 2; }

    struct ring r;
    int rc = ring_init(&r, QD);
    if (rc < 0) { fprintf(stderr, "setup: %s\n", strerror(-rc)); return 1; }

    static char bufs[QD][4096];
    int fds[QD];
    unsigned n = 0;
    for (int i = 1; i < argc && n < QD; i++) {
        int fd = open(argv[i], O_RDONLY);
        if (fd < 0) continue;
        fds[n] = fd;

        /* ---- SQE হাতে লিখুন ---- */
        unsigned tail  = *r.sq_tail;              /* আমরাই producer */
        unsigned index = tail & *r.sq_mask;
        struct io_uring_sqe *sqe = &r.sqes[index];

        memset(sqe, 0, sizeof *sqe);
        sqe->opcode    = IORING_OP_READ;
        sqe->fd        = fd;
        sqe->addr      = (unsigned long)bufs[n];
        sqe->len       = 4096;
        sqe->off       = 0;
        sqe->user_data = n;                        /* আমাদের ট্যাগ */

        r.sq_array[index] = index;                 /* indirection */
        __atomic_store_n(r.sq_tail, tail + 1, __ATOMIC_RELEASE);  /* ★ প্রকাশ */
        n++;
    }

    /* ---- একটাই syscall: n-টা জমা দাও, n-টার জন্য অপেক্ষা করো ---- */
    rc = sys_io_uring_enter(r.fd, n, n, IORING_ENTER_GETEVENTS);
    if (rc < 0) { perror("io_uring_enter"); return 1; }

    /* ---- CQ থেকে ফলাফল তোলো ---- */
    long total = 0;
    unsigned head = *r.cq_head;
    unsigned tail = __atomic_load_n(r.cq_tail, __ATOMIC_ACQUIRE);  /* ★ গ্রহণ */
    while (head != tail) {
        struct io_uring_cqe *cqe = &r.cqes[head & *r.cq_mask];
        if (cqe->res > 0) total += cqe->res;
        else fprintf(stderr, "job %llu: %s\n",
                     (unsigned long long)cqe->user_data, strerror(-cqe->res));
        head++;
    }
    __atomic_store_n(r.cq_head, head, __ATOMIC_RELEASE);  /* slot ফেরত দাও */

    printf("%u কাজ, মোট %ld বাইট\n", n, total);
    for (unsigned i = 0; i < n; i++) close(fds[i]);
    close(r.fd);
    return 0;
}

দুইটা ★ চিহ্নিত লাইনই পুরো protocol-এর হৃদয় — একটা release-store (submission প্রকাশ) আর একটা acquire-load (completion গ্রহণ)। বাকি সব শুধু index arithmetic। liburing-এর কয়েক হাজার লাইন কোড মূলত এই কয়েকটা লাইনকে নিরাপদ, portable আর সুবিধাজনক করে তোলার জন্য (প্রতিটা opcode-এর prep_* helper, SQPOLL-এর wakeup যুক্তি, buffer ring ব্যবস্থাপনা)।

নিজে বাড়ান

১. open+read+close-কে একটা linked চেইনে নিন। IORING_OP_OPENAT, IORING_OP_READ, IORING_OP_CLOSE তিনটা SQE-তে IOSQE_IO_LINK flag দিন, আর IORING_OP_READ-এ fd-র বদলে IOSQE_FIXED_FILE বা IORING_OP_OPENAT-এর ফলাফল ব্যবহার করুন। তারপর strace -c চালিয়ে দেখুন মোট syscall ১৫,০০০ থেকে কতে নামল। ইঙ্গিত: চেইনের মাঝে ব্যর্থতা হলে বাকিগুলো -ECANCELED পাবে — সেই ক্ষেত্রেও ঠিকভাবে হিসাব রাখুন।

২. BATCH কে ভেরিয়েবল করে একটা গ্রাফ বানান। BATCH কে argv[1] থেকে নিন, ১ থেকে ২০৪৮ পর্যন্ত দুইয়ের ঘাতে চালিয়ে প্রতিবার wall time আর io_uring_enter সংখ্যা রেকর্ড করুন। আপনি একটা স্পষ্ট hyperbola দেখবেন যা ~৬৪-১২৮-এর পর সমতল হয়ে যায় — সেটাই আপনার মেশিনে batching-এর প্রকৃত সীমা।

৩. SQPOLL চালু করুন এবং core-খরচ মাপুন। io_uring_queue_init_params দিয়ে IORING_SETUP_SQPOLL আর sq_thread_idle = 1000 দিন (CAP_SYS_NICE লাগতে পারে)। তারপর top -H-এ iou-sqp-\<pid\> thread-টার CPU% দেখুন — নিষ্ক্রিয় অবস্থাতেও ১০০% দেখলে বুঝবেন কেন এটা ডিফল্ট নয়।

৪. Registered buffer যোগ করে পার্থক্য মাপুন। io_uring_register_buffers দিয়ে সব buffer আগে register করুন, তারপর io_uring_prep_read_fixed ব্যবহার করুন। ছোট ফাইলে পার্থক্য সামান্য হবে; ৬৪ KB-র বড় read-এ ৫-১৫% পর্যন্ত দেখা যেতে পারে — কারণ তখন get_user_pages-এর খরচ আপেক্ষিকভাবে বড়।

৫. একটা সম্পূর্ণ echo server লিখুন io_uring-এ। IORING_OP_ACCEPT → IORING_OP_RECV → IORING_OP_SEND চক্র, প্রতিটা connection-এর state একটা struct-এ, user_data-তে সেই struct-এর pointer + একটা opcode tag (নিচের ২ বিটে)। এটা লেসন ২৫-এর graceful shutdown build-এর সাথে জোড়া লাগবে — তখন signalfd-এর বদলে IORING_OP_POLL_ADD দিয়ে signal সামলাতে পারবেন।