POSIX 异步 I/O

    POSIX 异步 I/O 接口为对不同类型的文件进行异步 I/O 提供了一套一致的方法。这些接口使用 AIO 控制块来描述 I/O 操作。aiocb 结构定义了 AIO 控制块，该结构至少包括下面这些字段：

#include <aio.h>
struct aiocb{
    int             aio_fildes;      // file descriptor
    off_t           aio_offset;      // file offset for I/O
    volatile  void *aio_buf;         // buffer for I/O
    size_t          aio_nbytes;      // number of bytes to transfer
    int             aio_reqprio;     // priority
    struct sigevent aio_sigevent;    // signal information
    int             aio_lio_opcode;  // operation for list I/O
};
struct sigevent{
    int             sigev_notify;    // notify type
    int             sigev_signo;     // signal number
    union sigval    sigev_value;     // notify argument
    void (*sigev_notify_function)(union sigval);    // notify function
    pthread_attr_t *sigev_notify_attributes;        // notify attrs
};

    在 aiocb 结构中，aio_fildes 字段表示被打开用来读或写的文件描述符。读或写操作从 aio_offset 指定的偏移量开始（注意，异步 I/O 操作必须显示地指定偏移量。只要不在同一个进程里把异步 I/O 函数和传统 I/O 函数混在一起用在同一个文件上，异步 I/O 接口是不会影响操作系统维护的文件偏移量的。另外，如果使用异步 I/O 接口向一个以追加模式打开的文件中写入数据，aio_offset 字段会被忽略）。读写数据的操作都是在 aio_buf 指定的缓冲区中进行，aio_nbytes 字段则包含了要读写的字节数。aio_reqprio 为异步 I/O 请求提示顺序（但系统对该顺序的控制力有限，因此不一定遵循）。aio_lio_opcode 字段只能用于基于列表的异步 I/O（见下）。aio_sigevent 使用 sigevent 结构来控制在 I/O 事件完成后，如何通知应用程序。
    在 sigevent 结构中，sigev_notify 字段控制通知的类型，其取值可能是以下 3 个中的一个。
    （1）SIGEV_NONE：异步 I/O 请求完成后，不通知进程。
    （2）SIGEV_SIGNAL：异步 I/O 请求完成后，产生由 sigev_signo 字段指定的信号。如果应用程序已选择捕捉信号，且在建立信号处理程序时指定了 SA_SIGINFO 标志，那么该信号将被入队（如果实现支持排队信号）。信号处理程序会传送给一个 siginfo 结构，该结构的 si_value 字段被设置为 sigev_value（如果使用了 SA_SIGINFO 标志）。
    （3）SIGEV_THREAD：当异步 I/O 请求完成时，由 sigev_notify_function 指定的函数会被调用，sigev_value 字段被作为它的唯一参数传入。除非 sigev_notify_attributes 字段被设置为 pthread 属性结构的地址，且该结构指定了另一个线程属性，否则该函数将在分离状态下的一个单独的线程中执行。
    进行异步 I/O 之前需要先初始化 AIO 控制块。aio_read 和 aio_write 函数可分别用来进行异步读和异步写操作。

#include <aio.h>
int aio_read(struct aiocb *aiocb);
int aio_write(struct aiocb *aiocb);
                            /* 两个函数的返回值：若成功，返回 0；否则，返回 -1 */

    当这些函数返回成功时，异步 I/O 请求便已经被系统放入等待处理的队列中了。这些返回值与实际 I/O 操作的结果没有任何关系。I/O 操作在等待时，必须注意确保 AIO 控制块和数据库缓冲区保持稳定，它们下面对应的内存必须始终是合法的，除非 I/O 操作完成，否则便不能被复用。
    要想强制所有等待中的异步操作不等待而写入持久化的存储中，可以调用 aio_fsync 函数。而要获知一个异步读、写或者同步操作的完成状态，可以调用 aio_error 和 aio_return 函数。

#include <aio.h>
int aio_fsync(int op, struct aiocb *aiocb);
                                      /* 返回值：若成功，返回 0；否则，返回 -1 */
int aio_error(const struct aiocb *aiocb);
ssize_t aio_return(const struct aiocb *aiocb);
                                     /* 两个函数的返回值：分别见下 */

    aio_fsync 在安排了同步后便返回，在异步同步操作完成前，数据不会被持久化。AIO 控制块中的 aio_fields 字段指定了其异步写操作被同步的文件。如果 op 参数被设置为 O_DSYNC，那么操作执行起来就会像调用了 fdatasync 一样。否则，如果 op 参数被设置为 O_SYNC，则操作执行起来就会像调用了 fsync 一样。
    aio_error 的返回值为下面 4 种情况中的一种。
    （1）0：表示异步操作成功完成，需要调用 aio_return 函数来获取操作返回值。
    （2）-1：表示对 aio_error 的调用失败，这可以通过查看 errno 来获知失败原因。
    （3）EINPROGRESS：表示异步读、写或同步操作仍在等待。
    （4）其他情况：其他任何的返回值是相关的异步操作失败返回的错误码。
    直到异步操作完成之前，都需要小心不要调用 aio_return 函数，否则结果是未定义的。而且还要小心对每个异步操作只调用一次 aio_return，因为一旦调用了该函数，操作系统就可以释放掉包含了 I/O 操作返回值的记录。如果 aio_return 本身失败，就会返回 -1，并设置 errno。否则其他情况下，它将直接返回 read、write 或 fsync 被成功调用时的结果。
    执行 I/O 操作时，如果还有其他事务要处理而不想被 I/O 操作阻塞，就可以使用异步 I/O。但如果在完成了所有事务后还有异步操作未完成时，就可以调用 aio_suspend 函数来阻塞进程，直到操作完成。而当还有不想再完成的等待中的异步 I/O 操作时，可以尝试用 aio_cancel 函数来取消它们。

