异步IO之AIO

什么是AIO

AIO：Asynchronous Input and Output，字面上讲就是异步的IO。说到异步IO，网络上的端口复用select、poll机制大家就很熟悉了，这个AIO到底是啥？

AIO是Linux2.6中新加的特性，支持异步的IO读写，IO繁忙或者IO耗时较多时，可以将IO操作交给AIO，程序处理其他逻辑，实现并行加速，AIO一定程度上也能够减少IO操作的频率，减少IO负担

AIO API

在linux /usr/include的下面，有aio.h这样一个文件，打开之后，可以看到如下函数：

API 函数	说明
aio_read	请求异步读操作
aio_error	检查异步请求的状态
aio_return	获得完成的异步请求的返回状态
aio_write	请求异步写操作
aio_suspend	挂起调用进程，直到一个或多个异步请求已经完成（或失败）
aio_cancel	取消异步 I/O 请求
aio_fsync	强制同步
lio_listio	发起一系列 I/O 操作

AIO控制结构

/* Asynchronous I/O control block.  */

 struct aiocb

 {

   int aio_fildes;       /* File desriptor.  */

   int aio_lio_opcode;       /* Operation to be performed.  */

   int aio_reqprio;      /* Request priority offset.  */

   volatile void *aio_buf;   /* Location of buffer.  */

   size_t aio_nbytes;        /* Length of transfer.  */

   struct sigevent aio_sigevent; /* Signal number and value.  */

 

   /* Internal members.  */

   struct aiocb *__next_prio;

   …

};

从定义中可以看到，核心结构还不算，句柄、操作码、offset都比较容易理解，buffer用volatile来修饰，应该是内部采用多线程的原因，长度就不用说了，这里需要研究的就剩下sigevent需要了解，而内部（internal）数据看到第一个指针就明白是链表了。

sigevent

明显，是一个处理signal事件的结构体，定义如下：

struct sigevent {

    int          sigev_notify; /* Notification method */

    int          sigev_signo;  /* Notification signal */

    union sigval sigev_value;  /* Data passed with

                                  notification */

    void       (*sigev_notify_function) (union sigval);

                     /* Function used for thread

                        notification (SIGEV_THREAD) */

    void        *sigev_notify_attributes;

                     /* Attributes for notification thread

                        (SIGEV_THREAD) */

    pid_t        sigev_notify_thread_id;

                     /* ID of thread to signal (SIGEV_THREAD_ID) */

};

Ø  sigev_notify为处理模式，为SIGEV_NONE、SIGEV_SIGNAL、SIGEV_THREAD、SIGEV_THREAD_ID（只针对linux）当中的一个；

Ø  sigev_signo为signal的值，当sigev_notify为SIGEV_SIGNAL时，会将这个signal发送给进程；

Ø  sigev_value，信号传递的数据；

Ø  sigev_notify_function，当sigev_notify为SIGEV_THREAD时，处理线程将调用这个处理函数；

Ø  sigev_notify_attributes，sigev_notify_function的参数；

Ø  sigev_notify_thread_id，当sigev_notify为SIGEV_THREAD_ID时的处理线程ID。

aio_read函数

函数原型：

/* Enqueue read request for given number of bytes and the given priority.  */

 extern int aio_read (struct aiocb *__aiocbp) __THROW;

先暂时不理sigevent，我们来一个简单的示例代码：

#include <unistd.h>

#include <stdio.h>

#include <aio.h>

#include <string.h>

#include <sys/types.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <errno.h>

#include <sys/timeb.h>

 

#define BUFFER_SIZE 1024*1024

 

void ptime(const char* tip = ""){

    struct timeb tb;

    ftime(&tb);

    fprintf(stdout, "%s %u : %u\n", tip, tb.time, tb.millitm);

}

 

int main(){

    /* 句柄，返回码 */

    int fd = -1, ret = -1;

 

    fd = open("./file.txt", O_RDONLY);

    if(fd <= 0){

        fprintf(stderr, "open file errro: %s\n", strerror(errno));

        return -1;

    }

 

    /* aio控制结构 */

    aiocb my_aiocb;

    memset(&my_aiocb, 0, sizeof(my_aiocb));

 

    /* 初始化 */

    my_aiocb.aio_fildes = fd;

    my_aiocb.aio_reqprio = 0;

    my_aiocb.aio_nbytes = BUFFER_SIZE;

    char buf[BUFFER_SIZE + 1] = {0};

    my_aiocb.aio_buf = buf;

   

    ptime("start read");

 

    /* aio 读 */

    ret = aio_read(&my_aiocb);

    if(ret < 0){

        fprintf(stderr, "aio read error: %s\n", strerror(errno));

        return -2;

    }

 

    ptime("reading");

 

    /* 检查状态 */

    while(aio_error(&my_aiocb) == EINPROGRESS);

   

    ptime("after read");

 

    /* aio 返回 */

    if(ret = aio_return(&my_aiocb) <= 0){

        fprintf(stdout, "return: %d\n", ret);

    }

    else{

        fprintf(stdout, "read: %10.10s\n", my_aiocb.aio_buf);

    }

 

    close(fd);

 

    return 0;

}

