成于堅持，敗于止步
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)生交疊。可移植操作系統(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 } 用戶可以使用 aio_suspend()函數(shù)來掛起（或阻塞）調(diào)用進程，直到異步請求完成為止，此時會產(chǎn)生一個信號，或者發(fā)生其他超時操作。調(diào)用者提供了一個 aiocb 引用列表，其中任何一個完成都會導致 aio_suspend()返回。aio_suspend 的函數(shù)原型如下： 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 );aio_cancel()函數(shù)允許用戶取消對某個文件描述符執(zhí)行的一個或所有 I/O 請求。其原型如下：
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 操作*/上述代碼第 9 行中，因為是進行異步讀操作，所以操作碼為 LIO_READ，對于寫操作來說，應該使用 LIO_WRITE 作為操作碼，而 LIO_NOP 意味著空操作。
使用信號作為 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 } 特別要注意上述代碼的第 39 行通過(struct aiocb*)info->si_value.sival_ptr 獲得了信號對應的 aiocb。 使用回調(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 } 上述程序在創(chuàng)建 aiocb 請求之后，使用 SIGEV_THREAD 請求了一個線程回調(diào)函數(shù)來作為通知方法。在回調(diào)函數(shù)中，通過(struct aiocb*)sigval.sival_ptr 可以獲得對應的aiocb 指針，使用 AIO 函數(shù)可驗證請求是否已經(jīng)完成。 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 } 上述代碼中最核心的是使用 aync_work（異步工作）結構體將操作延后執(zhí)行，aync_work 結構體定義如代碼清單所示，通過 schedule_delayed_work()函數(shù)可以調(diào)度其執(zhí)行。第 46 行對 aio_complete()的調(diào)用用于通知內(nèi)核驅(qū)動程序已經(jīng)完成了操作。1 struct async_work 2 { 3 struct kiocb *iocb; //kiocb 結構體指針 4 int result; //執(zhí)行結果 5 struct work_struct work; //工作結構體 6 }; 就到這里了，O(∩_∩)O~
