1. 简介

Linux的POSIX API由glibc提供，2000年年之前，glibc一直没有提供异步I/O的调用API。Red Hat公司基于Linux内核的符号表封装了一套异步I/O(简称aio)的接口，并提供了一些新的接口用来简化上下文配置，开成一个库，命名为libaio。
glibc后来有实现一套异步I/O，其实现是用多线程来封装同步的I/O以达到异步的效果，但是因为存在线程的切换，性能并不好。
Linux Kernel 2.6引入了libaio的几个接口定义，但glibc并没有提供相应的API接口，如果要使用libaio的接口，依然需要安装libaio库。
libaio有非常大的缺陷，例如只支持Direct I/O。I/O需求最大的网络I/O，目前的epoll已经够用，各方对异步I/O的的使用没有足够的动力。异步I/O的应用场景不多，主要用在一些性能基准测试上，对此libaio已经够用。
当下，如果想在Linux下使用异步I/O，有三种方式：

1. 自行封装Linux kerner的符号表，完成系统调用。
2. 使用libaio完成异步I/O的调用。
3. 使用glibc的glibc-aio来完成调用。

2. 源码

libaio目前的源码托管在https://pagure.io/libaio，近些年来很少修改，只是做一些兼容性维护。

2.1. libaio的目录结构

├── ChangeLog
├── COPYING
├── harness #dir
├── INSTALL
├── libaio.spec
├── Makefile
├── man #dir
├── README.md
├── src # dir
└── TODO
● harness目录存放测试用例集，并提供runtest.sh脚本自动测试。
● man目录存放接口使用手册。
● src存放源代码。

2.2. 源码分析

2.2.1. 目录结构

├── aio_ring.h
├── compat-0_1.c
├── io_cancel.c
├── io_destroy.c
├── io_getevents.c
├── io_pgetevents.c
├── io_queue_init.c
├── io_queue_release.c
├── io_queue_run.c
├── io_queue_wait.c
├── io_setup.c
├── io_submit.c
├── libaio.h
├── libaio.map
├── Makefile
├── raw_syscall.c
├── struct_offsets.c
├── syscall-alpha.h
├── syscall-arm.h
├── syscall-generic.h
├── syscall.h
├── syscall-i386.h
├── syscall-ia64.h
├── syscall-ppc.h
├── syscall-s390.h
├── syscall-sparc.h
├── syscall-x86_64.h
└── vsys_def.h

2.2.2. 解析

2.2.2.1. aio_ring.h

主要提供aio_ring_is_empty内部函数用于检测接口上下文指针参数是否为空。

static inline int aio_ring_is_empty(io_context_t ctx, struct timespec *timeout)
{struct aio_ring *ring = (struct aio_ring *)ctx;if (!ring || ring->magic != AIO_RING_MAGIC)return 0;if (!timeout || timeout->tv_sec || timeout->tv_nsec)return 0;if (ring->head != ring->tail)return 0;return 1;
}

2.2.2.2. compat-0_1.c

此代码主要是调用gcc的汇编指令.symver来完成接口兼容不同版本。

#define _SYMSTR(str)	#str
#define SYMSTR(str)	_SYMSTR(str)#define SYMVER(compat_sym, orig_sym, ver_sym)	\__asm__(".symver " SYMSTR(compat_sym) "," SYMSTR(orig_sym) "@LIBAIO_" SYMSTR(ver_sym));SYMVER(compat0_1_io_cancel, io_cancel, 0.1);
int compat0_1_io_cancel(io_context_t ctx, struct iocb *iocb)
{struct io_event event;/* FIXME: the old ABI would return the event on the completion queue */return io_cancel(ctx, iocb, &event);
}

2.2.2.3. io_cancel.c

DEFSYMVER(io_cancel_0_4, io_cancel, 0.4)是为了保证io_cancel_0_4和io_cancel的兼容，上面的代码展开之后，调用io_cance的调用实际会转为调用syscall(__NR_io_cancel, arg)的内核调用。因为内核只对外提供了函数调用的入口索引，也即__NR_io_cancel，并没有提供符号表，所以不能直接通过函数符号来调用内核函数。