编译时，需要加上-lrt选项，这一点在man手册中没有说明实在是有些不厚道。

start read 1370399057 : 740

reading 1370399057 : 740

after read 1370399057 : 747

read: 1234567890

可以看到，整个过程耗时约为7毫秒，至于效率如何，咱们先放到一边，往后继续。
有了以上实例，下面的函数理解起来就比较简单了。

aio_error函数

函数原型：

  /* Retrieve error status associated with AIOCBP.  */

 extern int aio_error (__const struct aiocb *__aiocbp) __THROW;

从man手册中可以看到，这个函数可以用来检测aio当前状态：

Ø  EINPROGRESS AIO处理中，请求尚未完成

Ø  ECANCELED AIO被取消

Ø  0 成功

Ø  其他 出错，错误码放在errno中

aio_return函数

函数原型

/* Return status associated with AIOCBP.  */

 extern __ssize_t aio_return (struct aiocb *__aiocbp) __THROW;

这个函数理解起来更加简单，用于获取AIO操作的返回码，类似一般read/write函数的返回值。

aio_write函数

函数原型：

/* Enqueue write request for given number of bytes and the given priority.  */

 extern int aio_write (struct aiocb *__aiocbp) __THROW;

这与 read 系统调用类似，但是有一点不一样的行为需要注意。

回想一下对于 read 调用来说，要使用的偏移量是非常重要的。然而，对于 write 来说，这个偏移量只有在没有设置 O_APPEND 选项的文件上下文中才会非常重要。

如果设置了 O_APPEND，那么这个偏移量就会被忽略，数据都会被附加到文件的末尾；否则，aio_offset 域就确定了数据在要写入的文件中的偏移量。

aio_suspend函数

这个函数需要稍微说明一下，功能是将调用进程（线程）挂起，直到AIO完成，有些类似wait的处理逻辑，函数原型如下：

/* Suspend calling thread until at least one of the asynchronous I/O

    operations referenced by LIST has completed.

 

    This function is a cancellation point and therefore not marked with

    __THROW.  */

 extern int aio_suspend (__const struct aiocb *__const __list[], int __nent,

             __const struct timespec *__restrict __timeout);

Ø  __list为AIO控制结构的数组；

Ø  __nent为__list中元素的个数；

Ø  __timeout为超时时间；

如果指定tiemout，则函数返回非0值，errno设置为EAGAIN

aio_cancel函数

不用多讲，应该很容易就能明白，取消一个AIO，函数原型如下：

/* Try to cancel asynchronous I/O requests outstanding against file

    descriptor FILDES.  */

 extern int aio_cancel (int __fildes, struct aiocb *__aiocbp) __THROW;

要取消一个请求，我们需要提供文件描述符和 aiocb 引用。如果这个请求被成功取消了，那么这个函数就会返回 AIO_CANCELED。如果请求完成了，这个函数就会返回 AIO_NOTCANCELED。

要取消对某个给定文件描述符的所有请求，我们需要提供这个文件的描述符，以及一个对 aiocbp 的 NULL 引用。

如果所有的请求都取消了，这个函数就会返回 AIO_CANCELED；如果至少有一个请求没有被取消，那么这个函数就会返回 AIO_NOT_CANCELED；如果没有一个请求可以被取消，那么这个函数就会返回 AIO_ALLDONE。我们然后可以使用 aio_error 来验证每个 AIO 请求。

如果这个请求已经被取消了，那么 aio_error 就会返回 -1，并且 errno 会被设置为 ECANCELED。

aio_fsync函数

在AIO是交给其他线程来完成的，如果需要手动执行同步，则需要调用这个函数，原型为：

/* Force all operations associated with file desriptor described by

    `aio_fildes' member of AIOCBP.  */

 extern int aio_fsync (int __operation, struct aiocb *__aiocbp) __THROW;

Ø  __operation为操作码；

Ø  __aiocbp为异步IO的控制结构；

函数执行时，将强制完成该AIO上的所有操作。具体来讲，如果操作码为O_SYNC，则同步异步IO数据，当前所有IO操作均将完成；如果操作码是O_DSYNC，则只是一个IO请求，并不等待所有的IO完成。

