原文地址：https://www.zoucz.com/blog/2022/06/01/210c97f0-e166-11ec-9fe7-534bbf9f369d/，参与评论

为什么要讨论异步IO

之前提到的IO，不管是阻塞IO还是非阻塞IO，在数据从内核态拷贝到用户态的阶段，都需要等待一段时间。

那么例如epoll高并发场景下，如果多个socket同时有数据到来，即使用非阻塞io也会串行同步读取，这个场景使用多线程读数据或者aio，理论上也是能获得一定收益的？那么为什么epoll的网络IO实现很多都没有采用多线程和异步IO呢？我们来看一下各类存储设备的读写速度：

上图可以看到，内存的读写速度是ns级别的，而磁盘的读写速度是ms级别的，两者相差6个数量级。

所以epoll在处理网络IO的过程中，内存从内核拷贝到用户，一般情况下并不会成为瓶颈，redis单线程也能做到高并发高qps。

但是相差6个数量级的磁盘IO就不一样了，在需要做磁盘文件读写的应用场景下，即使内核有对文件读写做一些缓存优化，IO读写等待环节仍然可能成为瓶颈。

DIRECT IO模式和非DIRECT IO模式

非DIRECT IO模式

前面说到内核会对文件IO的过程做缓存优化，其实指的就是内核页缓存。

因为硬盘读写速度相对内存来说，速度实在太慢，所以为了提升对文件的读写效率，Linux 内核会以页大小（4KB）为单位，将文件划分为多数据块。当用户对文件中的某个数据块进行读写操作时，内核首先会申请一个内存页（称为页缓存）与文件中的数据块进行绑定。如下图所示：

当用户对文件进行读写时，实际上是对文件的页缓存进行读写。所以对文件进行读写操作时，会分以下两种情况进行处理：

当从文件中读取数据时，如果要读取的数据所在的页缓存已经存在，那么就直接把页缓存的数据拷贝给用户即可。否则，内核首先会申请一个空闲的内存页（页缓存），然后从文件中读取数据到页缓存，并且把页缓存的数据拷贝给用户。
当向文件中写入数据时，如果要写入的数据所在的页缓存已经存在，那么直接把新数据写入到页缓存即可。否则，内核首先会申请一个空闲的内存页（页缓存），然后从文件中读取数据到页缓存，并且把新数据写入到页缓存中。对于被修改的页缓存，内核会定时把这些页缓存刷新到文件中。

DIRECT IO模式

在数据库应用和静态文件服务应用中，实现者希望自己管理IO缓存数据，那么内核页缓存实际上是一步多余的操作。为了在这种场景下提高效率，操作系统提供了DIRECT IO模式，即不经过内核页缓存，直接读写文件本身，此时需要开发者自己做好IO缓存，否则会十分低效。

要使用 DIRECT IO，需要在 open 文件的时候加上 O_DIRECT 的 flag。比如：

1	int fd = open(fname, O_RDWR \| O_DIRECT);

使用 DIRECT IO 有一个很大的限制：buffer 的内存地址、每次读写数据的大小、文件的 offset 三者都要与底层设备的逻辑块大小对齐（一般是 512 字节）:

1 2	$ blockdev --getss /dev/vdb1 512

linux kernel aio

基本原理

在上一节中可以看到，非DIRECT模式下虽然做了缓存，但是仍然存在内核页中无缓存，需要从磁盘读取数据的情况；DIRECT模式下，更是每次都是直接与磁盘交互。

所以同步IO在数据拷贝阶段的等待会非常长。为了不让这个等待过程长期占用进程又不让出CPU，我们需要一种异步IO的机制，让进程去做别的工作，在IO的数据拷贝完成后再通知进程，这个时候使用异步IO是有价值的，如下图：

相比于同步IO，异步IO没有拷贝数据的等待阶段，而是在数据准备好之后，直接将结果返回给用户进程。

网上很多文章描述的 linux kernel aio 只支持 DIRECT IO 模式的异步IO，不过我在 linux 5.10 下写的练习代码中，非 DIRECT IO 也是可以使用的，这里和网上绝大多数文章的描述有点不一致！

kernel aio api

内核提供了一系列API函数用于执行异步IO任务，见文档

由于我也没有直接使用过这些API，就不多说了。

