IORING_OP_LINK_TIMEOUT 还可为链接序列添加超时，避免因单个慢磁盘导致整个操作序列无限阻塞：

struct __kernel_timespec ts = { .tv_sec = 0, .tv_nsec = 100*1000*1000 }; // 100ms

struct io_uring_sqe *sqe_link = io_uring_get_sqe(˚);
io_uring_prep_link_timeout(sqe_link, &ts, 0);
io_uring_sqe_set_data(sqe_link, make_user_data(OP_LINK_TIMEOUT, txn_id));

四、核心实现：完整磁盘 IO 引擎

4.1 引擎初始化

typedef struct {
    struct io_uring      ring;           // io_uring 实例
    int                  direct_fd;       // O_DIRECT 文件描述符
    int                  eventfd;         // eventfd 用于唤醒等待线程
    BufferPool           pool;            // 缓冲区池
    IOScheduler          scheduler;       // IO 调度器
    _Atomic(uint64_t)    inflight;        // 在途 IO 计数
    uint32_t             queue_depth;     // 队列深度
} DiskEngine;

int disk_engine_init(DiskEngine *engine, const char *path) {
    // 参数配置
    struct io_uring_params params = {0};
    params.flags = IORING_SETUP_SQPOLL | IORING_SETUP_ATTACH_WQ;
    params.sq_thread_idle = 2000;
    
    // 双队列模式：使用 dedicated 工作线程
    if (io_uring_queue_init_params(QUEUE_DEPTH, &engine->ring, ¶ms) < 0 xss=removed>ring, ¶ms) < 0>sqpoll_mode = false;
    } else {
        engine->sqpoll_mode = true;
    }

    // 打开文件（O_DIRECT + O_CREAT + O_RDWR）
    int flags = O_RDWR | O_CREAT | O_DIRECT;
    engine->direct_fd = open(path, flags, 0644);
    if (engine->direct_fd < 0 xss=removed>direct_fd = open(path, flags, 0644);
    }

    // 预分配缓冲区池
    buffer_pool_init(&engine->pool, BUF_SIZE, BUF_COUNT);
    io_uring_register_buffers(&engine->ring, engine->pool.iovecs, BUF_COUNT);

    // 注册固定文件
    int fds[] = { engine->direct_fd };
    io_uring_register_files(˚, fds, 1);

    // 注册 eventfd（用于 event-driven 唤醒）
    engine->eventfd = eventfd(0, EFD_NONBLOCK);
    io_uring_register_eventfd(&engine->ring, engine->eventfd);

    engine->queue_depth = QUEUE_DEPTH;
    return 0;
}

4.2 批量提交引擎

// 批量 IO 收集 + 周期提交（μs 级批量窗口）
void engine_flush(DiskEngine *engine, uint64_t deadline_us) {
    // 从调度器取出待处理请求，批量填充 SQEs
    int count = 0;
    struct io_uring_sqe *sqes[MAX_BATCH];

    while (count < MAX>scheduler)) {
        IORequest *req = scheduler_dequeue(&engine->scheduler);
        if (!req) break;

        // 检测并执行合并
        IORequest *merged = try_merge(engine, req);
        if (merged) {
            req = merged;
        }

        // 准备 SQE
        struct io_uring_sqe *sqe = io_uring_get_sqe(&engine->ring);
        if (req->op == OP_READ) {
            io_uring_prep_read(sqe, FIXED_FILE_IDX, 
                               req->buf, req->len, req->offset);
        } else {
            io_uring_prep_write(sqe, FIXED_FILE_IDX,
                                req->buf, req->len, req->offset);
        }
        sqe->flags |= IOSQE_FIXED_FILE;
        // 设置优先级（通过 user_data 传递，由收割线程排序）
        sqe->user_data = (uint64_t)req;
        sqes[count++] = sqe;
    }

    if (count > 0) {
        io_uring_submit(&engine->ring);
        atomic_fetch_add(&engine->inflight, count);
    }
}

4.3 完成收割与缓冲区自动归还

void engine_reap(DiskEngine *engine, int min_completions) {
    struct io_uring_cqe *cqes[MAX_BATCH];
    int count = io_uring_peek_batch_cqe(&engine->ring, cqes, MAX_BATCH);

    for (int i = 0; i < count xss=removed xss=removed>user_data;

        if (cqe->res < 0>res == -EAGAIN) {
                // 资源暂时不可用，重入队列
                scheduler_requeue(&engine->scheduler, req);
                continue;
            } else if ((int)cqe->res == -EBUSY) {
                // 文件被固定 IO 占用，尝试降级为 buffered
                submit_buffered_fallback(engine, req);
                continue;
            } else {
                req->error = (int)cqe->res;
            }
        } else {
            req->bytes_transferred = cqe->res;
            
            // 处理 Registered Buffer 选择
            if (cqe->flags & IORING_CQE_F_BUFFER) {
                int buf_idx = cqe->flags >> IORING_CQE_BUFFER_SHIFT;
                req->buf = engine->pool.bufs[buf_idx];
                req->buf_owned = true;
                // 标记需要归还
            }
        }

        // 回调通知完成
        if (req->cb) {
            req->cb(req);
        }

        atomic_fetch_sub(&engine->inflight, 1);
    }

    if (count > 0) {
        io_uring_cq_advance(&engine->ring, count);
    }
}