与一般文件IO的fsync用法基本一致。

lio_listio函数

放在最后讲，是因为它居然不是aio打头的，函数原型如下

/* Initiate list of I/O requests.  */

 extern int lio_listio (int __mode,

                struct aiocb *__const __list[__restrict_arr],

                int __nent, struct sigevent *__restrict __sig) __THROW;

这个函数结构就复杂多了，我们来慢慢分析。

这个函数是用来发起一系列的IO请求，相当于并发读写多个文件，但是这样与写循环有什么区别呢？

先看mode参数，有LIO_WAIT、LIO_NOWAIT可选，这个容易理解，就是阻塞和非阻塞的区别。

第二个参数和第三个参数aio_suspend一致，不再赘述，需要说明的aiocb结构体在aio_read、aio_write中无需指定操作码，函数本身就已经确定了操作，回头看下结构，里面果然有aio_lio_opcode这样一个参数，man手册中看到该取值可以为LIO_READ、LIO_WRITE及LIO_NOP。读、写好理解，这个LIO_NOP又是什么呢？“The LIO_NOP operation causes the list entry to be ignored.”原来什么也不干也行。

还有一个sigevent，看来还是需要将sigevent搞明白才行，不过，可以看到man手册中，在mode为LIO_NOWAIT的时候，sigevent才会有效。这个也容易理解，非阻塞方式才需要回调信号处理函数，否则没有意义。

注意：在错误码中，有提到AIO_MAX、AIO_LISTIO_MAX限制，查了下资料，这个跟系统有关，当nent参数超过最大值时，会出现EINVAL错误。

sigevent使用解析

一般signal处理我们常使用signal函数，或者sigaction方法来进行绑定即可，sigevent接触的比较少。

Sigevent一般用于异步请求结果通知（咱们讨论的话题）、定时器（sigev_value函数）、消息抵达（mq_notify）。

union sigval {          /* Data passed with notification */

   int     sival_int;         /* Integer value */

   void   *sival_ptr;         /* Pointer value */

};

 

struct sigevent {

   int          sigev_notify; /* Notification method */

   int          sigev_signo;  /* Notification signal */

   union sigval sigev_value;  /* Data passed with

                             notification */

   void       (*sigev_notify_function) (union sigval);

                  /* Function used for thread

                     notification (SIGEV_THREAD) */

   void        *sigev_notify_attributes;

                  /* Attributes for notification thread

                     (SIGEV_THREAD) */

   pid_t        sigev_notify_thread_id;

                  /* ID of thread to signal (SIGEV_THREAD_ID) */

};

先不管其他模式，我们先看sigev_notify等于SIGEV_THREAD的情况。当为SIGEV_THREAD线程处理模式时，相关参数为：

Ø  sigev_notify_function线程处理函数指针。需要注意的是，函数是类型是返回值为空，且带一个union sigval参数，sigval包含两个值，一个int的数值型，一个void*的指针；

Ø  sigev_value这个就是上面函数指针的参数；

Ø  sigev_notify_attributes为处理线程的属性参数，void*类型有些让人摸不着头脑，一般情况下，给NULL值就好，如果需要指定处理线程属性的话，该值是一个指向pthread_attr_t结构体的指针，具体定义可以参见pthread。

OK，现在咱们清晰明了，说白了sigevent就是当信号到达时，指定特定的处理方式来处理该信号。当sigev_notify为其他模式时，使用方式也不复杂，具体可以参见http://man7.org/linux/man-pages/man7/sigevent.7.html

说了那么多，我们还是来看个例子吧，简单明了：

#include <unistd.h>

#include <stdio.h>

#include <aio.h>

#include <string.h>

#include <sys/types.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <errno.h>

#include <sys/timeb.h>

#include <pthread.h>

 

#define BUFFER_SIZE 1024*1014

 

void ptime(const char* tip = ""){

    struct timeb tb;

    ftime(&tb);

    fprintf(stdout, "%s %u : %u\n", tip, tb.time, tb.millitm);

}

 

void aio_finish_hander(union sigval para){

    ptime("finish read");

    struct aiocb my_aiocb = *(struct aiocb*)(para.sival_ptr);

 

    int ret = -1;

 

    /* 检查状态 */

    ret = aio_error(&my_aiocb);

    if(0 == ret){

        /* aio 返回 */

        if(ret = aio_return(&my_aiocb) <= 0){

            fprintf(stdout, "return: %d\n", ret);

            fprintf(stderr, "error: %s\n" ,strerror(errno));

        }

        else{

            fprintf(stdout, "read: %10.10s\n", my_aiocb.aio_buf);

        }

    }

    else{

        fprintf(stderr, "%d error: %s\n" ,ret, strerror(errno));

    }

}

 