#define DEFSYMVER(compat_sym, orig_sym, ver_sym)	\__asm__(".symver " SYMSTR(compat_sym) "," SYMSTR(orig_sym) "@@LIBAIO_" SYMSTR(ver_sym));#define _body_io_syscall(sname, args...)	\
{						\int ret, saved_errno;			\saved_errno = errno;			\ret= syscall(__NR_##sname, ## args);	\if (ret < 0) {				\ret = -errno;			\errno = saved_errno;		\}					\return ret;				\
}#define io_syscall1(type,fname,sname,type1,arg1) \
type fname(type1 arg1) \
_body_io_syscall(sname, (long)arg1)#define io_syscall2(type,fname,sname,type1,arg1,type2,arg2) \
type fname(type1 arg1,type2 arg2) \
_body_io_syscall(sname, (long)arg1, (long)arg2)#define io_syscall3(type,fname,sname,type1,arg1,type2,arg2,type3,arg3) \
type fname(type1 arg1,type2 arg2,type3 arg3) \
_body_io_syscall(sname, (long)arg1, (long)arg2, (long)arg3)io_syscall3(int, io_cancel_0_4, io_cancel, io_context_t, ctx, struct iocb *, iocb, struct io_event *, event)
DEFSYMVER(io_cancel_0_4, io_cancel, 0.4)

2.2.2.4. io_destory.c

io_syscall1(int, io_destroy, io_destroy, io_context_t, ctx)

2.2.2.5. io_getevents.c

io_syscall5(int, __io_getevents_0_4, io_getevents, io_context_t, ctx, long, min_nr, long, nr, struct io_event *, events, struct timespec *, timeout)int io_getevents_0_4(io_context_t ctx, long min_nr, long nr, struct io_event * events, struct timespec * timeout)
{if (aio_ring_is_empty(ctx, timeout))return 0;return __io_getevents_0_4(ctx, min_nr, nr, events, timeout);
}DEFSYMVER(io_getevents_0_4, io_getevents, 0.4)

2.2.2.6. io_pgetevents.c

如果内核有实现__NR_io_pgetevents，则调用内核的，如果没有则调用一个自定义的空函数。

#ifdef __NR_io_pgetevents
io_syscall6(int, __io_pgetevents, io_pgetevents, io_context_t, ctx, long,min_nr, long, nr, struct io_event *, events,struct timespec *, timeout, void *, sigmask);int io_pgetevents(io_context_t ctx, long min_nr, long nr,struct io_event *events, struct timespec *timeout,sigset_t *sigmask)
{struct {unsigned long ss;unsigned long ss_len;} data;if (aio_ring_is_empty(ctx, timeout))return 0;data.ss = (unsigned long)sigmask;data.ss_len = _NSIG / 8;return __io_pgetevents(ctx, min_nr, nr, events, timeout, &data);
}
#else
int io_pgetevents(io_context_t ctx, long min_nr, long nr,struct io_event *events, struct timespec *timeout,sigset_t *sigmask){return -ENOSYS;
}
#endif /* __NR_io_pgetevents */

2.2.2.7. io_queue_inti.c

提供对io_setup接口的再封装接口io_queue_init。

int io_queue_init(int maxevents, io_context_t *ctxp)
{if (maxevents > 0) {*ctxp = NULL;return io_setup(maxevents, ctxp);}return -EINVAL;
}

2.2.2.8. io_rueue_release.c

int io_queue_release(io_context_t ctx)
{return io_destroy(ctx);
}

2.2.2.9. io_queue_run.c

int io_queue_run(io_context_t ctx)
{static struct timespec timeout = { 0, 0 };struct io_event event;int ret;/* FIXME: batch requests? */while (1 == (ret = io_getevents(ctx, 0, 1, &event, &timeout))) {io_callback_t cb = (io_callback_t)event.data;struct iocb *iocb = event.obj;cb(ctx, iocb, event.res, event.res2);}return ret;
}

2.2.2.10. io_queue_wait.c

int io_queue_wait_0_4(io_context_t ctx, struct timespec *timeout)
{return io_getevents(ctx, 0, 0, NULL, timeout);
}
DEFSYMVER(io_queue_wait_0_4, io_queue_wait, 0.4)

2.2.2.11. io_setup.c

io_syscall2(int, io_setup, io_setup, int, maxevents, io_context_t *, ctxp)

2.2.2.12. io_submit.c

io_syscall3(int, io_submit, io_submit, io_context_t, ctx, long, nr, struct iocb **, iocbs)

2.2.2.13. libaio.h

除了提供上述几个对内核的调用接口声明外，还有提供几个静态函数，用来简化对上下文的赋值。

static inline void io_set_callback(struct iocb *iocb, io_callback_t cb)
{iocb->data = (void *)cb;
}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;
}static inline void io_prep_pwrite(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_PWRITE;iocb->aio_reqprio = 0;iocb->u.c.buf = buf;iocb->u.c.nbytes = count;iocb->u.c.offset = offset;
}

