AIO 概念與 GNU C 庫函數

Linux 系統中最常用的輸入/輸出（I/O）模型是同步 I/O。在這個模型中，當請求發出之後，應用程序就會阻塞，直到請求滿足為止。這是很好的一種解決方案，因為調用應用程序在等待 I/O 請求完成時不需要使用任何中央處理單元（CPU）。但是在某些情況下，I/O 請求可能需要與其他進程產生交疊。可移植操作系統接口（POSIX）異步 I/O（AIO）應用程序接口（API）就提供了這種功能。

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

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

在異步非阻塞 I/O 中，我們可以同時發起多個傳輸操作。這需要每個傳輸操作都有惟一的上下文，這樣才能在它們完成時區分到底是哪個傳輸操作完成了。在 AIO 中，通過 aiocb（AIO I/O Control Block）結構體進行區分。這個結構體包含了有關傳輸的所有信息，包括為數據准備的用戶緩沖區。在產生 I/O（稱為完成）通知時，aiocb 結構就被用來惟一標識所完成的 I/O 操作。

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

1．aio_read

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

aio_read 函數的原型如下：

int aio_read( struct aiocb *aiocbp );

aio_read()函數在請求進行排隊之後會立即返回。如果執行成功，返回值就為 0；如果出現錯誤，返回值就為?1，並設置 errno 的值。

2．aio_write

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

int aio_write( struct aiocb *aiocbp );

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

3．aio_error

aio_error 函數被用來確定請求的狀態，其原型如下：

int aio_error( struct aiocb *aiocbp );

這個函數可以返回以下內容。

EINPROGRESS：說明請求尚未完成。

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

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

4．aio_return

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

ssize_t aio_return( struct aiocb *aiocbp );

只有在 aio_error()調用確定請求已經完成（可能成功，也可能發生了錯誤）之後，才會調用這個函數。aio_return()的返回值就等價於同步情況中 read 或 write 系統調用的返回值（所傳輸的字節數，如果發生錯誤，返回值就為?1）。

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

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

用戶可以使用 aio_suspend()函數來掛起（或阻塞）調用進程，直到異步請求完成為止，此時會產生一個信號，或者發生其他超時操作。調用者提供了一個 aiocb 引用列表，其中任何一個完成都會導致 aio_suspend()返回。aio_suspend 的函數原型如下：

1structaioct*cblist[MAX_LIST]
2/*清零aioct結構體鏈表*/
3bzero((char*)cblist,sizeof(cblist));
4/*將一個或更多的aiocb放入aioct結構體鏈表*/
5cblist[0]=&my_aiocb;
6ret=aio_read(&my_aiocb);
7ret=aio_suspend(cblist,MAX_LIST,NULL); aio_cancel()函數允許用戶取消對某個文件描述符執行的一個或所有 I/O 請求。其原型如下：

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

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

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

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

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

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

上述代碼第 9 行中，因為是進行異步讀操作，所以操作碼為 LIO_READ，對於寫操作來說，應該使用 LIO_WRITE 作為操作碼，而 LIO_NOP 意味著空操作。

使用信號作為 AIO 的通知

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

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

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

特別要注意上述代碼的第 39 行通過(struct aiocb*)info->si_value.sival_ptr 獲得了信號對應的 aiocb。

使用回調函數作為 AIO 的通知

除了信號之外，應用程序還可提供一個回調（Callback）函數給內核，以便 AIO 的請求完成後內核調用這個函數。

一般來說，下層對上層（如內核對應用）的調用都稱為“回調”，而上層對下層（如進行 Linux 系統調用）的調用稱為“調用”，如圖 9.3 所示。

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

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

上述程序在創建 aiocb 請求之後，使用 SIGEV_THREAD 請求了一個線程回調函數來作為通知方法。在回調函數中，通過(struct aiocb*)sigval.sival_ptr 可以獲得對應的aiocb 指針，使用 AIO 函數可驗證請求是否已經完成。

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

/proc/sys/fs/aio-nr 文件提供了系統范圍異步 I/O 請求的數目。

/proc/sys/fs/aio-max-nr 文件是所允許的並發請求的最大個數，最大個數通常是 64KB，這對於大部分應用程序來說都已經足夠了。

AIO 與設備驅動

在內核中，每個 I/O 請求都對應於一個 kiocb 結構體，其 ki_filp 成員指向對應的file 指針，通過 is_sync_kiocb()可以判斷某 kiocb 是否為同步 I/O 請求，如果返回非真，表示為異步 I/O 請求。

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

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 參數不同，它直接傳遞值，而後者傳遞的是指針，這是因為 AIO 從來不需要改變文件的位置。

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

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

上述代碼中最核心的是使用 aync_work（異步工作）結構體將操作延後執行，aync_work 結構體定義如代碼清單所示，通過 schedule_delayed_work()函數可以調度其執行。第 46 行對 aio_complete()的調用用於通知內核驅動程序已經完成了操作。

1. 1structasync_work
2. 2{
3. 3structkiocb*iocb;//kiocb結構體指針
4. 4intresult;//執行結果
5. 5structwork_structwork;//工作結構體
6. 6};