libaio

我用的是这个，这个库基于内核提供的aio api，进行了一些易用性的封装，可以基于这个库来开发aio代码，会简单一些。在centos上：

1	yum install -y libaio # 安装libaio库，编译的时候 -L/usr/lib64 -laio 连接libaio库 yum install -y libaio-devel # 安装libaio开发环境需要的头文件

一个基本的使用流程是：

int io_setup(int maxevents, io_context_t *ctxp); //初始化异步io
void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset);//准备一个异步io任务
void io_set_callback(struct iocb *iocb, io_callback_t cb);//设置回调函数（如果需要的话）
int io_submit(io_context_t ctx, long nr, struct iocb *ios[]);//提交异步io任务
int io_getevents(io_context_t ctx_id, long min_nr, long nr, struct io_event *events, struct timespec *timeout);//不断查询异步io任务状态，直到拿到结果
int io_destroy(io_context_t ctx); //销毁异步io

异步通知方式

io_event

linux kernel aio默认的通知方式是 io_event，可以通过调用 io_getevents 阻塞进程，IO完成时唤醒进程

/**
 * @brief 使用 io_getevents 阻塞并等待结果
 * 
 * @param ctx 
 * @param max_event 
 */
void CheckByIOGetevents(io_context_t &ctx, const int &max_event){
    ////step3. 阻塞，等待结果，执行回调
    io_event event[max_event];
    int num = io_getevents(ctx, max_event, max_event, event, NULL);
    printf("io_getevents num:%d\n", num);
    //有异步队列读取完了
    for (int i = 0; i < num; i++) {
        //调用回调对象中的回调函数（这里实际上只是打印了下信息）
        io_event &e = event[i];
        io_callback_t io_callback = (io_callback_t)e.obj->data;
        io_callback(ctx, event[i].obj, event[i].res, event[i].res2);
    }
}

eventfd + epoll

在学习epoll的过程中，有一个问题是普通文件的IO无法使用epoll来监视，因为ext4格式文件未实现 poll 方法，不过 linux kernel aio可以通过一些办法，来通过 epoll 监视io：

//创建eventfd
int evfd = eventfd(0, 0);
//将eventfd设置给iocb TODO：指针，考虑分配到堆内存
io_set_eventfd(io_cb_p[i], evfd);
//添加 eventfd 到epoll的监听中，只监视可读事件
epoll_event in_ev{EPOLLIN, {.fd=evfd}};
epoll_ctl(efd, EPOLL_CTL_ADD, evfd, &in_ev);
//epoll阻塞等待
int ret = epoll_wait(efd, evs, MAX_EVENT, -1);
//阻塞结束，获取结果，执行回调
io_event event[max_event];
int num = io_getevents(ctx, max_event, max_event, event, NULL);
printf("io_getevents num:%d\n", num);
//有异步队列读取完了
for (int i = 0; i < num; i++) {
    //调用回调对象中的回调函数（这里实际上只是打印了下信息）
    io_event &e = event[i];
    io_callback_t io_callback = (io_callback_t)e.obj->data;
    io_callback(ctx, event[i].obj, event[i].res, event[i].res2);
}

练习代码

练习代码仓库：https://github.com/zouchengzhuo/linux-io-learn/tree/master/4.aio/kernel_aio

不常写c++代码，练习的时候也踩了一些坑，记录一些编码易错点吧

calback绑定顺序错误

这个属于对 libaio 不了解导致的常见错误，习惯性的先绑定callback，再准备 aio 任务。结果发现执行callback的时候程序会core，将callback的指针打印出来，发现是空指针。

1 2	io_set_callback(io_cb_p[i], LibioCallback); io_prep_pread(io_cb_p[i], fd, buf, SEQ_BUF_SIZE, offset);

原因是，io_prep_pread的实现是

static inline void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)
{
    memset(iocb, 0, sizeof(*iocb));
    iocb->aio_fildes = fd;
    iocb->aio_lio_opcode = IO_CMD_PREAD;
    iocb->aio_reqprio = 0;
    iocb->u.c.buf = buf;
    iocb->u.c.nbytes = count;
    iocb->u.c.offset = offset;
}

上来就将 iocb 给清空了，前面设置的 callback 就没了，坑爹啊。。。