2.2.2.14. 其他文件

剩余的其他头文件主要包括描述不同架构的系统符号编号宏文件。

2.3. 编译

直接在当前目录执行make命令即可。

2.4. 安装

在当前目录执行命令：

make prefix=pwd/usr install

3. 应用

linux下默认是没有安装libaio的，可以直接通过命令安装。

sudo apt install libaio

3.1. 基本流程

1. 当应用程序调用 io_submit 系统调用发起一个异步 IO 操作后，会向内核的 IO 任务队列中添加一个 IO 任务，并且返回成功。
2. 内核会在后台处理 IO 任务队列中的 IO 任务，然后把处理结果存储在 IO 任务中。
3. 应用程序可以调用 io_getevents 系统调用来获取异步 IO 的处理结果，如果 IO 操作还没完成，那么返回失败信息，否则会返回 IO 处理结果。

从上面的流程可以看出，Linux 的异步 IO 操作主要由两个步骤组成：

1. 调用 io_submit 函数发起一个异步 IO 操作。
2. 调用 io_getevents 函数获取异步 IO 的结果。

3.2. 代码流程

1. 通过调用 open 系统调用打开要进行异步 IO 的文件，要注意的是 AIO 操作必须设置 O_DIRECT 直接 IO 标志位。
2. 调用 io_setup 系统调用创建一个异步 IO 上下文。
3. 调用 io_prep_pwrite 或者 io_prep_pread 函数创建一个异步写或者异步读任务。
4. 调用 io_submit 系统调用把异步 IO 任务提交到内核。
5. 调用 io_getevents 系统调用获取异步 IO 的结果。
   代码：

#define _GNU_SOURCE#include <stdlib.h>
#include <string.h>
#include <libaio.h>
#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>#define FILEPATH "./aio.txt"
static void wr_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
{if (res2 != 0){printf("aio write error\n");}if (res != iocb->u.c.nbytes){printf("write missed bytes expect %d got %d\n", iocb->u.c.nbytes, res);exit(1);}
}int main()
{io_context_t context;struct iocb io[1], *p[1] = {&io[0]};struct io_event e[1];unsigned nr_events = 10;struct timespec timeout;char *wbuf;int wbuflen = 1024;int ret, num = 0, i;posix_memalign((void **)&wbuf, 512, wbuflen);memset(wbuf, '@', wbuflen);memset(&context, 0, sizeof(io_context_t));timeout.tv_sec = 0;timeout.tv_nsec = 10000000;int fd = open(FILEPATH, O_CREAT|O_RDWR|O_DIRECT, 0644); // 1. 打开要进行异步IO的文件if (fd < 0) {printf("open error: %d\n", errno);return 0;}if (0 != io_setup(nr_events, &context)) {               // 2. 创建一个异步IO上下文printf("io_setup error: %d\n", errno);return 0;}io_prep_pwrite(&io[0], fd, wbuf, wbuflen, 0);           // 3. 创建一个异步IO任务io_set_callback(iocb, wr_done);if ((ret = io_submit(context, 1, p)) != 1) {            // 4. 提交异步IO任务printf("io_submit error: %d\n", ret);io_destroy(context);return -1;}while (1) {struct io_event events[64] = {};ret = io_getevents(context, 1, 1, events, &timeout);     // 5. 获取异步IO的结果if (ret < 0) {printf("io_getevents error: %d\n", ret);break;}if (ret > 0) {printf("result, res2: %d, res: %d\n", events[0].res2, events[0].res);break;}io_callback_t cb = (io_callback_t)events[i].data;struct iocb *io = events[i].obj;cb(context, io, events[i].res, events[i].res2);}return 0;
}