Linux开发之libaio源码分析及应用

1. 简介

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

2. 源码

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

2.1. libaio的目录结构

2.2. 源码分析

2.2.1. 目录结构

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  
     #include  
     #include  
     #include  
     #include  
     #include  
     #include  
     #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 = ; 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; }