http://blog.csdn.net/xinyuwuxian/article/details/9359685

成于堅持，敗于止步

Linux 2.6 異步 I/O 

AIO 概念與 GNU C 庫函數(shù) 

Linux 系統(tǒng)中最常用的輸入/輸出（I/O）模型是同步 I/O。在這個模型中，當請求發(fā)出之后，應用程序就會阻塞，直到請求滿足為止。這是很好的一種解決方案，因為調(diào)用應用程序在等待 I/O 請求完成時不需要使用任何中央處理單元（CPU）。但是在某
些情況下，I/O 請求可能需要與其他進程產(chǎn)生交疊?？梢浦膊僮飨到y(tǒng)接口（POSIX）異步 I/O（AIO）應用程序接口（API）就提供了這種功能。 

Linux 異步 I/O 是 2.6 版本內(nèi)核的一個標準特性，但是我們在 2.4 版本內(nèi)核的補丁中也可以找到它。AIO 基本思想是允許進程發(fā)起很多 I/O 操作，而不用阻塞或等待任何操作完成。稍后或在接收到 I/O 操作完成的通知時，進程就可以檢索 I/O 操作的結
果。 

select()函數(shù)所提供的功能（異步阻塞 I/O）與 AIO 類似，它對通知事件進行阻塞，而不是對 I/O 調(diào)用進行阻塞。 

在異步非阻塞 I/O 中，我們可以同時發(fā)起多個傳輸操作。這需要每個傳輸操作都有惟一的上下文，這樣才能在它們完成時區(qū)分到底是哪個傳輸操作完成了。在 AIO 中，通過 aiocb（AIO I/O Control Block）結構體進行區(qū)分。這個結構體包含了有關傳輸?shù)乃行畔ⅲ閿?shù)據(jù)準備的用戶緩沖區(qū)。在產(chǎn)生 I/O（稱為完成）通知時，aiocb 結構就被用來惟一標識所完成的 I/O 操作。 

AIO 系列 API 被 GNU C 庫函數(shù)所包含，它被 POSIX.1b 所要求，主要包括如下函數(shù)。 

1．a(chǎn)io_read 

aio_read()函數(shù)請求對一個有效的文件描述符進行異步讀操作。這個文件描述符可以表示一個文件、套接字甚至管道

aio_read 函數(shù)的原型如下： 

int aio_read( struct aiocb *aiocbp ); 

aio_read()函數(shù)在請求進行排隊之后會立即返回。如果執(zhí)行成功，返回值就為 0；如果出現(xiàn)錯誤，返回值就為?1，并設置 errno 的值。 

2．a(chǎn)io_write 

aio_write()函數(shù)用來請求一個異步寫操作，其函數(shù)原型如下： 

int aio_write( struct aiocb *aiocbp ); 

aio_write()函數(shù)會立即返回，說明請求已經(jīng)進行排隊（成功時返回值為 0，失敗時返回值為?1，并相應地設置 errno。 

3．a(chǎn)io_error 

aio_error 函數(shù)被用來確定請求的狀態(tài)，其原型如下： 

int aio_error( struct aiocb *aiocbp ); 

這個函數(shù)可以返回以下內(nèi)容。 

EINPROGRESS：說明請求尚未完成。 

ECANCELLED：說明請求被應用程序取消了。 

-1：說明發(fā)生了錯誤，具體錯誤原因由 errno 記錄。 

4．a(chǎn)io_return 

異步 I/O 和標準塊 I/O 之間的另外一個區(qū)別是不能立即訪問這個函數(shù)的返回狀態(tài)，因為并沒有阻塞在 read()調(diào)用上。在標準的 read()調(diào)用中，返回狀態(tài)是在該函數(shù)返回時提供的。但是在異步 I/O 中，我們要使用 aio_return()函數(shù)。這個函數(shù)的原型如下：

ssize_t aio_return( struct aiocb *aiocbp ); 

只有在 aio_error()調(diào)用確定請求已經(jīng)完成（可能成功，也可能發(fā)生了錯誤）之后，才會調(diào)用這個函數(shù)。aio_return()的返回值就等價于同步情況中 read 或 write 系統(tǒng)調(diào)用的返回值（所傳輸?shù)淖止?jié)數(shù)，如果發(fā)生錯誤，返回值就為?1）。 

下面代碼清單給出了用戶空間應用程序進行異步讀操作的一個例程，它首先打開文件，然后準備 aiocb 結構體，之后調(diào)用 aio_read（&my_aiocb）進行提出異步讀請求，當 aio_error(&my_aiocb) = = EINPROGRESS 即操作還在進行中時，一直等待，結束后通過 aio_return（&my_aiocb）獲得返回值。

