級別： 中級
M. Tim Jones ( mtj@mtjones.com ), 顧問工程師, Emulex
2006 年 9 月 28 日
Linux? 中最常用的輸入/輸出（I/O）模型是同步 I/O。在這個模型中，當請求發出之后，應用程序就會阻塞，直到請求滿足為止。這是很好的一種解決方案，因為調用應用程序在等待 I/O 請求完成時不需要使用任何中央處理單元（CPU）。但是在某些情況中，I/O 請求可能需要與其他進程產生交疊。可移植操作系統接口（POSIX）異步 I/O（AIO）應用程序接口（API）就提供了這種功能。在本文中，我們將對這個 API 概要進行介紹，并來了解一下如何使用它。
Linux 異步 I/O 是 Linux 內核中提供的一個相當新的增強。它是 2.6 版本內核的一個標準特性，但是我們在 2.4 版本內核的補丁中也可以找到它。AIO 背后的基本思想是允許進程發起很多 I/O 操作，而不用阻塞或等待任何操作完成。稍后或在接收到 I/O 操作完成的通知時，進程就可以檢索 I/O 操作的結果。
在深入介紹 AIO API 之前，讓我們先來探索一下 Linux 上可以使用的不同 I/O 模型。這并不是一個詳盡的介紹，但是我們將試圖介紹最常用的一些模型來解釋它們與異步 I/O 之間的區別。圖 1 給出了同步和異步模型，以及阻塞和非阻塞的模型。
每個 I/O 模型都有自己的使用模式，它們對于特定的應用程序都有自己的優點。本節將簡要對其一一進行介紹。
|
|
最常用的一個模型是同步阻塞 I/O 模型。在這個模型中，用戶空間的應用程序執行一個系統調用，這會導致應用程序阻塞。這意味著應用程序會一直阻塞，直到系統調用完成為止（數據傳輸完成或發生錯誤）。調用應用程序處于一種不再消費 CPU 而只是簡單等待響應的狀態，因此從處理的角度來看，這是非常有效的。
圖 2 給出了傳統的阻塞 I/O 模型，這也是目前應用程序中最為常用的一種模型。其行為非常容易理解，其用法對于典型的應用程序來說都非常有效。在調用
read
系統調用時，應用程序會阻塞并對內核進行上下文切換。然后會觸發讀操作，當響應返回時（從我們正在從中讀取的設備中返回），數據就被移動到用戶空間的緩沖區中。然后應用程序就會解除阻塞（
read
調用返回）。
從應用程序的角度來說，
read
調用會延續很長時間。實際上，在內核執行讀操作和其他工作時，應用程序的確會被阻塞。
同步阻塞 I/O 的一種效率稍低的變種是同步非阻塞 I/O。在這種模型中，設備是以非阻塞的形式打開的。這意味著 I/O 操作不會立即完成，
read
操作可能會返回一個錯誤代碼，說明這個命令不能立即滿足（
EAGAIN
或
EWOULDBLOCK
），如圖 3 所示。
非阻塞的實現是 I/O 命令可能并不會立即滿足，需要應用程序調用許多次來等待操作完成。這可能效率不高，因為在很多情況下，當內核執行這個命令時，應用程序必須要進行忙碌等待，直到數據可用為止，或者試圖執行其他工作。正如圖 3 所示的一樣，這個方法可以引入 I/O 操作的延時，因為數據在內核中變為可用到用戶調用
read
返回數據之間存在一定的間隔，這會導致整體數據吞吐量的降低。
另外一個阻塞解決方案是帶有阻塞通知的非阻塞 I/O。在這種模型中，配置的是非阻塞 I/O，然后使用阻塞
select
系統調用來確定一個 I/O 描述符何時有操作。使
select
調用非常有趣的是它可以用來為多個描述符提供通知，而不僅僅為一個描述符提供通知。對于每個提示符來說，我們可以請求這個描述符可以寫數據、有讀數據可用以及是否發生錯誤的通知。
圖 4. 異步阻塞 I/O 模型的典型流程 (select)
select
調用的主要問題是它的效率不是非常高。盡管這是異步通知使用的一種方便模型，但是對于高性能的 I/O 操作來說不建議使用。
最后，異步非阻塞 I/O 模型是一種處理與 I/O 重疊進行的模型。讀請求會立即返回，說明
read
請求已經成功發起了。在后臺完成讀操作時，應用程序然后會執行其他處理操作。當
read
的響應到達時，就會產生一個信號或執行一個基于線程的回調函數來完成這次 I/O 處理過程。
在一個進程中為了執行多個 I/O 請求而對計算操作和 I/O 處理進行重疊處理的能力利用了處理速度與 I/O 速度之間的差異。當一個或多個 I/O 請求掛起時，CPU 可以執行其他任務；或者更為常見的是，在發起其他 I/O 的同時對已經完成的 I/O 進行操作。
下一節將深入介紹這種模型，探索這種模型使用的 API，然后展示幾個命令。
|
|
從前面 I/O 模型的分類中，我們可以看出 AIO 的動機。這種阻塞模型需要在 I/O 操作開始時阻塞應用程序。這意味著不可能同時重疊進行處理和 I/O 操作。同步非阻塞模型允許處理和 I/O 操作重疊進行，但是這需要應用程序根據重現的規則來檢查 I/O 操作的狀態。這樣就剩下異步非阻塞 I/O 了，它允許處理和 I/O 操作重疊進行，包括 I/O 操作完成的通知。
除了需要阻塞之外，
select
函數所提供的功能（異步阻塞 I/O）與 AIO 類似。不過，它是對通知事件進行阻塞，而不是對 I/O 調用進行阻塞。
|
|
|
本節將探索 Linux 的異步 I/O 模型，從而幫助我們理解如何在應用程序中使用這種技術。
在傳統的 I/O 模型中，有一個使用惟一句柄標識的 I/O 通道。在 UNIX? 中，這些句柄是文件描述符（這對等同于文件、管道、套接字等等）。在阻塞 I/O 中，我們發起了一次傳輸操作，當傳輸操作完成或發生錯誤時，系統調用就會返回。
|
|
|
在異步非阻塞 I/O 中，我們可以同時發起多個傳輸操作。這需要每個傳輸操作都有惟一的上下文，這樣我們才能在它們完成時區分到底是哪個傳輸操作完成了。在 AIO 中，這是一個
aiocb
（AIO I/O Control Block）結構。這個結構包含了有關傳輸的所有信息，包括為數據準備的用戶緩沖區。在產生 I/O （稱為完成）通知時，
aiocb
結構就被用來惟一標識所完成的 I/O 操作。這個 API 的展示顯示了如何使用它。
|
|
|
AIO 接口的 API 非常簡單，但是它為數據傳輸提供了必需的功能，并給出了兩個不同的通知模型。表 1 給出了 AIO 的接口函數，本節稍后會更詳細進行介紹。
API 函數 說明
aio_read
|
請求異步讀操作 |
aio_error
|
檢查異步請求的狀態 |
aio_return
|
獲得完成的異步請求的返回狀態 |
aio_write
|
請求異步寫操作 |
aio_suspend
|
掛起調用進程，直到一個或多個異步請求已經完成（或失敗） |
aio_cancel
|
取消異步 I/O 請求 |
lio_listio
|
發起一系列 I/O 操作 |
每個 API 函數都使用
aiocb
結構開始或檢查。這個結構有很多元素，但是清單 1 僅僅給出了需要（或可以）使用的元素。
struct aiocb {
int aio_fildes; // File Descriptor
int aio_lio_opcode; // Valid only for lio_listio (r/w/nop)
volatile void *aio_buf; // Data Buffer
size_t aio_nbytes;// Number of Bytes in Data Buffer
struct sigevent aio_sigevent; // Notification Structure
/* Internal fields */
...
};
|
sigevent
結構告訴 AIO 在 I/O 操作完成時應該執行什么操作。我們將在 AIO 的展示中對這個結構進行探索。現在我們將展示各個 AIO 的 API 函數是如何工作的，以及我們應該如何使用它們。
aio_read
函數請求對一個有效的文件描述符進行異步讀操作。這個文件描述符可以表示一個文件、套接字甚至管道。
aio_read
函數的原型如下：
int
aio_read
( struct aiocb *aiocbp );
|
aio_read
函數在請求進行排隊之后會立即返回。如果執行成功，返回值就為 0；如果出現錯誤，返回值就為 -1，并設置
errno
的值。
要執行讀操作，應用程序必須對
aiocb
結構進行初始化。下面這個簡短的例子就展示了如何填充
aiocb
請求結構，并使用
aio_read
來執行異步讀請求（現在暫時忽略通知）操作。它還展示了
aio_error
的用法，不過我們將稍后再作解釋。
#include <aio.h>
...
int fd, ret;
struct aiocb my_aiocb;
fd = open( "file.txt", O_RDONLY );
if (fd < 0) perror("open");
/* Zero out the aiocb structure (recommended) */
bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
/* Allocate a data buffer for the aiocb request */
my_aiocb.aio_buf = malloc(BUFSIZE+1);
if (!my_aiocb.aio_buf) perror("malloc");
/* Initialize the necessary fields in the aiocb */
my_aiocb.aio_fildes = fd;
my_aiocb.aio_nbytes = BUFSIZE;
my_aiocb.aio_offset = 0;
ret =
aio_read
( &my_aiocb );
if (ret < 0) perror("aio_read");
while (
aio_error
( &my_aiocb ) == EINPROGRESS ) ;
if ((ret = aio_return( &my_iocb )) > 0) {
/* got ret bytes on the read */
} else {
/* read failed, consult errno */
}
|
在清單 2 中，在打開要從中讀取數據的文件之后，我們就清空了
aiocb
結構，然后分配一個數據緩沖區。并將對這個數據緩沖區的引用放到
aio_buf
中。然后，我們將
aio_nbytes
初始化成緩沖區的大小。并將
aio_offset
設置成 0（該文件中的第一個偏移量）。我們將
aio_fildes
設置為從中讀取數據的文件描述符。在設置這些域之后，就調用
aio_read
請求進行讀操作。我們然后可以調用
aio_error
來確定
aio_read
的狀態。只要狀態是
EINPROGRESS
，就一直忙碌等待，直到狀態發生變化為止。現在，請求可能成功，也可能失敗。
|
|
|
注意使用這個 API 與標準的庫函數從文件中讀取內容是非常相似的。除了
aio_read
的一些異步特性之外，另外一個區別是讀操作偏移量的設置。在傳統的
read
調用中，偏移量是在文件描述符上下文中進行維護的。對于每個讀操作來說，偏移量都需要進行更新，這樣后續的讀操作才能對下一塊數據進行尋址。對于異步 I/O 操作來說這是不可能的，因為我們可以同時執行很多讀請求，因此必須為每個特定的讀請求都指定偏移量。
aio_error
函數被用來確定請求的狀態。其原型如下：
int
aio_error
( struct aiocb *aiocbp );
|
這個函數可以返回以下內容：
-
EINPROGRESS，說明請求尚未完成 -
ECANCELLED，說明請求被應用程序取消了 -
-1，說明發生了錯誤，具體錯誤原因可以查閱errno
異步 I/O 和標準塊 I/O 之間的另外一個區別是我們不能立即訪問這個函數的返回狀態，因為我們并沒有阻塞在
read
調用上。在標準的
read
調用中，返回狀態是在該函數返回時提供的。但是在異步 I/O 中，我們要使用
aio_return
函數。這個函數的原型如下：
ssize_t
aio_return
( struct aiocb *aiocbp );
|
只有在
aio_error
調用確定請求已經完成（可能成功，也可能發生了錯誤）之后，才會調用這個函數。
aio_return
的返回值就等價于同步情況中
read
或
write
系統調用的返回值（所傳輸的字節數，如果發生錯誤，返回值就為
-1
）。
aio_write
函數用來請求一個異步寫操作。其函數原型如下：
int
aio_write
( struct aiocb *aiocbp );
|
aio_write
函數會立即返回，說明請求已經進行排隊（成功時返回值為
0
，失敗時返回值為
-1
，并相應地設置
errno
）。
這與
read
系統調用類似，但是有一點不一樣的行為需要注意。回想一下對于
read
調用來說，要使用的偏移量是非常重要的。然而，對于
write
來說，這個偏移量只有在沒有設置
O_APPEND
選項的文件上下文中才會非常重要。如果設置了
O_APPEND
，那么這個偏移量就會被忽略，數據都會被附加到文件的末尾。否則，
aio_offset
域就確定了數據在要寫入的文件中的偏移量。
我們可以使用
aio_suspend
函數來掛起（或阻塞）調用進程，直到異步請求完成為止，此時會產生一個信號，或者發生其他超時操作。調用者提供了一個
aiocb
引用列表，其中任何一個完成都會導致
aio_suspend
返回。
aio_suspend
的函數原型如下：
int
aio_suspend
( const struct aiocb *const cblist[],
int n, const struct timespec *timeout );
|
aio_suspend
的使用非常簡單。我們要提供一個
aiocb
引用列表。如果任何一個完成了，這個調用就會返回
0
。否則就會返回
-1
，說明發生了錯誤。請參看清單 3。
清單 3. 使用 aio_suspend 函數阻塞異步 I/O
struct aioct *cblist[MAX_LIST]
/* Clear the list. */
bzero( (char *)cblist, sizeof(cblist) );
/* Load one or more references into the list */
cblist[0] = &my_aiocb;
ret =
aio_read
( &my_aiocb );
ret =
aio_suspend
( cblist, MAX_LIST, NULL );
|
注意，
aio_suspend
的第二個參數是
cblist
中元素的個數，而不是
aiocb
引用的個數。
cblist
中任何
NULL
元素都會被
aio_suspend
忽略。
如果為
aio_suspend
提供了超時，而超時情況的確發生了，那么它就會返回
-1
，
errno
中會包含
EAGAIN
。
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
。
最后，AIO 提供了一種方法使用
lio_listio
API 函數同時發起多個傳輸。這個函數非常重要，因為這意味著我們可以在一個系統調用（一次內核上下文切換）中啟動大量的 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
會將其忽略。
sigevent
引用定義了在所有 I/O 操作都完成時產生信號的方法。
對于
lio_listio
的請求與傳統的
read
或
write
請求在必須指定的操作方面稍有不同，如清單 4 所示。
struct aiocb aiocb1, aiocb2;
struct aiocb *list[MAX_LIST];
...
/* Prepare the first aiocb */
aiocb1.aio_fildes = fd;
aiocb1.aio_buf = malloc( BUFSIZE+1 );
aiocb1.aio_nbytes = BUFSIZE;
aiocb1.aio_offset = next_offset;
aiocb1.aio_lio_opcode = LIO_READ;
...
bzero( (char *)list, sizeof(list) );
list[0] = &aiocb1;
list[1] = &aiocb2;
ret =
lio_listio
( LIO_WAIT, list, MAX_LIST, NULL );
|
對于讀操作來說，
aio_lio_opcode
域的值為
LIO_READ
。對于寫操作來說，我們要使用
LIO_WRITE
，不過
LIO_NOP
對于不執行操作來說也是有效的。
|
|
|
現在我們已經看過了可用的 AIO 函數，本節將深入介紹對異步通知可以使用的方法。我們將通過信號和函數回調來探索異步函數的通知機制。
使用信號進行進程間通信（IPC）是 UNIX 中的一種傳統機制，AIO 也可以支持這種機制。在這種范例中，應用程序需要定義信號處理程序，在產生指定的信號時就會調用這個處理程序。應用程序然后配置一個異步請求將在請求完成時產生一個信號。作為信號上下文的一部分，特定的
aiocb
請求被提供用來記錄多個可能會出現的請求。清單 5 展示了這種通知方法。
void setup_io( ... )
{
int fd;
struct sigaction sig_act;
struct aiocb my_aiocb;
...
/* Set up the signal handler */
sigemptyset(&sig_act.sa_mask);
sig_act.sa_flags = SA_SIGINFO;
sig_act.sa_sigaction = aio_completion_handler;
/* Set up the AIO request */
bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
my_aiocb.aio_fildes = fd;
my_aiocb.aio_buf = malloc(BUF_SIZE+1);
my_aiocb.aio_nbytes = BUF_SIZE;
my_aiocb.aio_offset = next_offset;
/* Link the AIO request with the Signal Handler */
my_aiocb.aio_sigevent.sigev_notify = SIGEV_SIGNAL;
my_aiocb.aio_sigevent.sigev_signo = SIGIO;
my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;
/* Map the Signal to the Signal Handler */
ret = sigaction( SIGIO, &sig_act, NULL );
...
ret =
aio_read
( &my_aiocb );
}
void aio_completion_handler( int signo, siginfo_t *info, void *context )
{
struct aiocb *req;
/* Ensure it's our signal */
if (info->si_signo == SIGIO) {
req = (struct aiocb *)info->si_value.sival_ptr;
/* Did the request complete? */
if (
aio_error
( req ) == 0) {
/* Request completed successfully, get the return status */
ret =
aio_return
( req );
}
}
return;
}
|
在清單 5 中，我們在
aio_completion_handler
函數中設置信號處理程序來捕獲
SIGIO
信號。然后初始化
aio_sigevent
結構產生
SIGIO
信號來進行通知（這是通過
sigev_notify
中的
SIGEV_SIGNAL
定義來指定的）。當讀操作完成時，信號處理程序就從該信號的
si_value
結構中提取出
aiocb
，并檢查錯誤狀態和返回狀態來確定 I/O 操作是否完成。
對于性能來說，這個處理程序也是通過請求下一次異步傳輸而繼續進行 I/O 操作的理想地方。采用這種方式，在一次數據傳輸完成時，我們就可以立即開始下一次數據傳輸操作。
另外一種通知方式是系統回調函數。這種機制不會為通知而產生一個信號，而是會調用用戶空間的一個函數來實現通知功能。我們在
sigevent
結構中設置了對
aiocb
的引用，從而可以惟一標識正在完成的特定請求。請參看清單 6。
void setup_io( ... )
{
int fd;
struct aiocb my_aiocb;
...
/* Set up the AIO request */
bzero( (char *)&my_aiocb, sizeof(struct aiocb) );
my_aiocb.aio_fildes = fd;
my_aiocb.aio_buf = malloc(BUF_SIZE+1);
my_aiocb.aio_nbytes = BUF_SIZE;
my_aiocb.aio_offset = next_offset;
/* Link the AIO request with a thread callback */
my_aiocb.aio_sigevent.sigev_notify = SIGEV_THREAD;
my_aiocb.aio_sigevent.notify_function = aio_completion_handler;
my_aiocb.aio_sigevent.notify_attributes = NULL;
my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;
...
ret =
aio_read
( &my_aiocb );
}
void aio_completion_handler( sigval_t sigval )
{
struct aiocb *req;
req = (struct aiocb *)sigval.sival_ptr;
/* Did the request complete? */
if (
aio_error
( req ) == 0) {
/* Request completed successfully, get the return status */
ret =
aio_return
( req );
}
return;
}
|
在清單 6 中，在創建自己的
aiocb
請求之后，我們使用
SIGEV_THREAD
請求了一個線程回調函數來作為通知方法。然后我們將指定特定的通知處理程序，并將要傳輸的上下文加載到處理程序中（在這種情況中，是個對
aiocb
請求自己的引用）。在這個處理程序中，我們簡單地引用到達的
sigval
指針并使用 AIO 函數來驗證請求已經完成。
|
|
|
proc 文件系統包含了兩個虛擬文件，它們可以用來對異步 I/O 的性能進行優化：
- /proc/sys/fs/aio-nr 文件提供了系統范圍異步 I/O 請求現在的數目。
- /proc/sys/fs/aio-max-nr 文件是所允許的并發請求的最大個數。最大個數通常是 64KB，這對于大部分應用程序來說都已經足夠了。
|
|
|
使用異步 I/O 可以幫助我們構建 I/O 速度更快、效率更高的應用程序。如果我們的應用程序可以對處理和 I/O 操作重疊進行，那么 AIO 就可以幫助我們構建可以更高效地使用可用 CPU 資源的應用程序。盡管這種 I/O 模型與在大部分 Linux 應用程序中使用的傳統阻塞模式都不同，但是異步通知模型在概念上來說卻非常簡單，可以簡化我們的設計。
學習
-
您可以參閱本文在 developerWorks 全球站點上的
英文原文
。
-
POSIX.1b 實現
從 GNU Library 的角度介紹了 AIO 的詳細內幕。
-
Realtime Support in Linux
解釋了更多有關 AIO 和很多實時擴展的信息，內容從調度、POSIX I/O 到 POSIX 線程和高分辨率的定時器（HRT）。
-
在為 2.5 版本內核集成而編寫的
Design Notes
中，我們可以學習有關 Linux 中 AIO 的設計和實現的知識。
-
在
developerWorks Linux 專區
中可以找到為 Linux 開發人員準備的更多資源。
- 隨時關注 developerWorks 技術事件和網絡廣播 。
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