#include <aio.h>
int aio_suspend(const struct aiocb *const list[], int nent, const struct timespec *timeout);
                                      /* 返回值：若成功，返回 0；否则，返回 -1 */
int aio_cancel(int fd, struct aiocb *aiocb);    /* 返回值：见下 */

    aio_suspend 的 list 参数是一个指向 AIO 控制块数组的指针，nent 参数表明了数组中的条目数，其中的空指针会被跳过，其他条目都必须指向已用于初始化异步 I/O 操作的 AIO控制块。如果该函数被一个信号中断，它会在返回 -1 的同时将 errno 设置为 EINTR，而在阻塞时间超时时，它会将 errno 设置为 EAGAIN（timeout 参数为空指针时表示无时间限制）。
    aio_cancel 的 fd 参数指定了那个未完成的异步 I/O 操作的文件描述符。如果 aiocb 参数为 NULL，系统将会尝试取消所有该文件上未完成的异步 I/O 操作。其他情况下，系统将尝试取消由 AIO 控制块描述的单个异步 I/O 操作。之所以说是“尝试”取消，是因为无法保证系统能够取消正在进程中的任何操作。该函数可能会返回以下 4 个值之一：
    （1）AIO_ALLDONE：所有操作在尝试取消之前已经完成。
    （2）AIO_CANCELED：所以要求的操作已被取消。
    （3）AIO_NOTCANCELED：至少有一个要求的操作没有被取消。
    （4）-1：函数调用失败，并会设置 errno。
    如果异步 I/O 操作被成功取消，则对相应的 AIO 控制块调用 aio_error 函数将会返回 ECANCELED；如果操作不能被取消，则相应的 AIO 控制块不会因为对 aio_cancel 的调用而被修改。
    还有一个函数 lio_listio 也被包含在异步 I/O 接口当中，尽管它既能以同步的方式来使用，又能以异步的方式来使用。该函数会提交一系列由一个 AIO 控制块列表描述的 I/O 请求。

#include <aio.h>
int lio_listio(int mode, struct aiocb *restrict const list[restrict], 
               int nent, struct sigevent *restrict sigev);
                                      /* 返回值：若成功，返回 0；否则，返回 -1 */

    其中 mode 参数决定了 I/O 是否真的是异步的：如果该参数被设置为 LIO_WAIT，则该函数将在所有由列表指定的 I/O 操作完成后返回，此时的 sigev 参数将被忽略；如果它被设置为 LIO_NOWAIT，则该函数将在 I/O 请求入队后立即返回，进程会在所有 I/O 操作完成后，按照 sigev 参数指定的事件被异步地通知。如果不想被通知，可以把 sigev 设置为 NULL（注意，每个 AIO 控制块本身也可能启用了在各自操作完成时的异步通知，被 sigev 参数指定的异步通知是在此之外另加的，并且只会在所有的 I/O 操作完成后发送）。
    list 参数指向 AIO 控制块列表，该列表指定了要运行的 I/O 操作，其中可以包含 NULL 指针，那些条目会被忽略。nent 参数指定了数组中的元素个数。在每一个 AIO 控制块中，aio_lio_opcode 字段指定了该操作是一个读操作（LIO_READ）、写操作（LIO_WRITE），还是将被忽略的空操作（LIO_NOP）。读操作会按照对应的 AIO 控制块被传给 aio_read 函数处理，写操作则按对应的 AIO 控制块被传给 aio_write 函数处理。
    异步 I/O 操作的数量受下表所示的运行时限制。

    可以通过调用 sysconf 函数并分别把 name 参数设置为 _SC_IO_LISTIO_MAX、_SC_AIO_MAX 和 _SC_AIO_PRIO_DELTA_MAX 来设置 AIO_LISTIO_MAX、AIO_MAX 和 AIO_PRIO_DELTA_MAX 的值。
    下面这个示例演示了如何使用异步 I/O 接口来实现一个 20 世纪 80 年代流行的 USENET 新闻系统中使用的 ROT-13 算法，该算法原本用于将文本中的带有侵犯性的或者含有剧透和笑话笑点部分的文章模糊化，它将文本中的英文字符 a~z 和 A~Z 分别循环向右偏移 13 个字母，其他字符则保持不变（其中为了避免代码臃肿，省略了除 aio_error 和 aio_return 调用之外的返回检测）。