int main(){

    /* 句柄，返回码 */

    int fd = -1, ret = -1;

 

    fd = open("./file.txt", O_RDONLY);

    if(fd <= 0){

        fprintf(stderr, "open file errro: %s\n", strerror(errno));

        return -1;

    }

 

    fprintf(stdout, "fd no %d\n", fd);

 

    /* aio控制结构 */

    struct aiocb my_aiocb;

    memset(&my_aiocb, 0, sizeof(my_aiocb));

 

    /* 初始化 */

    my_aiocb.aio_fildes = fd;

    my_aiocb.aio_reqprio = 0;

    my_aiocb.aio_nbytes = BUFFER_SIZE;

    char buf[BUFFER_SIZE+1] = {0};

    my_aiocb.aio_buf = buf;

 

    /* sigevent初始化 */

    my_aiocb.aio_sigevent.sigev_notify = SIGEV_THREAD;

    my_aiocb.aio_sigevent.sigev_notify_function = aio_finish_hander;

    my_aiocb.aio_sigevent.sigev_notify_attributes = NULL;

    my_aiocb.aio_sigevent.sigev_value.sival_int = fd;

    fprintf(stdout, "int para %d\n",my_aiocb.aio_sigevent.sigev_value.sival_int);

    my_aiocb.aio_sigevent.sigev_value.sival_ptr = (void*)(&my_aiocb);

   

    ptime("start read");

 

    /* aio 读 */

    ret = aio_read(&my_aiocb);

    if(ret < 0){

        fprintf(stderr, "aio read error: %s\n", strerror(errno));

        return -2;

    }

 

    ptime("after read");

 

    sleep(3);

 

    close(fd);

 

    return 0;

}

需要说明的是，aio_read之后，不要先将文件句柄关闭，否则sigevent处理线程内aio函数将报错，当然，也不能在线程处理函数还未结束时，就将主线程退出（sleep一下），这样压根就看不到效果了。

程序运行结果如下：

fd no 4

int para 4

start read 1370399183 : 475

after read 1370399183 : 475

finish read 1370399183 : 476

read: 1234567890

看来比轮询aio_error查询AIO状态要快，耗时约为1毫秒

AIO效率如何

简单测试

AIO效率到底如何？咱们先直接读取试一试（代码就不贴了，太简单），运行结果如下：

fd no 4

start read 1370399316 : 870

after read 1370399316 : 871

read: 1234567890

耗时约为1毫秒，基本与AIO一致！看来在单次件读写AIO基本没有啥优势，考虑到线程开销，应该比一般IO要慢才对。

由于测试资源有限，读取大文件50M，各个方法耗时如下：

直接Read读	AIO读，aio_error轮询检查	AIO Sigevent读
start read 1370400772 : 951 after read 1370400773 : 59 read: 1234567890	start read 1370400993 : 722 reading 1370400993 : 722 after read 1370400993 : 900 read: 1234567890	start read 1370401097 : 898 after read 1370401097 : 898 finish read 1370401098 : 8 read: 1234567890
耗时：108毫秒	耗时：178毫秒	耗时：110毫秒

更加可以确定，在单次IO上，AIO效率并没有提升。

同样的道理，采用多线程并发去读写，与AIO读写差别也不大。

Lio_listio并发测试

上面的测试结果多少在意料之中，也没太多失望，那么，作为可以单次处理多个异步IO的AIO核心函数，读写效率是否会上升呢？

由于在我测试环境中，AIO_LISTIO_MAX仅仅为21，没有较强的说明性，不过咱们还是简单的来看一下，21次读取50M的文件，耗时是多少呢？854毫秒，其中，listio耗时为25毫秒。

对比上面的简单测试内容，如果直接单线程循环的话，21次读写理论值为2秒多一点儿，看来，对比单线程来讲，效率提升2.5倍，而且，这个数据应该还是与CPU内核数量有关系（测试环境仅为双核处理器）。

到这里，咱们应该很清楚了，对比自己写多线程处理，AIO的肯定会慢一些，跟简单测试中的结果应该类似，大家不妨试一试。

小结

AIO说白了，就是封装好了的多线程处理异步IO，这多少让我想起Java下面的NIO。对比NIO在java中的大放异彩，AIO却很少被人提及。

AIO在哪里被使用？目前，数据库使用的比较多，如Oracle、InnoDB，其他的就不太清楚了。

什么时候使用 AIO ？了解 AIO 机制之后，不难发现， AIO 其实是用于解决大量 IO 并发操作而出现的，牺牲一些信号处理耗时，用多线程的方式加速 IO ，提高并行效率。