多重指针使用错误

这里是一个易错点，io_submit 的定义是：

1	int io_submit(io_context_t ctx, long nr, struct iocb *ios[]);

最后一个参数是一个多重指针，也就是一个指向 iocb 数组里每一个元素的指针的数组。

下面的写法，核心问题在于，通过 io_cb_ptr+i 可以得到指向 iocb 数组的下一个元素的指针地址，然而通过 &io_cb_ptr 拿到的地址是一个指向指针的指针，而不是一个指向指针数组的指针，所以 io_submit 只能拿到第一个 iocb，此时如果 io_getevents 传入最小事件为1，那么能拿到一个event后面的拿不到，如果传入的最小event数量大于1，则会永久阻塞。

/**
 * @brief 【错误2】：这种写法，由于错用了双重指针，导致 io_submit 提交成功的只有一个任务
 * @return std::string 
 */
std::string ReadByKernelAIOSubmitError1(){
    const int MAX_EVENT = 5;
    const int SEQ_BUF_SIZE = 1024;
    char f_path[] = "./test.txt";
    //很多文章说kernel aio只支持 O_DIRECT，这里测试 O_RDONLY 也是可以的，O_DIRECT 模式下 io_event.res 的值是-22
    //更新：只有正确初始化堆内存，而且内存对齐时 alignment 使用512，才不会报错
    int fd = open(f_path, O_DIRECT);
    ////step1. 初始化异步io的context
    io_context_t      ctx;
    memset(&ctx, 0, sizeof(ctx));
    if(io_setup(MAX_EVENT, &ctx)){
        printf("io_setup error\n");
        return "";
    }
    ////step2. 创建回调并提交异步任务
    //创建回调对象
    iocb io_cb[MAX_EVENT];
    iocb *io_cb_ptr = &io_cb[0];

    for (int i = 0; i < MAX_EVENT; i++) {
        int offset = i*SEQ_BUF_SIZE;
        void *buf;
        posix_memalign((void**)&buf, 512, SEQ_BUF_SIZE);
        io_prep_pread(io_cb_ptr+i, fd, buf, SEQ_BUF_SIZE, offset);
        io_set_callback(io_cb_ptr+i, LibioCallback);
    }
    //一次性提交提交多个任务，如果第三个指针参数设置错误，实际上只有第一个地址是指向 iocb 的，那么 io_getevents 总是只能拿到一个event，如果试图拿到更多，就永远阻塞了
    if (io_submit(ctx, MAX_EVENT, &io_cb_ptr) < 0) {
        printf("io_submit error\n");
        return "";
    }
    //通过 io_getevents 阻塞并等待异步任务完成
    CheckByIOGetevents(ctx, MAX_EVENT);
    io_destroy(ctx);
    close(fd);
    return "";
}

局部变量被释放的错误

在for循环中，通过 io_prep_pread 准备异步io任务时，是在局部作用域内通过声明 void *buf; 的方式在栈内存内获取了一个地址，在循环中，此局部变量会被认为不再使用，下一个循环中可能再次分配到这个地址，或者地址分配给别的程序。
此时栈内存如果没被其它数据覆盖，会表现为能拿到5个event，但是offset和读取的数据都是最后一个的;如果栈内存已经被覆盖，则程序可能core掉。

/**
 * @brief 【错误3】：callback放在局部变量里边，添加的循环结束后即被覆盖
 * @return std::string 
 */
std::string ReadByKernelAIOSubmitError2(){
    const int MAX_EVENT = 5;
    const int SEQ_BUF_SIZE = 1024;
    char f_path[] = "./test.txt";
    //很多文章说kernel aio只支持 O_DIRECT，这里测试 O_RDONLY 也是可以的，O_DIRECT 模式下 io_event.res 的值是-22，只有正确初始化堆内存，而且内存对齐时 alignment 使用512，才不会报错
    int fd = open(f_path, O_DIRECT);
    ////step1. 初始化异步io的context
    io_context_t      ctx;
    memset(&ctx, 0, sizeof(ctx));
    if(io_setup(MAX_EVENT, &ctx)){
        printf("io_setup error\n");
        return "";
    }
    ////step2. 创建回调并提交异步任务
    for (int i = 0; i < MAX_EVENT; i++) {
        //创建回调对象
        iocb io_cb;
        iocb *io_cb_ptr = &io_cb;
        int offset = i*SEQ_BUF_SIZE;
        void *buf;
        posix_memalign((void**)&buf, 512, SEQ_BUF_SIZE);
        io_prep_pread(io_cb_ptr, fd, buf, SEQ_BUF_SIZE, offset);
        io_set_callback(io_cb_ptr, LibioCallback);
        if(io_submit(ctx, 1, &io_cb_ptr) < 0) {
            printf("io_submit error\n");
            return "";
        }
    }
    //通过 io_getevents 阻塞并等待异步任务完成
    CheckByIOGetevents(ctx, MAX_EVENT);
    io_destroy(ctx);
    close(fd);
    return "";
}