1  #include <aio.h> 
2  ... 
3  int fd, ret; 
4  struct aiocb my_aiocb; 
5   
6  fd = open("file.txt", O_RDONLY); 
7  if (fd < 0) 
8    perror("open"); 
9   
10 /* 清零 aiocb 結構體 */ 
11 bzero((char*) &my_aiocb, sizeof(struct aiocb)); 
12  
13 /* 為 aiocb 請求分配數(shù)據(jù)緩沖區(qū) */ 
14 my_aiocb.aio_buf = malloc(BUFSIZE + 1); 
15 if (!my_aiocb.aio_buf) 
16   perror("malloc"); 
17  
18 /* 初始化 aiocb 的成員 */ 
19 my_aiocb.aio_fildes = fd; 
20 my_aiocb.aio_nbytes = BUFSIZE; 
21 my_aiocb.aio_offset = 0; 
22  
23 ret = aio_read(&my_aiocb); 
24 if (ret < 0) 
25   perror("aio_read"); 
26  
27 while (aio_error(&my_aiocb) == EINPROGRESS) 
28   ; 
29  
30 if ((ret = aio_return(&my_iocb)) > 0) 
31 { 
32   /* 獲得異步讀的返回值 */ 
33 } 
34 else 
35 { 
36   /* 讀失敗，分析 errorno */ 
37 } 

int aio_suspend( const struct aiocb *const cblist[], int n, const struct timespec *timeout ); 

下面代碼清單給出了用戶空間異步讀操作時使用 aio_suspend()函數(shù)的例子。

1 struct aioct *cblist[MAX_LIST] 
2 /* 清零 aioct 結構體鏈表 */ 
3 bzero((char *)cblist, sizeof(cblist)); 
4 /* 將一個或更多的 aiocb 放入 aioct 結構體鏈表 */
5 cblist[0] = &my_aiocb; 
6 ret = aio_read( &my_aiocb ); 
7 ret = aio_suspend( cblist, MAX_LIST, NULL ); 

int aio_cancel( int fd, struct aiocb *aiocbp ); 

如果要取消一個請求，用戶需提供文件描述符和 aiocb 引用。如果這個請求被成功取消了，那么這個函數(shù)就會返回 AIO_CANCELED。如果請求完成了，這個函數(shù)就會返回AIO_NOTCANCELED。 如果要取消對某個給定文件描述符的所有請求，用戶需要提供這個文件的描述符以及一個對 aiocbp 的 NULL 引用。如果所有的請求都取消了，這個函數(shù)就會返回AIO_CANCELED ；如果至少有一個請求沒有被取消，那么這個函數(shù)就會返回AIO_NOT_CANCELED；如果沒有一個請求可以被取消，那么這個函數(shù)就會返回AIO_ALLDONE。然后，可以使用 aio_error()來驗證每個 AIO 請求，如果某請求已經(jīng)被取消了，那么 aio_error()就會返回?1，并且 errno 會被設置為 ECANCELED。 lio_listio()函數(shù)可用于同時發(fā)起多個傳輸。這個函數(shù)非常重要，它使得用戶可以在一個系統(tǒng)調(diào)用（一次內(nèi)核上下文切換）中啟動大量的 I/O 操作。lio_listio API 函數(shù)的原型如下： 

int lio_listio( int mode, struct aiocb *list[], int nent, struct sigevent *sig ); 

mode 參數(shù)可以是 LIO_WAIT 或 LIO_NOWAIT。LIO_WAIT 會阻塞這個調(diào)用，直到所有的 I/O 都完成為止。在操作進行排隊之后，LIO_NOWAIT 就會返回。list 是一個 aiocb 引用的列表，最大元素的個數(shù)是由 nent 定義的。如果 list 的元素為 NULL，lio_listio()會將其忽略。 

代碼清單給出了用戶空間異步 I/O 操作時使用 lio_listio()函數(shù)的例子。

1  struct aiocb aiocb1, aiocb2; 
2  struct aiocb *list[MAX_LIST]; 
3  ... 
4  /* 準備第一個 aiocb */ 
5  aiocb1.aio_fildes = fd; 
6  aiocb1.aio_buf = malloc( BUFSIZE+1 ); 
7  aiocb1.aio_nbytes = BUFSIZE; 
8  aiocb1.aio_offset = next_offset; 
9  aiocb1.aio_lio_opcode = LIO_READ; /*異步讀操作*/ 
10 ...  /*準備多個 aiocb */ 
11 bzero( (char *)list, sizeof(list) ); 
12  
13 /*將 aiocb 填入鏈表*/ 
14 list[0] = &aiocb1;
15 list[1] = &aiocb2; 
16 ... 
17 ret = lio_listio( LIO_WAIT, list, MAX_LIST, NULL );/*發(fā)起大量 I/O 操作*/ 

