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

Posted 2022-06-17 -飞鹤-

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;