接上篇 linux IO —— 同步IO

为什么要讨论异步IO

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

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

image.png

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

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

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

DIRECT IO模式和非DIRECT IO模式

非DIRECT IO模式

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

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

image.png

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

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

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是有价值的,如下图:

image.png

相比于同步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开发环境需要的头文件

一个基本的使用流程是:

1
2
3
4
5
6
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完成时唤醒进程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* @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:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
//创建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的实现是

1
2
3
4
5
6
7
8
9
10
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,则会永久阻塞。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
/**
* @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掉。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
/**
* @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; 再将声明的指针作为参数对齐内存
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
/**
* @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 "";
}

参考文章:

☞ 参与评论