Linux 异步 I/O

最近看了一篇不错的文章《Linux Asynchronous I/O》,特别翻译分享出来。其中,我自己补充的内容会用引用标记出来。

介绍

异步 I/O (Asynchronous I/O, AIO) 是一种提升 I/O 操作性能的方法。AIO可以允许进程发起很多 I/O 操作,而不用阻塞或等待任何操作完成。稍后或在接收到 I/O 操作完成的通知时,进程就可以检查 I/O 操作的结果。

在 Linux 系统上有三种方式来实现 AIO:

  • 内核系统调用
  • 对内核系统调用进行封装进而在用户空间提供服务,例如 libaio
  • 完全在用户空间实现 AIO,并且不使用内核支持,例如librt和部分libc

IO模型

关于 IO 模型详见《UNIX 网络编程 卷1》的6.2节 “IO 模型”,一共介绍了 Unix 中的5种 IO 模型:

  • 阻塞式 I/O
  • 非阻塞式 I/O
  • IO 复用 (select/poll)
  • 信号驱动式 I/O (SIGIO)
  • 异步 I/O (aio)
模型 阻塞 非阻塞
同步 read/write read/write
(O_NONBLOCK)
异步 I/O multiplexing
(select/poll/epoll)
AIO

AIO 系统调用

接口

AIO 系统系统调用入口点位于内核源码文件的fs/aio.c中。导出到用户空间的类型和常量都声明在/usr/include/linux/aio_abi.h这个头文件中。

Linux 内核提供了5个系统调用来实现 AIO:

1
2
3
4
5
6
7
#include <linux/aio_abi.h>
int io_setup(unsigned nr_events, aio_context_t *ctxp);
int io_destroy(aio_context_t ctx);
int io_submit(aio_context_t ctx, long nr, struct iocb **iocbpp);
int io_cancel(aio_context_t ctx, struct iocb *, struct io_event *result);
int io_getevents(aio_context_t ctx, long min_nr, long nr,
struct io_event *events, struct timespec *timeout);

每个提交到 AIO 上下文的 I/O 请求都由一个 I/O 控制块结构struct iocb来表示。

在传统的 I/O 模型中,有一个使用唯一句柄标识的 I/O 通道。在 UNIX 中,这些句柄是文件描述符(等同于文件、管道、套接字等等)。在阻塞 I/O 中,我们发起了一次传输操作,当传输操作完成或发生错误时,系统调用就会返回。

在异步非阻塞 I/O 中,我们可以同时发起多个传输操作。这需要每个传输操作都有唯一的上下文,这样我们才能在它们完成时区分到底是哪个传输操作完成了。在 AIO 中,这是一个 aiocb(AIO I/O Control Block)结构。这个结构包含了有关传输的所有信息,包括为数据准备的用户缓冲区。在产生 I/O (称为完成)通知时,aiocb 结构就被用来惟一标识所完成的 I/O 操作。

io_submit()参数如下。注意,iocb 数据的成员是指向 iocb 的指针。

参数 含义
aio_context_t ctx AIO 上下文 ID
long nr iocb 数组长度
struct iocb **iocbpp iocb 数据缓冲区

io_submit()返回值有一下几种:

  1. $ret = nr$。是最好的情况,所有的 iocbs 都成功处理;
  2. $0 < ret < nr$。io_submit()系统调用从数组起始位置逐个处理所有的 iocb。如果提交的某个 iocb 失败了,该函数会停止处理后续 iocb,并返回失败 iocb 的索引值。io_submit函数不返回失败的具体原因。如果第一个 iocb 就失败,见后文;
  3. $ret < 0​$。有两种情况会导致返回值小于0:
    • 错误发生在迭代处理 iocb 之前。例如,AIO 上下文无效;
    • 第一个 iocb 提交失败。

在 iocb 提交成功后,进程可以继续做其他事情而不用等待 I/O 完成。对于每个完成的 I/O 请求(无论成功与否),内核都会创建一个 io_event结构。io_getevents()函数的功能就是获取 io_event 列表(和所有已完成的iocbs)。函数参数如下所示。

参数 含义
aio_context_t ctx AIO 上下文 ID
long min_nr 获取事件的最小值
long nr 获取事件的最大值
struct io_event *events 一个 io_event 的缓存区
struct timespec *timeout 超时时间。

特别的,如果已完成 iocb 数量没有达到期望最小值,则函数会阻塞;而指定超时时间,即使没有获取足够的事件,则等待一定时间后返回,不会一直阻塞。超时时间设置为NULL意味着一直等待。如果想要函数完全不等待、立马返回,则要将timespec结构中的秒和纳秒结构都初始化为0。

