Linux AIO

Linux aio是Linux下的異步讀寫模型。Linux 異步 I/O 是 Linux 內核中提供的一個至關新的加強。它是 2.6 版本內核的一個標準特性。對於文件的讀寫，即便以O_NONBLOCK方式來打開一個文件，也會處於"阻塞"狀態。由於文件時時刻刻處於可讀狀態。而從磁盤到內存所等待的時間是驚人的。爲了充份發揮把數據從磁盤複製到內存的時間，引入了aio模型。AIO 背後的基本思想是容許進程發起不少 I/O 操做，而不用阻塞或等待任何操做完成。稍後或在接收到 I/O 操做完成的通知時，進程就能夠檢索 I/O 操做的結果。

 

I/O 模型

在深刻介紹 AIO API 以前，讓咱們先來探索一下 Linux 上可使用的不一樣 I/O 模型。這並非一個詳盡的介紹，可是咱們將試圖介紹最經常使用的一些模型來解釋它們與異步 I/O 之間的區別。圖 1 給出了同步和異步模型，以及阻塞和非阻塞的模型。


圖 1. 基本 Linux I/O 模型的簡單矩陣
 
每一個 I/O 模型都有本身的使用模式，它們對於特定的應用程序都有本身的優勢。本節將簡要對其一一進行介紹。

 

同步阻塞 I/O

最經常使用的一個模型是同步阻塞 I/O 模型。在這個模型中，用戶空間的應用程序執行一個系統調用，這會致使應用程序阻塞。這意味着應用程序會一直阻塞，直到系統調用完成爲止（數據傳輸完成或發生錯誤）。調用應用程序處於一種再也不消費 CPU 而只是簡單等待響應的狀態，所以從處理的角度來看，這是很是有效的。

圖 2 給出了傳統的阻塞 I/O 模型，這也是目前應用程序中最爲經常使用的一種模型。其行爲很是容易理解，其用法對於典型的應用程序來講都很是有效。在調用 read 系統調用時，應用程序會阻塞並對內核進行上下文切換。而後會觸發讀操做，當響應返回時（從咱們正在從中讀取的設備中返回），數據就被移動到用戶空間的緩衝區中。而後應用程序就會解除阻塞（read 調用返回）。

圖 2. 同步阻塞 I/O 模型的典型流程
 
從應用程序的角度來講，read 調用會延續很長時間。實際上，在內核執行讀操做和其餘工做時，應用程序的確會被阻塞。

 

同步非阻塞 I/O

同步阻塞 I/O 的一種效率稍低的變種是同步非阻塞 I/O。在這種模型中，設備是以非阻塞的形式打開的。這意味着 I/O 操做不會當即完成，read 操做可能會返回一個錯誤代碼，說明這個命令不能當即知足（EAGAIN 或 EWOULDBLOCK），如圖 3 所示。


圖 3. 同步非阻塞 I/O 模型的典型流程
 
非阻塞的實現是 I/O 命令可能並不會當即知足，須要應用程序調用許屢次來等待操做完成。這可能效率不高，由於在不少狀況下，當內核執行這個命令時，應用程序必需要進行忙碌等待，直到數據可用爲止，或者試圖執行其餘工做。正如圖 3 所示的同樣，這個方法能夠引入 I/O 操做的延時，由於數據在內核中變爲可用到用戶調用 read 返回數據之間存在必定的間隔，這會致使總體數據吞吐量的下降。

 

異步阻塞 I/O

另一個阻塞解決方案是帶有阻塞通知的非阻塞 I/O。在這種模型中，配置的是非阻塞 I/O，而後使用阻塞 select 系統調用來肯定一個 I/O 描述符什麼時候有操做。使 select 調用很是有趣的是它能夠用來爲多個描述符提供通知，而不只僅爲一個描述符提供通知。對於每一個提示符來講，咱們能夠請求這個描述符能夠寫數據、有讀數據可用以及是否發生錯誤的通知。


圖 4. 異步阻塞 I/O 模型的典型流程 (select)
 
select 調用的主要問題是它的效率不是很是高。儘管這是異步通知使用的一種方便模型，可是對於高性能的 I/O 操做來講不建議使用。

 

異步非阻塞 I/O（AIO）

最後，異步非阻塞 I/O 模型是一種處理與 I/O 重疊進行的模型。讀請求會當即返回，說明 read 請求已經成功發起了。在後臺完成讀操做時，應用程序而後會執行其餘處理操做。當 read 的響應到達時，就會產生一個信號或執行一個基於線程的回調函數來完成此次 I/O 處理過程。


