当前位置：首页 > news >正文

解放io_uring编程：liburing实战指南与经典cat示例解析

news 2025/7/29 7:01:57

一、为什么需要liburing？

二、liburing核心API解析

1. 初始化与清理

2. 请求生命周期管理

3. 操作准备函数（部分）

三、经典实战：liburing版cat程序

关键代码解析

1. io_uring 核心工作流程

2. 柔性数组成员（struct file_info）

3. 内存对齐（posix_memalign）

4. 同步与异步的平衡

5. 内存管理注意事项

关键函数posix_memalign

1、函数原型与参数

2、为什么需要内存对齐？

3、代码中的具体应用

4、对比普通malloc

5、何时必须使用posix_memalign？

6、内存释放注意事项

7、总结

四、编译与运行指南

安装liburing

编译程序

运行示例

五、性能对比：liburing vs 传统cat

六、liburing高级技巧

1. 批量提交优化

2. 完成事件批处理

3. 固定资源注册

七、常见问题解决方案

1. 请求未执行？

2. 性能未达预期？

3. 内存泄漏？

八、扩展应用场景

1. 高性能Web服务器

2. 数据库日志写入

3. 网络代理

九、下一步：深入io_uring内核

在探索了io_uring的底层原理后，你是否渴望一个更简洁的开发方式？liburing正是为此而生！这个由io_uring创始人Jens Axboe开发的官方库，将复杂的内存映射和队列管理封装成简洁API。本文将通过经典cat示例，带你轻松掌握liburing的精髓。

一、为什么需要liburing？

直接使用io_uring原始接口需要：

手动计算内存映射大小
管理环形缓冲区索引
处理复杂的系统调用参数

liburing的价值：

二、liburing核心API解析

1. 初始化与清理