io_getevents函数返回值如下:

  1. $ret = nr​$。从内核获取到了所有事件,这有可能意味着内核中还有更多挂起的事件;
  2. $min\_nr < ret < nr$。目前内核中所有的可用事件都被读取且没有发生阻塞;
  3. $0 < ret < min\_nr$。目前内核中所有的可用事件都被读取,同时阻塞用户声明的超时时间;
  4. $ret = 0$。目前内核中没有事件可以返回;
  5. $ret < 0​$。发生错误。

关键数据结构

io_event结构如下:

1
2
3
4
5
6
7
/* 该结构由 read() 函数从 /dev/aio 读取数据后返回. */
struct io_event {
__u64 data; /* 数据段,对应 iocb 的 aio_data 的值 */
__u64 obj; /* 事件对应的 iocb,指向对应 iocb 的指针 */
__s64 res; /* 对应 IO 请求的结果*/
__s64 res2; /* 对应 IO 请求的结果 */
};

iocb结构如下:

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
/*
* 和用户空间通信的 off_t 是64bit,因此 iocb 的结构
* 具体取决于库的实现来做大小的填充和错误的抽象。
*/

struct iocb {
/* these are internal to the kernel/libc. */
__u64 aio_data; /* data to be returned in event's data */
__u32 PADDED(aio_key, aio_reserved1);
/* the kernel sets aio_key to the req # */

/* 通用字段 */
__u16 aio_lio_opcode; /* 请求类型,见后文 AIO 请求 */
__s16 aio_reqprio;
__u32 aio_fildes;

__u64 aio_buf;
__u64 aio_nbytes;
__s64 aio_offset;

/* 额外参数 */
__u64 aio_reserved2; /* TODO: use this for a (struct sigevent *) */

/* 可选IOCB_FLAG_RESFD标记,表示异步请求处理完成时使用eventfd进行通知 */
__u32 aio_flags;

/* 如果 aio_flags 被设置为 IOCB_FLAG_RESFD ,用作接收通知的 eventfd */
__u32 aio_resfd;
}; /* 64 bytes */

AIO 请求

上文aio_lio_opcode的值如下所示。结构中其他字段的语义取决于 AIO 请求的类型。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# /usr/include/linux/aio_abi.h
enum {
IOCB_CMD_PREAD = 0,
IOCB_CMD_PWRITE = 1,
IOCB_CMD_FSYNC = 2,
IOCB_CMD_FDSYNC = 3,
/* These two are experimental.
* IOCB_CMD_PREADX = 4,
* IOCB_CMD_POLL = 5,
*/
IOCB_CMD_NOOP = 6,
IOCB_CMD_PREADV = 7,
IOCB_CMD_PWRITEV = 8,
};
  • IOCB_CMD_PREAD。对应pread()系统调用;

  • IOCB_CMD_PWRITE。对应 pwrite() 系统调用;

  • IOCB_CMD_FSYNC。同步文件数据和元数据到稳定存储介质,对应fsync()系统调用;

fdatasync的功能与fsync类似,但是仅仅在必要的情况下才会同步metadata,因此可以减少一次IO写操作。那么,什么是“必要的情况”呢?根据man page中的解释:

fdatasync does not flush modified metadata unless that metadata is needed in order to allow a subsequent data retrieval to be corretly handled.

举例来说,文件的尺寸(st_size)如果变化,是需要立即同步的,否则OS一旦崩溃,即使文件的数据部分已同步,由于metadata没有同步,依然读不到修改的内容。而最后访问时间(atime)/修改时间(mtime)只要程序对两个参数没有精准要求,元数据是不需要每次都同步的。

  • IOCB_CMD_FDSYNC。同步文件数据和元数据到稳定存储介质,但是元数据只有在必要的时候才会同步,对应fdatasync()系统调用;

  • IOCB_CMD_PREADV。对应preadv()系统调用;

  • IOCB_CMD_PWRITEV。对应pwritev() 系统调用

  • IOCB_CMD_NOOP。未使用。

AIO 上下文

AIO 的上下文是内核提供的一组实现 AIO 的数据结构。

每个进程可以拥有多个 AIO 上下文,因此每个 AIO 上下文需要一个标识符来在一个进程中识别自己。

io_setup() 的第二个参数是 AIO 上下文ctx指针,由内核给这个上下文分配一个 ID。有趣的是,aio_context_t的类型在内核中实际上是一个unsigned long(位于 linux/aio_abi.h):