使用信號作為 AIO 的通知 

上面講述的信號作為異步通知的機制在 AIO 中仍然是適用的，為使用信號，使用 AIO 的應用程序同樣需要定義信號處理程序，在指定的信號被產(chǎn)生時會觸發(fā)調(diào)用這個處理程序。作為信號上下文的一部分，特定的 aiocb 請求被提供給信號處理函數(shù)用來區(qū)分 AIO 請求。 

下面代碼清單給出了使用信號作為 AIO 異步 I/O 通知機制的例子。

1  /*設置異步 I/O 請求*/ 
2  void setup_io(...) 
3  { 
4    int fd; 
5    struct sigaction sig_act; 
6    struct aiocb my_aiocb; 
7    ... 
8    /* 設置信號處理函數(shù) */ 
9    sigemptyset(&sig_act.sa_mask); 
10   sig_act.sa_flags = SA_SIGINFO; 
11   sig_act.sa_sigaction = aio_completion_handler; 
12  
13   /* 設置 AIO 請求 */ 
14   bzero((char*) &my_aiocb, sizeof(struct aiocb)); 
15   my_aiocb.aio_fildes = fd; 
16   my_aiocb.aio_buf = malloc(BUF_SIZE + 1); 
17   my_aiocb.aio_nbytes = BUF_SIZE; 
18   my_aiocb.aio_offset = next_offset; 
19  
20   /* 連接 AIO 請求和信號處理函數(shù) */ 
21   my_aiocb.aio_sigevent.sigev_notify = SIGEV_SIGNAL; 
22   my_aiocb.aio_sigevent.sigev_signo = SIGIO; 
23   my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb; 
24  
25   /* 將信號與信號處理函數(shù)綁定 */ 
26   ret = sigaction(SIGIO, &sig_act, NULL); 
27   ... 
28   ret = aio_read(&my_aiocb); /*發(fā)出異步讀請求*/ 
29 } 
30  
31 /*信號處理函數(shù)*/ 
32  void  aio_completion_handler(int  signo,  siginfo_t  *info,  void *context) 
33 { 
34   struct aiocb *req; 
35  
36   /* 確定是我們需要的信號*/ 
37   if (info->si_signo == SIGIO) 
38   { 
39     req = (struct aiocb*)info->si_value.sival_ptr; /*獲得 aiocb*/ 
40  
41     /* 請求的操作完成了嗎? */ 
42     if (aio_error(req) == 0) 
43     { 
44       /* 請求的操作完成，獲取返回值 */ 
45       ret = aio_return(req); 
46     } 
47   } 
48   return ; 
49 } 

使用回調(diào)函數(shù)作為 AIO 的通知 

除了信號之外，應用程序還可提供一個回調(diào)（Callback）函數(shù)給內(nèi)核，以便 AIO 的請求完成后內(nèi)核調(diào)用這個函數(shù)。 

一般來說，下層對上層（如內(nèi)核對應用）的調(diào)用都稱為“回調(diào)”，而上層對下層（如進行 Linux 系統(tǒng)調(diào)用）的調(diào)用稱為“調(diào)用”，如圖 9.3 所示。

代碼清單給出了使用回調(diào)函數(shù)作為 AIO 異步 I/O 請求完成的通知機制的例子。

1  /*設置異步 I/O 請求*/ 
2  void setup_io(...) 
3  { 
4    int fd; 
5    struct aiocb my_aiocb; 
6    ... 
7    /* 設置 AIO 請求 */ 
8    bzero((char*) &my_aiocb, sizeof(struct aiocb)); 
9    my_aiocb.aio_fildes = fd; 
10   my_aiocb.aio_buf = malloc(BUF_SIZE + 1); 
11   my_aiocb.aio_nbytes = BUF_SIZE; 
12   my_aiocb.aio_offset = next_offset; 
13  
14   /* 連接 AIO 請求和線程回調(diào)函數(shù) */ 
15   my_aiocb.aio_sigevent.sigev_notify = SIGEV_THREAD; 
16   my_aiocb.aio_sigevent.notify_function = aio_completion_handler; 
17     /*設置回調(diào)函數(shù)*/ 
18   my_aiocb.aio_sigevent.notify_attributes = NULL; 
19   my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb; 
20   ... ret = aio_read(&my_aiocb); //發(fā)起 AIO 請求 
21 } 
22  
23 /* 異步 I/O 完成回調(diào)函數(shù) */ 
24 void aio_completion_handler(sigval_t sigval) 
25 { 
26   struct aiocb *req; 
27   req = (struct aiocb*)sigval.sival_ptr; 
28  
29   /* AIO 請求完成? */ 
30   if (aio_error(req) == 0) 
31   { 
32     /* 請求完成，獲得返回值 */ 
33     ret = aio_return(req); 
34   } 
35  
36   return ; 
37 } 