// 初始化io_uring实例
int io_uring_queue_init(unsigned entries, // 队列深度struct io_uring *ring, // ring结构体unsigned flags); // 配置标志// 清理资源
void io_uring_queue_exit(struct io_uring *ring);

2. 请求生命周期管理

// 获取提交队列条目(SQE)
struct io_uring_sqe *io_uring_get_sqe(struct io_uring *ring);

// 提交请求
int io_uring_submit(struct io_uring *ring);

// 等待完成事件
int io_uring_wait_cqe(struct io_uring *ring,struct io_uring_cqe **cqe_ptr);

// 标记完成事件已处理
void io_uring_cqe_seen(struct io_uring *ring, struct io_uring_cqe *cqe);

3. 操作准备函数（部分）

// 准备读操作
void io_uring_prep_read(struct io_uring_sqe *sqe,int fd,         // 文件描述符void *buf,      // 缓冲区unsigned nbytes,// 字节数off_t offset);   // 偏移量

// 准备写操作
void io_uring_prep_write(struct io_uring_sqe *sqe,int fd,const void *buf,unsigned nbytes,off_t offset);

// 准备关闭文件
void io_uring_prep_close(struct io_uring_sqe *sqe, int fd);

三、经典实战：liburing版cat程序

以下是基于liburing官方示例的精简版cat实现，该代码使用 liburing 库（封装了 io_uring 系统调用）实现异步文件读取，支持读取多个文件并将内容输出到控制台。核心流程为：初始化 io_uring 实例 → 提交文件读取请求 → 等待读取完成 → 输出内容 → 清理资源：

#include <fcntl.h>       // 包含open、O_RDONLY等定义
#include <stdio.h>       // 包含fputc、fprintf等I/O函数
#include <string.h>      // 包含内存操作函数
#include <sys/stat.h>    // 包含fstat、struct stat等定义
#include <sys/ioctl.h>   // 包含ioctl系统调用
#include <liburing.h>    // 包含io_uring相关函数（liburing库）
#include <stdlib.h>      // 包含malloc、free等内存函数

// 定义io_uring队列深度（同时处理的最大请求数）
#define QUEUE_DEPTH 1
// 定义文件读取的块大小（1024字节）
#define BLOCK_SZ    1024

// 文件信息结构体（使用柔性数组成员存储多个iovec）
struct file_info {off_t file_sz;               // 文件总大小（字节）struct iovec iovecs[];       // 柔性数组：存储每个读取块的缓冲区信息（地址+长度）
};

/*** 获取文件大小（支持普通文件和块设备）* @param fd 已打开的文件描述符* @return 成功返回文件大小（字节），失败返回-1*/
off_t get_file_size(int fd) {struct stat st;
// 先用fstat获取文件基本信息if(fstat(fd, &st) < 0) {perror("fstat failed");  // 打印错误信息return -1;}
// 处理块设备（如硬盘分区）if (S_ISBLK(st.st_mode)) {unsigned long long bytes;// 通过ioctl获取块设备总大小（BLKGETSIZE64返回64位大小）if (ioctl(fd, BLKGETSIZE64, &bytes) != 0) {perror("ioctl failed");return -1;}return bytes;} // 处理普通文件else if (S_ISREG(st.st_mode)) {return st.st_size;  // 普通文件直接返回st_size}
// 不支持的文件类型return -1;
}

/*** 将缓冲区内容输出到控制台（逐个字符）* @param buf 缓冲区地址* @param len 缓冲区长度（字节）*/
void output_to_console(char *buf, int len) {while (len--) {  // 循环输出每个字符fputc(*buf++, stdout);  // 输出当前字符并移动指针}
}

/*** 等待io_uring完成事件，获取结果并打印文件内容* @param ring io_uring实例指针* @return 0成功，1失败*/
int get_completion_and_print(struct io_uring *ring) {struct io_uring_cqe *cqe;  // 完成队列条目（CQE）
// 等待完成队列中有可用条目（阻塞直到有事件）int ret = io_uring_wait_cqe(ring, &cqe);if (ret < 0) {perror("io_uring_wait_cqe failed");  // 等待失败return 1;}
// 检查CQE结果（res为读取的字节数，<0表示失败）if (cqe->res < 0) {fprintf(stderr, "Async readv failed (error: %d)\n", cqe->res);return 1;}
// 通过CQE获取关联的用户数据（file_info指针）struct file_info *fi = io_uring_cqe_get_data(cqe);
// 计算总块数（与submit_read_request中一致）int blocks = (int) fi->file_sz / BLOCK_SZ;if (fi->file_sz % BLOCK_SZ != 0) {  // 处理最后一个不完整块blocks++;}
// 循环输出每个块的内容for (int i = 0; i < blocks; i++) {output_to_console(fi->iovecs[i].iov_base, fi->iovecs[i].iov_len);}
// 标记CQE为已处理（内核可重用该条目）io_uring_cqe_seen(ring, cqe);
// 释放分配的内存（避免内存泄漏）for (int i = 0; i < blocks; i++) {free(fi->iovecs[i].iov_base);  // 释放每个块的对齐缓冲区}free(fi);  // 释放file_info（包含iovec数组）// 注意：原代码未释放buff，此处可补充free(buff)
return 0;
}

/*** 提交文件读取请求到io_uring* @param file_path 目标文件路径* @param ring io_uring实例指针* @return 0成功，1失败*/
int submit_read_request(char *file_path, struct io_uring *ring) {// 打开文件（只读模式）int file_fd = open(file_path, O_RDONLY);if (file_fd < 0) {perror("open failed");  // 打开失败return 1;}
// 获取文件大小off_t file_sz = get_file_size(file_fd);if (file_sz < 0) {  // 失败处理close(file_fd);return 1;}
off_t bytes_remaining = file_sz;  // 剩余未读字节数off_t offset = 0;                 // 当前读取偏移量（初始0）int current_block = 0;            // 当前块索引// 计算总块数（文件大小/块大小，有余数则+1）int blocks = (int) file_sz / BLOCK_SZ;if (file_sz % BLOCK_SZ != 0) {blocks++;}
// 分配file_info结构体+柔性数组（iovec数组）// 柔性数组大小为blocks * sizeof(struct iovec)struct file_info *fi = malloc(sizeof(*fi) +                  // file_info结构体大小(sizeof(struct iovec) * blocks)  // iovec数组大小);if (!fi) {  // 分配失败fprintf(stderr, "malloc failed for file_info and iovecs\n");close(file_fd);return 1;}
// 分配总缓冲区（大小为文件大小，原代码未使用，可用于后续整合数据）char *buff = malloc(file_sz);if (!buff) {fprintf(stderr, "Unable to allocate total buffer\n");free(fi);close(file_fd);return 1;}
// 为每个块分配对齐的缓冲区，并初始化iovecwhile (bytes_remaining > 0) {// 计算当前块的读取长度（不超过BLOCK_SZ）off_t bytes_to_read = bytes_remaining;if (bytes_to_read > BLOCK_SZ) {bytes_to_read = BLOCK_SZ;}
// 更新偏移量（实际读取偏移由readv的offset参数控制，此处仅记录）offset += bytes_to_read;
// 为当前块分配对齐的缓冲区（地址对齐到BLOCK_SZ）void *buf;// posix_memalign确保缓冲区地址是BLOCK_SZ的整数倍（提升I/O效率）if (posix_memalign(&buf, BLOCK_SZ, BLOCK_SZ) != 0) {perror("posix_memalign failed");// 释放已分配资源free(buff);free(fi);close(file_fd);return 1;}
// 初始化当前块的iovec（缓冲区地址和长度）fi->iovecs[current_block].iov_base = buf;  // 对齐的缓冲区地址fi->iovecs[current_block].iov_len = bytes_to_read;  // 读取长度
current_block++;  // 移动到下一块bytes_remaining -= bytes_to_read;  // 减少剩余字节数}
fi->file_sz = file_sz;  // 记录文件总大小
// 获取一个空闲的提交队列条目（SQE）struct io_uring_sqe *sqe = io_uring_get_sqe(ring);if (!sqe) {  // 没有可用SQE（队列满）fprintf(stderr, "Failed to get SQE (queue full)\n");// 释放资源free(buff);free(fi);close(file_fd);return 1;}
// 准备SQE为readv操作（向量读取）// 参数：文件描述符、iovec数组、块数、读取偏移量（0表示从文件开头）io_uring_prep_readv(sqe, file_fd, fi->iovecs, blocks, 0);
// 将file_info指针关联到SQE（完成时通过CQE获取）io_uring_sqe_set_data(sqe, fi);
// 提交SQE到io_uring（将请求提交给内核）if (io_uring_submit(ring) < 0) {perror("io_uring_submit failed");// 释放资源free(buff);free(fi);close(file_fd);return 1;}
return 0;  // 提交成功
}

/*** 主函数：初始化io_uring，处理输入文件，清理资源*/
int main(int argc, char *argv[]) {struct io_uring ring;  // io_uring实例
// 检查输入参数（至少需要一个文件名）if (argc < 2) {fprintf(stderr, "Usage: %s [file name] <[file name] ...>\n", argv[0]);return 1;}
// 初始化io_uring队列// 参数：队列深度（QUEUE_DEPTH=1）、ring实例、 flags=0（默认配置）io_uring_queue_init(QUEUE_DEPTH, &ring, 0);
// 循环处理每个输入文件for (int i = 1; i < argc; i++) {// 提交文件读取请求int ret = submit_read_request(argv[i], &ring);if (ret) {fprintf(stderr, "Error reading file: %s\n", argv[i]);io_uring_queue_exit(&ring);  // 清理资源return 1;}
// 等待读取完成并打印内容get_completion_and_print(&ring);}
// 清理io_uring资源io_uring_queue_exit(&ring);return 0;
}

关键代码解析

1. io_uring 核心工作流程

代码遵循 io_uring 的标准使用流程：

初始化：io_uring_queue_init 创建队列（指定深度和配置）。
提交请求：
- 通过 io_uring_get_sqe 获取空闲 SQE（提交队列条目）。
- 用 io_uring_prep_readv 初始化 SQE 为 readv 操作（向量读取）。
- 用 io_uring_sqe_set_data 关联用户数据（file_info）。
- 用 io_uring_submit 提交请求到内核。
等待完成：io_uring_wait_cqe 阻塞等待 CQE（完成队列条目）。
处理结果：通过 io_uring_cqe_get_data 获取用户数据，处理读取内容，用 io_uring_cqe_seen 标记 CQE 为已处理。
清理：io_uring_queue_exit 释放队列资源。

2. 柔性数组成员（struct file_info）

struct file_info {off_t file_sz;struct iovec iovecs[];
};

作用：动态存储每个读取块的 iovec 信息（缓冲区地址和长度）。由于文件大小不确定，块数 blocks 动态计算，柔性数组可避免二次分配（直接在 file_info 后分配连续内存），提升内存访问效率。
分配方式：malloc(sizeof(*fi) + blocks * sizeof(struct iovec))，确保 file_info 和 iovecs 数组在连续内存中。

3. 内存对齐（posix_memalign）

posix_memalign(&buf, BLOCK_SZ, BLOCK_SZ);

作用：确保每个块的缓冲区地址对齐到 BLOCK_SZ（1024 字节）。I/O 操作（尤其是直接 I/O）对内存对齐要求严格，非对齐缓冲区可能导致性能下降或操作失败。
对比 malloc：malloc 仅保证基本对齐（如 8 字节），无法满足块设备的对齐要求，因此必须用 posix_memalign。

4. 同步与异步的平衡

代码中 QUEUE_DEPTH=1，且循环中 “提交一个请求→等待完成→处理下一个”，实际是同步处理（每次只处理一个文件）。若增大 QUEUE_DEPTH，可批量提交多个请求，再统一等待完成，实现真正的异步并行读取，提升效率（例如：先提交所有文件的读取请求，再批量等待完成）。

5. 内存管理注意事项

每个块的缓冲区通过 posix_memalign 分配，需在处理完成后用 free 释放（否则内存泄漏）。
file_info 及其柔性数组需在处理完成后释放。
代码中 char *buff 未实际使用，可根据需求保留（如整合所有块数据）或删除。

关键函数posix_memalign

在异步 I/O 场景中，posix_memalign 是一个关键函数，其作用是分配内存并确保地址对齐到指定边界。在您提供的代码中，这个函数用于为每个 I/O 块分配缓冲区，确保内存地址与块大小（BLOCK_SZ，通常为 4096 字节）对齐。

1、函数原型与参数

#include <stdlib.h>

int posix_memalign(void **memptr, size_t alignment, size_t size);

memptr：指向指针的指针，用于存储分配的内存地址。
alignment：要求的对齐边界，必须是 2 的幂（如 4096、8192）。
size：需要分配的内存大小（字节）。

返回值：成功返回 0，失败返回错误码（如ENOMEM）。

2、为什么需要内存对齐？

硬件限制 现代存储设备（如 SSD、HDD）通常以块（Block）为单位进行读写，典型块大小为 4KB。若内存地址未对齐到块边界，可能导致：
- 性能下降：需跨块读取，增加 I/O 操作次数。
- 功能错误：某些硬件或文件系统（如使用O_DIRECT标志）要求严格对齐，否则 I/O 请求会失败。
异步 I/O 优化 在io_uring中，对齐内存可减少内核与用户空间的数据拷贝，提升异步操作效率。

3、代码中的具体应用

在您提供的代码中，posix_memalign 用于为每个 I/O 块分配对齐的缓冲区：

// 为当前块分配对齐的缓冲区（posix_memalign确保内存地址按BLOCK_SZ对齐，提升I/O效率）
void *buf;
if (posix_memalign(&buf, BLOCK_SZ, BLOCK_SZ) != 0) {perror("posix_memalign failed");// 错误处理...
}// 初始化当前块的iovec：指向分配的缓冲区，长度为bytes_to_read
fi->iovecs[current_block].iov_base = buf;  // 缓冲区地址
fi->iovecs[current_block].iov_len = bytes_to_read;  // 读取长度

关键点：

对齐值（alignment）：使用BLOCK_SZ（通常为 4096），确保缓冲区地址是 4KB 的整数倍。
分配大小（size）：每次分配BLOCK_SZ字节，即使当前块实际读取的字节数不足（如文件末尾块）。
与iovec的配合：每个iovec的iov_base指向对齐的缓冲区，iov_len指定实际读取长度。

4、对比普通`malloc`

普通的malloc分配的内存通常对齐到 8 字节或 16 字节（取决于系统），无法保证 4KB 对齐。若直接使用malloc，可能导致：

I/O 性能下降：非对齐内存可能触发额外的内存拷贝。
功能错误：若后续代码使用O_DIRECT标志（直接 I/O），会因地址未对齐而失败。

5、何时必须使用`posix_memalign`？

直接 I/O（O_DIRECT）：要求缓冲区地址、长度、偏移量均对齐到块大小。
高性能场景：减少内存碎片，优化硬件访问效率。
特定硬件或驱动要求：某些设备（如 GPU、网卡）需要对齐内存以加速数据传输。