1
typedef unsigned long aio_context_t;

io_setup()第一个参数是该上下文可以获取事件的最大值。

syscall()

1
2
3
4
5
#define _GNU_SOURCE         /* See feature_test_macros(7) */
#include <unistd.h>
#include <sys/syscall.h> /* For SYS_xxx definitions */

int syscall(int number, ...);

syscall() 执行一个系统调用,根据指定的参数number和所有系统调用的汇编语言接口来确定调用哪个系统调用。这个函数非常有用,当我们想使用一个没有 封装的系统调用时,就可以使用它。系统调用所使用的符号常量可以在头文件<sys/syscll.h>里面找到。

syscall()在进行系统调用之前保存 CPU 寄存器内容,并且在调用完成之后恢复寄存器内容。当系统调用发生错误时,可以从errno中读取到错误码。

样例程序

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#include <stdio.h>
#include <string.h>
#include <inttypes.h>

#include <unistd.h>
#include <fcntl.h>
#include <sys/syscall.h>
#include <linux/aio_abi.h>

inline int io_setup(unsigned nr, aio_context_t *ctxp) {
return syscall(__NR_io_setup, nr, ctxp);
}

inline int io_destroy(aio_context_t ctx) {
return syscall(__NR_io_destroy, ctx);
}

inline int io_submit(aio_context_t ctx, long nr, struct iocb **iocbpp) {
return syscall(__NR_io_submit, ctx, nr, iocbpp);
}

inline int io_getevents(aio_context_t ctx, long min_nr, long max_nr,
struct io_event *events, struct timespec *timeout) {
return syscall(__NR_io_getevents, ctx, min_nr, max_nr, events, timeout);
}

int main(int argc, char *argv[]) {
aio_context_t ctx;
struct iocb cb;
struct iocb *cbs[1];
char data[4096];
struct io_event events[1];
int ret;
int fd;

fd = open("/tmp/test", O_RDWR | O_CREAT);
if (fd < 0) {
perror("open");
return -1;
}

ctx = 0;

ret = io_setup(128, &ctx);
if (ret < 0) {
perror("io_setup");
return -1;
}

/* 初始化I/O 控制块 */
memset(&cb, 0, sizeof(cb));
cb.aio_fildes = fd;
cb.aio_lio_opcode = IOCB_CMD_PWRITE;

/* 指定 AIO 请求 */
int i;
for (i = 0; i < 4096; ++i)
data[i] = 'A';
cb.aio_buf = (uint64_t)data;
cb.aio_offset = 0;
cb.aio_nbytes = 4096;

cbs[0] = &cb;

ret = io_submit(ctx, 1, cbs);
if (ret != 1) {
if (ret < 0) perror("io_submit");
else fprintf(stderr, "io_submit failed\n");
return -1;
}

/* 获取返回值 */
ret = io_getevents(ctx, 1, 1, events, NULL);
printf("events: %d\n", ret);
ret = io_destroy(ctx);
if (ret < 0) {
perror("io_destroy");
return -1;
}
return 0;
}

参数调节

/proc/sys/fs/aio-max-nr:系统可分配的 AIO 上下文指定请求的最大值;

/proc/sys/fs/aio-nr:系统当前由io_setup系统调用创建的活跃 AIO 上下文指定的事件总数。

aio-nr值大于aio-max-nr时,io_setup会返回错误EAGAIN

libaio

安装 libaio

1
2
3
4
5
6
7
8
9
10
11
[oxnz@localhost aio]$ sudo yum install libaio-devel
[oxnz@localhost aio]$ rpm -ql libaio
/lib64/libaio.so.1
/lib64/libaio.so.1.0.0
/lib64/libaio.so.1.0.1
/usr/share/doc/libaio-0.3.109
/usr/share/doc/libaio-0.3.109/COPYING
/usr/share/doc/libaio-0.3.109/TODO
[oxnz@localhost aio]$ rpm -ql libaio-devel
/usr/include/libaio.h
/usr/lib64/libaio.so

系统调用封装

1
2
3
4
5
6
7
/* /usr/include/libaio.h */
/* Actual syscalls */
extern int io_setup(int maxevents, io_context_t *ctxp);
extern int io_destroy(io_context_t ctx);
extern int io_submit(io_context_t ctx, long nr, struct iocb *ios[]);
extern int io_cancel(io_context_t ctx, struct iocb *iocb, struct io_event *evt);
extern int io_getevents(io_context_t ctx_id, long min_nr, long nr, struct io_event *events, struct timespec *timeout);