proc 文件系統(tǒng)包含了兩個虛擬文件，它們可以用來對異步 I/O 的性能進行優(yōu)化。 

/proc/sys/fs/aio-nr 文件提供了系統(tǒng)范圍異步 I/O 請求的數(shù)目。 

/proc/sys/fs/aio-max-nr 文件是所允許的并發(fā)請求的最大個數(shù)，最大個數(shù)通常是 64KB，這對于大部分應用程序來說都已經(jīng)足夠了。 

AIO 與設備驅(qū)動 

在內(nèi)核中，每個 I/O 請求都對應于一個 kiocb 結構體，其 ki_filp 成員指向?qū)膄ile 指針，通過 is_sync_kiocb()可以判斷某 kiocb 是否為同步 I/O 請求，如果返回非真，表示為異步 I/O 請求。 

塊設備和網(wǎng)絡設備本身是異步的，只有字符設備必須明確表明應支持 AIO。AIO對于大多數(shù)字符設備而言都不是必須的，只有極少數(shù)設備需要。比如，對于磁帶機，由于 I/O 操作很慢，這時候使用異步 I/O 將改善性能。 字符設備驅(qū)動程序中file_operations 包含 3 個與 AIO 相關的成員函數(shù)，如下所示： 

ssize_t (*aio_read) (struct kiocb *iocb, char *buffer, size_t count, loff_t offset); 

ssize_t (*aio_write) (struct kiocb *iocb, const char *buffer, size_t count, loff_t offset); 

int (*aio_fsync) (struct kiocb *iocb, int datasync); 

aio_read()和 aio_write()與 file_operations 中的 read()和 write()中的 offset 參數(shù)不同，它直接傳遞值，而后者傳遞的是指針，這是因為 AIO 從來不需要改變文件的位置。 

aio_read()和 aio_write()函數(shù)本身不一定完成了讀和寫操作，它只是發(fā)起、初始化讀和寫操作，下面代碼清單給出了驅(qū)動程序中aio_read()和aio_write()函數(shù)的實現(xiàn)例子。

1  /*異步讀*/ 
2  static ssize_t xxx_aio_read(struct kiocb *iocb, char *buf, size_t count, loff_t 
3    pos) 
4  { 
5    return xxx_defer_op(0, iocb, buf, count, pos); 
6  } 
7   
8  /*異步寫*/ 
9  static ssize_t xxx_aio_write(struct kiocb *iocb, const char *buf, size_t count, 
10   loff_t pos) 
11 { 
12   return xxx_defer_op(1, iocb, (char*)buf, count, pos);
13 } 
14  
15 /*初始化異步 I/O*/ 
16 static int xxx_defer_op(int write, struct kiocb *iocb, char *buf, size_t count, 
17   loff_t pos) 
18 { 
19   struct async_work *async_wk; 
20   int result; 
21   /* 當可以訪問 buffer 時進行復制*/ 
22   if (write) 
23     result = xxx_write(iocb->ki_filp, buf, count, &pos); 
24   else 
25     result = xxx_read(iocb->ki_filp, buf, count, &pos); 
26   /* 如果是同步 IOCB，立即返回狀態(tài) */ 
27   if (is_sync_kiocb(iocb)) 
28     return result; 
29      
30   /* 否則，推后幾μ s 執(zhí)行 */ 
31   async_wk = kmalloc(sizeof(*async_wk), GFP_KERNEL); 
32   if (async_wk == NULL) 
33     return result; 
34   /*調(diào)度延遲的工作*/ 
35   async_wk->iocb = iocb; 
36   async_wk->result = result; 
37   INIT_WORK(&async_wk->work, xxx_do_deferred_op, async_wk); 
38   schedule_delayed_work(&async_wk->work, HZ / 100); 
39   return  - EIOCBQUEUED; /*控制權返回用戶空間*/ 
40 } 
41  
42 /*延遲后執(zhí)行*/ 
43 static void xxx_do_deferred_op(void *p) 
44 { 
45   struct async_work *async_wk = (struct async_work*)p; 
46   aio_complete(async_wk->iocb, async_wk->result, 0); 
47   kfree(async_wk); 
48 } 

1 struct async_work 
2 { 
3     struct kiocb *iocb;  //kiocb 結構體指針 
4     int result;  //執(zhí)行結果 
5     struct work_struct work; //工作結構體 
6 }; 

就到這里了，O(∩_∩)O~