4.4 主事件循环

void engine_run(DiskEngine *engine) {
    while (engine->running) {
        // 1. 收割已完成 IO
        engine_reap(engine, 0);  // 非阻塞收割

        // 2. 处理超时 LO_IO 请求
        scheduler_reap_timeouts(&engine->scheduler);

        // 3. 填充并批量提交 SQEs
        engine_flush(engine, NOW_US() + BATCH_WINDOW_US);

        // 4. 如果没有在途 IO，进入阻塞等待（eventfd 唤醒）
        if (atomic_load(&engine->inflight) == 0) {
            uint64_t val;
            read(engine->eventfd, &val, sizeof(val));
        }
    }
}

// SQPOLL 模式优化主循环
void engine_run_sqpoll(DiskEngine *engine) {
    while (engine->running) {
        // 非阻塞收割
        engine_reap(engine, 0);
        
        // 直接往 SQ 写 SQE（SQPOLL 线程会自动消费）
        engine_fill_sq(engine);

        // 进入等待（有完成或有超时）
        struct io_uring_cqe *cqe;
        int ret = io_uring_wait_cqe_timeout(&engine->ring, &cqe, &batch_timeout);
        if (ret == 0) {
            // 处理完成…
        }
    }
}

五、性能优化进阶

5.1 writev 聚合：vectored IO 减少 submit 次数

对于需要写入多个不连续缓冲区的场景（如 LSM-Tree 的 SSTable 写入），writev/ prep_writev 可用一次系统调用完成：

struct iovec iov[3] = {
    { .iov_base = header,  .iov_len = HEADER_SIZE },
    .iov_base = data_buf, .iov_len = data_len },
    { .iov_base = footer,  .iov_len = FOOTER_SIZE }
};

struct io_uring_sqe *sqe = io_uring_get_sqe(˚);
io_uring_prep_writev(sqe, fd, iov, 3, file_offset);
sqe->flags |= IOSQE_FIXED_FILE;

5.2 fallocate 预分配 + 模式选择

文件系统预分配可减少文件碎片化并提高 O_DIRECT 的写入性能：

// 预分配空间（仅占用 inode，不实际写盘）
struct io_uring_sqe *sqe = io_uring_get_sqe(˚);
io_uring_prep_fallocate(sqe, fd, 0, 0, file_size);
sqe->flags |= IOSQE_FIXED_FILE;
io_uring_submit(˚);

// FALLOC_FL_ZERO_RANGE 预清零（安全场景）
io_uring_prep_fallocate(sqe, fd, FALLOC_FL_ZERO_RANGE, 0, file_size);

5.3 IO 优先级：RWF_HIPRI 与轮询式低延迟路径

NVMe 支持两个独立的 SQ/CQ 队列——_admin_ 队列（高优先级）和 _io_ 队列（普通优先级）。通过 RWF_HIPRI 标志，可以让特定 IO 走高优先级路径：

// 高优先级同步读（适用于 WAL 读取、元数据读取）
struct io_uring_sqe *sqe = io_uring_get_sqe(˚);
io_uring_prep_read(sqe, fd, buf, len, offset);
sqe->ioprio |= IOPRIO_PRIO_VALUE(IOPRIO_CLASS_RT, 0);  // RT 最高级
sqe->flags |= IOSQE_FIXED_FILE;

// 检查内核是否支持轮询式 IO
if (io_uring_params.features & IORING_FEAT_SQPOLL_NONFIXED) {
    // 可使用 NVMe polling mode（NVMe Polling Mode）
    // 避免中断延迟，纯用户态轮询完成
}

5.4 io_uring 与 SPDK 的协同

在极端低延迟场景（延迟 < 10>

// SPDK 负责：NVMe SQ/CQ 用户态轮询（内核旁路）
// io_uring 负责：标准文件系统接口（XFS/ext4）+ 网络事件
//
// 数据路径：
// 写入 → SPDK NVMe SQ → 硬件 → 完成 → 用户态回调
// 读缓存 → io_uring + Page Cache（IO 命中 → 0 设备 IO）
//
// 适用场景：NVMe-oF 目标端、分布式 KV 引擎

六、生产环境部署与调优

6.1 内核参数调优

# /etc/sysctl.conf

# --- 块层队列 ---
# NVMe 队列深度（默认 64，建议 1024-4096）
echo 1024 > /sys/block/nvme0n1/queue/nr_requests
echo 2048 > /sys/block/nvme0n1/queue/read_ahead_kb
echo none > /sys/block/nvme0n1/queue/scheduler  # NVMe 无需 IO 调度器