#include <stdio.h>
#include <stdlib.h>
#include <aio.h>
#include <ctype.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <errno.h>

#define BSZ 40 //4096
#define NBUF 8
#define FILE_MODE (S_IRUSR |S_IWUSR |S_IRGRP |S_IROTH)

enum rwop{ UNUSED = 0, READ_PENDING = 1, WRITE_PENDING = 2 };
struct buf{
	enum rwop		op;
	int				last;
	struct aiocb	aiocb;
	unsigned char	data[BSZ];
};
struct buf	bufs[NBUF];

unsigned char translate(unsigned char c){
	if(isalpha(c)){
		if(c >= 'n') c -= 13;
		else if(c >= 'a') c += 13;
		else if(c >= 'N') c -= 13;
		else c += 13;
	}
	return c;
}

int main(int argc, char *argv[]){
	if(argc != 3){
		printf("usage: %s <infile> <outfile>\n", argv[0]);
		exit(1);
	}
	int ifd = open(argv[1], O_RDONLY);
	int ofd = open(argv[2], O_RDWR |O_CREAT |O_TRUNC, FILE_MODE);
	struct stat	sbuf;
	fstat(ifd, &sbuf);

	const struct aiocb	*aiolist[NBUF];
	int i, j;
	for(i=0; i<NBUF; i++){		// initialize the buffers
		bufs[i].op = UNUSED;
		bufs[i].aiocb.aio_buf = bufs[i].data;
		bufs[i].aiocb.aio_sigevent.sigev_notify = SIGEV_NONE;
		aiolist[i] = NULL;
	}
	int	numop = 0;
	off_t	off = 0;
	for(;;){
		for(i=0; i<NBUF; i++){
			int n, err;
			switch(bufs[i].op){
			case UNUSED:
				if(off < sbuf.st_size){
					bufs[i].op = READ_PENDING;
					bufs[i].aiocb.aio_fildes = ifd;
					bufs[i].aiocb.aio_offset = off;
					off += BSZ;
					if(off >= sbuf.st_size)
						bufs[i].last = 1;
					bufs[i].aiocb.aio_nbytes = BSZ;
					aio_read(&bufs[i].aiocb);
					aiolist[i] = &bufs[i].aiocb;
					numop++;
				}
				break;
			case READ_PENDING:
				if((err = aio_error(&bufs[i].aiocb)) == EINPROGRESS)
					continue;
				if(err != 0){
					if(err == -1)
						printf("aio_error failed\n");
					else
						printf("read failed, errno = %d\n", err);
					continue;
				}
				if((n = aio_return(&bufs[i].aiocb)) < 0){
					printf("aio_return failed\n");
					continue;
				}
				if(n != BSZ && !bufs[i].last)
					printf("short read (%d/%d)\n", n, BSZ);
				for(j=0; j<n; j++)	// translate the buffer
					bufs[i].data[j] = translate(bufs[i].data[j]);
				bufs[i].op = WRITE_PENDING;
				bufs[i].aiocb.aio_fildes = ofd;
				bufs[i].aiocb.aio_nbytes = n;
				aio_write(&bufs[i].aiocb);
				break;					// retain our spot in aiolist
			case WRITE_PENDING:
				if((err = aio_error(&bufs[i].aiocb)) == EINPROGRESS)
					continue;
				if(err != 0){
					if(err == -1)
						printf("aio_error failed\n");
					else
						printf("read failed, errno = %d\n", err);
					continue;
				}
				if((n = aio_return(&bufs[i].aiocb)) < 0){
					printf("aio_return failed\n");
					continue;
				}
				if(n != bufs[i].aiocb.aio_nbytes)
					printf("short write (%d/%d)\n", n, BSZ);
				aiolist[i] = NULL;
				bufs[i].op = UNUSED;
				numop--;
				break;
			}
		}
		if(numop == 0)
			if(off >= sbuf.st_size)
				break;
		else
			aio_suspend(aiolist, NBUF, NULL);
	}
	bufs[0].aiocb.aio_fildes = ofd;
	aio_fsync(O_SYNC, &bufs[0].aiocb);
	exit(0);
}

    这里使用了 8 个缓冲区，因此可以有最多 8 个异步 I/O 请求处于等待状态。但令人惊讶的是，这可能会降低性能，因为如果读操作是以无序的方式提交给文件系统的，那么操作系统的预读算法便会失效。本示例没有使用异步通知，如果在 I/O 操作进行时还有别的事要做，那么额外的工作可以包含在 for 循环当中。而如果要阻止这些额外的工作延迟翻译文件的任务，就应该考虑使用异步通知。多任务情况下，可能还要考虑各个任务的优先级。
    如果将本代码文件作为输入文件运行该程序，可得到下面的结果。

$ gcc -lrt aio_rot13.c -o aio_rot13.out    # 编译文件
$ ./aio_rot13.out aio_rot13.c aio.test     # 执行程序
$ head -n 5 aio.test                       # 查看翻译后的文件的前五行
#vapyhqr <fgqvb.u>
#vapyhqr <fgqyvo.u>
#vapyhqr <nvb.u>
#vapyhqr <pglcr.u>
#vapyhqr <spagy.u>
$