圖 5. 異步非阻塞 I/O 模型的典型流程
 

在一個進程中爲了執行多個 I/O 請求而對計算操做和 I/O 處理進行重疊處理的能力利用了處理速度與 I/O 速度之間的差別。當一個或多個 I/O 請求掛起時，CPU 能夠執行其餘任務；或者更爲常見的是，在發起其餘 I/O 的同時對已經完成的 I/O 進行操做。

 

從前面 I/O 模型的分類中，咱們能夠看出 AIO 的動機。這種阻塞模型須要在 I/O 操做開始時阻塞應用程序。這意味着不可能同時重疊進行處理和 I/O 操做。同步非阻塞模型容許處理和 I/O 操做重疊進行，可是這須要應用程序根據重現的規則來檢查 I/O 操做的狀態。這樣就剩下異步非阻塞 I/O 了，它容許處理和 I/O 操做重疊進行，包括 I/O 操做完成的通知。除了須要阻塞以外，select 函數所提供的功能（異步阻塞 I/O）與 AIO 相似。不過，它是對通知事件進行阻塞，而不是對 I/O 調用進行阻塞。


Linux 上的 AIO 簡介


linux下有aio封裝，aio_*系列的調用是glibc提供的，是glibc用線程+阻塞調用來模擬的，性能不好，爲了能更多的控制io行爲，可使用更爲低級libaio。
libaio項目: http://oss.oracle.com/projects/libaio-oracle/

libaio的使用並不複雜，過程爲：libaio的初始化，io請求的下發和回收，libaio銷燬。

1、libaio接口
libaio提供下面五個主要API函數：

int io_setup(int maxevents, io_context_t *ctxp);
int io_destroy(io_context_t ctx);
int io_submit(io_context_t ctx, long nr, struct iocb *ios[]);
int io_cancel(io_context_t ctx, struct iocb *iocb, struct io_event *evt);
int io_getevents(io_context_t ctx_id, long min_nr, long nr, struct io_event *events, struct timespec *timeout);

五個宏定義：

void io_set_callback(struct iocb *iocb, io_callback_t cb);
void io_prep_pwrite(struct iocb *iocb, int fd, void *buf, size_t count, long long offset);
void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset);
void io_prep_pwritev(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset);
void io_prep_preadv(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset);

這五個宏定義都是操做struct iocb的結構體。struct iocb是libaio中很重要的一個結構體，用於表示IO，可是其結構略顯複雜，爲了保持封裝性不建議直接操做其元素而用上面五個宏定義操做。

2、libaio的初始化和銷燬

觀察libaio五個主要API，都用到類型爲io_context的變量，這個變量爲libaio的工做空間。不用具體去了解這個變量的結構，只須要了解其相關操做。建立和銷燬libaio分別用到io_setup（也能夠用io_queue_init，區別只是名字不同而已）和io_destroy。

int io_setup(int maxevents, io_context_t *ctxp);
int io_destroy(io_context_t ctx);

3、libaio讀寫請求的下發和回收

1. 請求下發

libaio的讀寫請求都用io_submit下發。下發前經過io_prep_pwrite和io_prep_pread生成iocb的結構體，作爲io_submit的參數。這個結構體中指定了讀寫類型、起始扇區、長度和設備標誌符。

libaio的初始化不是針對一個具體設備進行初始，而是建立一個libaio的工做環境。讀寫請求下發到哪一個設備是經過open函數打開的設備標誌符指定。


2. 請求返回

讀寫請求下發以後，使用io_getevents函數等待io結束信號：

int io_getevents(io_context_t ctx_id, long min_nr, long nr, struct io_event *events, struct timespec *timeout);

io_getevents返回events的數組，其參數events爲數組首地址，nr爲數組長度（即最大返回的event數），min_nr爲最少返回的events數。timeout可填NULL表示無等待超時。io_event結構體的聲明爲：

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

其中，res爲實際完成的字節數；res2爲讀寫成功狀態，0表示成功；obj爲以前下發的struct iocb結構體。這裏有必要了解一下struct iocb這個結構體的主要內容：

iocbp->iocb.u.c.nbytes 字節數
iocbp->iocb.u.c.offset 偏移
iocbp->iocb.u.c.buf 緩衝空間
iocbp->iocb.u.c.flags 讀寫

3. 自定義字段