# --- 内存 ---
# O_DIRECT 缓冲区必须挂在固定内存（mlock）
vm.min_free_kbytes = 262144          # 256MB 最小空闲内存（保护 O_DIRECT）
vm.dirty_ratio = 10                  # 脏页比率限制（降低尾部延迟）
vm.dirty_background_ratio = 5

# --- io_uring ---
# 每个进程最大 ring 数
sysctl -w kernel.io_uring_max_rings = 4096

# SQPOLL 线程 CPU 亲和性
taskset -c 0-1 ionice -c 1 -n 0 sqpoll_thread
ionice -c 2 -n 4 engine_thread

6.2 SQPOLL 权限与 CPU 绑核

SQPOLL 内核线程运行于发起 io_uring 的进程上下文中，需要 CAP_SYS_ADMIN 权限或者 CAP_SYS_NICE 绑核：

# 方式 1: SQPOLL 绑核（0 号核心处理 SQ，1 号核心处理 CQ）
struct io_uring_params params = {0};
params.flags = IORING_SETUP_SQPOLL | IORING_SETUP_SQ_AFF;
params.sq_thread_cpu = 0;  // SQ 处理核心
io_uring_queue_init_params(QUEUE_DEPTH, ˚, ¶ms);

// 方式 2: cgroup 限制（容器化部署）
// io_uring worker 线程受 cgroup cpu.max 限制

// 方式 3: CAP_SYS_ADMIN 权限
// SQPOLL 需 CAP_SYS_ADMIN（较新内核放宽了要求）
$ sudo setcap cap_sys_admin+ep /path/to/engine

6.3 关键性能基准对比

* 测试平台：AMD EPYC 7763 × 2, 512GB DDR4-3200, Samsung PM9A3 NVMe, Linux 6.5 内核, 队列深度 256

6.4 降级与故障恢复策略

typedef enum {
    IO_MODE_SQPOLL_FIXED,     // 最高性能：SQPOLL + Fixed Buffers + Fixed Files
    IO_MODE_SQPOLL,           // SQPOLL + 动态缓冲区
    IO_MODE_INTERRUPT_FIXED,  // 中断模式 + Fixed Buffers  
    IO_MODE_INTERRUPT,        // 中断模式 + 动态缓冲区
    IO_MODE_PREAD             // 降级为同步 pread/pwrite
} IOMode;

IOMode engine_select_mode(DiskEngine *engine) {
    if (access("/proc/sys/kernel/io_uring", F_OK) != 0)
        return IO_MODE_PREAD;  // 内核不支持 io_uring
    
    // 检查 NDirect 对齐是否可用
    if (posix_memalign_test(4096, 1024 * 1024) != 0)
        return IO_MODE_INTERRUPT;
    
    // 尝试 SQPOLL
    struct io_uring_params test_params = {0};
    test_params.flags = IORING_SETUP_SQPOLL;
    if (io_uring_queue_init_params(1, &test_ring, &test_params) < 0>

// 内核版本宏 (include/uapi/linux/io_uring.h)
#define IOURNG_FEATURE(dtype, flag) \
    ((params.features & (flag)) ? "✅ " : "❌ "), IOURNG_DESC(flag)

printf("=== io_uring 特性探测 ===\n");
printf("%s SQPOLL_NONFIXED（非固定文件 SQPOLL）\n", IOURNG_FEATURE(params, IORING_FEAT_SQPOLL_NONFIXED));
printf("%s BUFFER_SELECT（自动缓冲区选择）\n",        IOURNG_FEATURE(params, IORING_FEAT_BUFFER_SELECT));
printf("%s SUBMIT_STABLE（SQ 门铃一次性提交）\n",   IOURNG_FEATURE(params, IORING_FEAT_SUBMIT_STABLE));
printf("%s RW_CUR_POS（自动文件位置更新）\n",       IOURNG_FEATURE(params, IORING_FEAT_RW_CUR_POS));
printf("%s CUR_PERSONALITY（per-SQE personality）\n", IOURNG_FEATURE(params, IORING_FEAT_CUR_PERSONALITY));
printf("%s FAST_POLL（SQPOLL 快速轮询）\n",        IOURNG_FEATURE(params, IORING_FEAT_FAST_POLL));
printf("%s 32BIT CQE（7 字节 CQE，节省带宽）\n",  IOURNG_FEATURE(params, IORING_FEAT_NATIVE_WORKERS));

// 关键最低版本要求
// Linux 5.1  - io_uring 基础（io_uring_setup/enter/register）
// Linux 5.4  - Fixed Buffers、Fixed Files
// Linux 5.10 - SQPOLL + 非固定文件
// Linux 5.15 - BUFFER_SELECT、Linked SQEs 改进
// Linux 6.1  - IORING_SETUP_DEFER_TASKRUN（减少 task_work 开销）
// Linux 6.6  - IORING_SETUP_SUBMIT_ALL（失败安全模式）