帮助函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
static inline void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)
static inline void io_prep_pwrite(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)
static inline void io_prep_preadv(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset)
static inline void io_prep_pwritev(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset)

static inline void io_prep_poll(struct iocb *iocb, int fd, int events)
static inline void io_prep_fsync(struct iocb *iocb, int fd)
static inline void io_prep_fdsync(struct iocb *iocb, int fd)

static inline int io_poll(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd, int events)
static inline int io_fsync(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd)
static inline int io_fdsync(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd)

static inline void io_set_eventfd(struct iocb *iocb, int eventfd);

struct iocb

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
46
struct io_iocb_poll {
PADDED(int events, __pad1);
}; /* result code is the set of result flags or -'ve errno */

struct io_iocb_sockaddr {
struct sockaddr *addr;
int len;
}; /* result code is the length of the sockaddr, or -'ve errno */

struct io_iocb_common {
PADDEDptr(void *buf, __pad1);
PADDEDul(nbytes, __pad2);
long long offset;
long long __pad3;
unsigned flags;
unsigned resfd;
}; /* result code is the amount read or -'ve errno */

struct io_iocb_vector {
const struct iovec *vec;
int nr;
long long offset;
}; /* result code is the amount read or -'ve errno */

struct iocb {
PADDEDptr(void *data, __pad1); /* Return in the io completion event */
PADDED(unsigned key, __pad2); /* For use in identifying io requests */

short aio_lio_opcode;
short aio_reqprio;
int aio_fildes;

union {
struct io_iocb_common c;
struct io_iocb_vector v;
struct io_iocb_poll poll;
struct io_iocb_sockaddr saddr;
} u;
};

struct io_event {
PADDEDptr(void *data, __pad1);
PADDEDptr(struct iocb *obj, __pad2);
PADDEDul(res, __pad3);
PADDEDul(res2, __pad4);
};

样例程序

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
46
47
48
49
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <err.h>
#include <errno.h>

#include <unistd.h>
#include <fcntl.h>
#include <libaio.h>

int main() {
io_context_t ctx;
struct iocb iocb;
struct iocb * iocbs[1];
struct io_event events[1];
struct timespec timeout;
int fd;

fd = open("/tmp/test", O_WRONLY | O_CREAT);
if (fd < 0) err(1, "open");

memset(&ctx, 0, sizeof(ctx));
if (io_setup(10, &ctx) != 0) err(1, "io_setup");

const char *msg = "hello";
io_prep_pwrite(&iocb, fd, (void *)msg, strlen(msg), 0);
iocb.data = (void *)msg;

iocbs[0] = &iocb;

if (io_submit(ctx, 1, iocbs) != 1) {
io_destroy(ctx);
err(1, "io_submit");
}

while (1) {
timeout.tv_sec = 0;
timeout.tv_nsec = 500000000;
if (io_getevents(ctx, 0, 1, events, &timeout) == 1) {
close(fd);
break;
}
printf("not done yet\n");
sleep(1);
}
io_destroy(ctx);

return 0;
}

编译:

1
cc libaio.c -o libaio -laio

POSIX AIO

1
2
/lib64/librt.so
/usr/include/aio.h

接口

POSIX AIO 接口有以下函数:

  • aio_read(3):入队列一个读请求。为read(2)的异步函数;
  • aio_write(3):入队列一个写请求。为write(2)的异步函数;
  • aio_fsync(3):入队列一个同步文件描述符 I/O 操作的请求。为fsync(2)fdatasync(2)的异步函数;
  • aio_error(3):获取一个已入队的 I/O 请求的错误状态;
  • aio_return(3):获取一个已完成的 I/O 请求的返回状态;
  • aio_suspend(3):将调用进程挂起,直到有 I/O 请求完成;
  • aio_cancel(3):用于取消对某个文件描述符执行的一个或所有 I/O 请求;
  • lio_listio(3):同时入队列多个 I/O 请求;

当前的Linux POSIX AIO由glibc在用户空间中实现。POSIX AIO 有许多限制,最值得注意的是,维护多个线程来执行I/O操作是昂贵的,并且可扩展性很差。在基于内核状态机的 AIO 实现上已经进行了一段时间的工作,但是这个实现还没有成熟到可以使用内核系统调用完全重新实现POSIX AIO实现的程度。

参考

  1. 使用异步 I/O 大大提高应用程序的性能

  2. Linux kernel AIO这个奇葩