堆/栈内存分配方式错误

分配内存并对齐的过程中，有一些注意点：

需要保证内存分配在堆内存中，而不是局部的栈内存，不然有可能被覆盖
.O_DIRECT需要调用 posix_memalign 来进行分配，且与官方的描述不同，只有在特定的 alignment （可通过 blockdev –getss /dev/vdb1 查看，centos8下，512字节可以，256不行）下才不报-22（即-EINVAL）
io_prep_pread的时候，传入的读取buf size 也必须是512/1024这样的2的n次方，否则 nbytes 正确，但是 res 会报-22，且拿不到buf结果
先在栈空间分配一块内存，如 char buf[SEQ_BUF_SIZE];，再 posix_memalign((void**)&buf, 512, SEQ_BUF_SIZE) 进行对齐，实际上不会执行对齐操作，需要先声明指针，如 void *buf; 或者 char *buf; 再将声明的指针作为参数对齐内存

/**
 * @brief 【错误4】： 内存分配方式错误，这种方式 offset 没问题，但是内容都是最后一段的
 * @return std::string 
 */
std::string ReadByKernelAIOAllocError(){
    const int MAX_EVENT = 5;
    const int SEQ_BUF_SIZE = 1024;
    char f_path[] = "./test.txt";
    int fd = open(f_path, O_RDONLY);
    ////step1. 初始化异步io的context
    io_context_t      ctx;
    memset(&ctx, 0, sizeof(ctx));
    if(io_setup(MAX_EVENT, &ctx)){
        printf("io_setup error\n");
        return "";
    }
    ////step2. 创建回调并提交异步任务
    //创建回调对象
    iocb io_cb[MAX_EVENT];
    iocb *io_cb_p[MAX_EVENT];

    //注意点1：这种初始化方式，每个callback 都往这块栈内存里边写，会覆盖内容，且执行内存对齐时不会真的去对齐
    char buf[SEQ_BUF_SIZE];

    for (int i = 0; i < MAX_EVENT; i++) {
        io_cb_p[i] = &io_cb[i];
        int offset = i*SEQ_BUF_SIZE;
        //注意点2：写到里边来每次分配新的栈内存，也不行，局部变量作用域结束后仍然会被覆盖
        //char buf[SEQ_BUF_SIZE];
        //注意点3：先在栈空间分配，再调用posix_memalign做对齐内存分配，是无效的操作，并不会去对齐分配，需要声明  void *buf; 或者 char *buf; 再对齐内存
        //posix_memalign((void**)&buf, 512, SEQ_BUF_SIZE);
        io_prep_pread(io_cb_p[i], fd, buf, SEQ_BUF_SIZE, offset);
        io_set_callback(io_cb_p[i], LibioCallback);
    }
    //一次性提交提交多个任务，如果第三个指针参数设置错误，实际上只有第一个地址是指向 iocb 的，那么 io_getevents 总是只能拿到一个event，如果试图拿到更多，就永远阻塞了
    if (io_submit(ctx, MAX_EVENT, io_cb_p) < 0) {
        printf("io_submit error");
        return "";
    }
    //通过 io_getevents 阻塞并等待异步任务完成
    CheckByIOGetevents(ctx, MAX_EVENT);
    io_destroy(ctx);
    close(fd);
    return "";
}

参考文章：

本文链接：https://www.zoucz.com/blog/2022/06/01/210c97f0-e166-11ec-9fe7-534bbf9f369d/

☞ 参与评论

linux IO —— 异步IO之 kernel aio