struct iocb除了自帶的元素外，還留有供用戶自定義的元素，包括回調函數和void *的data指針。若是在請求下發前用io_set_callback綁定用戶自定義的回調函數，那麼請求返回後就能夠顯示的調用該函數。回調函數的類型爲：

void callback_function(io_context_t ctx, struct iocb *iocb, long res, long res2);

另外，還能夠經過iocbp->data指針掛上用戶本身的數據。

注意：實際使用中發現回調函數和data指針不能同時用，可能回調函數自己就是使用的data指針。


4、使用例子經過上面的說明並不能完整的瞭解libaio的用法，下面經過簡單的例子進一步說明。#include <stdlib.h>#include <stdio.h>#include <libaio.h>#include <sys/stat.h>#include <fcntl.h>#include <libaio.h>int srcfd=-1;int odsfd=-1;#define AIO_BLKSIZE  1024#define AIO_MAXIO 64static 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);       }        free(iocb->u.c.buf);       free(iocb);}static void rd_done(io_context_t ctx, struct iocb *iocb, long res, long res2){       /*library needs accessors to look at iocb*/       int iosize = iocb->u.c.nbytes;       char *buf = (char *)iocb->u.c.buf;       off_t offset = iocb->u.c.offset;       int  tmp;       char *wrbuff = NULL;        if(res2 != 0)       {             printf(「aio read\n」);       }       if(res != iosize)       {              printf( 「read missing bytes expect %d got %d」, iocb->u.c.nbytes, res);              exit(1);       }        /*turn read into write*/       tmp = posix_memalign((void **)&wrbuff, getpagesize(), AIO_BLKSIZE);       if(tmp < 0)       {              printf(「posix_memalign222\n」);              exit(1);       }        snprintf(wrbuff, iosize + 1, 「%s」, buf);        printf(「wrbuff-len = %d:%s\n」, strlen(wrbuff), wrbuff);       printf(「wrbuff_len = %d\n」, strlen(wrbuff));       free(buf);              io_prep_pwrite(iocb, odsfd, wrbuff, iosize, offset);       io_set_callback(iocb, wr_done);              if(1!= (res=io_submit(ctx, 1, &iocb)))              printf(「io_submit write error\n」);              printf(「\nsubmit  %d  write request\n」, res);}void main(int args,void * argv[]){    int length = sizeof(「abcdefg」);    char * content = (char * )malloc(length);    io_context_t myctx;    int rc;    char * buff=NULL;    int offset=0;    int num,i,tmp;        if(args<3)    {        printf(「the number of param is wrong\n」);        exit(1);    }      if((srcfd=open(argv[1],O_RDWR))<0)      {        printf(「open srcfile error\n」);        exit(1);      }      printf(「srcfd=%d\n」,srcfd);      lseek(srcfd,0,SEEK_SET);      write(srcfd,」abcdefg」,length);            lseek(srcfd,0,SEEK_SET);      read(srcfd,content,length);      printf(「write in the srcfile successful,content is %s\n」,content);      if((odsfd=open(argv[2],O_RDWR))<0)      {        close(srcfd);        printf(「open odsfile error\n」);        exit(1);      }    memset(&myctx, 0, sizeof(myctx));    io_queue_init(AIO_MAXIO, &myctx);             struct iocb *io = (struct iocb*)malloc(sizeof(struct iocb));       int iosize = AIO_BLKSIZE;       tmp = posix_memalign((void **)&buff, getpagesize(), AIO_BLKSIZE);        if(tmp < 0)        {              printf(「posix_memalign error\n」);              exit(1);         }        if(NULL == io)        {            printf( 「io out of memeory\n」);             exit(1);        }                                   io_prep_pread(io, srcfd, buff, iosize, offset);                              io_set_callback(io, rd_done);                         printf(「START…\n\n」);                         rc = io_submit(myctx, 1, &io);              if(rc < 0)               printf(「io_submit read error\n」);                    printf(「\nsubmit  %d  read request\n」, rc);                                  //m_io_queue_run(myctx);                 struct io_event events[AIO_MAXIO];          io_callback_t cb;                   num = io_getevents(myctx, 1, AIO_MAXIO, events, NULL);          printf(「\n%d io_request completed\n\n」, num);                for(i=0;i<num;i++)         {              cb = (io_callback_t)events[i].data;              struct iocb *io = events[i].obj;                          printf(「events[%d].data = %x, res = %d, res2 = %d\n」, i, cb, events[i].res, events[i].res2);              cb(myctx, io, events[i].res, events[i].res2);          }     }