2. linux AIO 異步讀寫

1.異步IO概念

在傳統的 I/O 模型中，有一個使用惟一句柄標識的 I/O 通道。在 UNIX 中，這些句柄是文件描述符（這對等同於文件、管道、套接字等等）。在阻塞 I/O 中，我們發起了一次傳輸操作，當傳輸操作完成或發生錯誤時，系統調用就會返回。

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

2.異步IO基本API

核心結構體

struct aiocb
{
    //要異步操作的文件描述符
    int aio_fildes;
    //用於lio操作時選擇操作何種異步I/O類型
    int aio_lio_opcode;
    //異步讀或寫的緩沖區的緩沖區
    volatile void *aio_buf;
    //異步讀或寫的字節數
    size_t aio_nbytes;
    //異步通知的結構體
    struct sigevent aio_sigevent;
}


struct sigevent
{
    sigval_t sigev_value;
    int sigev_signo;
    int sigev_notify;
    union {
        int _pad[SIGEV_PAD_SIZE];
         int _tid;


        struct {
            void (*_function)(sigval_t);
            void *_attribute;   /* really pthread_attr_t */
        } _sigev_thread;
    } _sigev_un;
}


#define sigev_notify_function   _sigev_un._sigev_thread._function
#define sigev_notify_attributes _sigev_un._sigev_thread._attribute
#define sigev_notify_thread_id   _sigev_un._tid

API	說明
aio_read	請求異步讀操作
aio_error	請求異步讀操作
aio_return	請求異步讀操作
aio_write	請求異步讀操作
aio_suspend	請求異步讀操作
aio_cancel	請求異步讀操作
aio_listio	請求異步讀操作

3. 異步讀取aio_read

/*
該函數請求對文件進行異步讀操作，若請求失敗返回-1，成功則返回0，並將該請求進行排隊，然后就開始對文件的異步讀操作需要注意的是，我們得先對aiocb結構體進行必要的初始化
*/
int aio_read(struct aiocb *paiocb);


/*
當其狀態處於EINPROGRESS則I/O還沒完成，當處於ECANCELLED則操作已被取消，發生錯誤返回-1
*/
int aio_error(struct aiocb *aiopcb);


//返回讀寫的字節數
//如果操作沒完成調用此函數，則會產生錯誤
ssize_t aio_return(struct aiocb *paiocb);

例程

#include<stdio.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<assert.h>
#include<unistd.h>
#include<stdlib.h>
#include<errno.h>
#include<string.h>
#include<sys/types.h>
#include<fcntl.h>
#include<aio.h>



#define BUFFER_SIZE 1024


int MAX_LIST = 2;


int main(int argc,char **argv)
{
    //aio操作所需結構體
    struct aiocb rd;


    int fd,ret,couter;


    fd = open("test.txt",O_RDONLY);
    if(fd < 0)
    {
        perror("test.txt");
    }




    //將rd結構體清空
    bzero(&rd,sizeof(rd));


    //為rd.aio_buf分配空間
    rd.aio_buf = malloc(BUFFER_SIZE + 1);


    //填充rd結構體
    rd.aio_fildes = fd;
    rd.aio_nbytes =  BUFFER_SIZE;
    rd.aio_offset = 0;


    //進行異步讀操作
    ret = aio_read(&rd);
    if(ret < 0)
    {
        perror("aio_read");
        exit(1);
    }


    couter = 0;
//  循環等待異步讀操作結束
    while(aio_error(&rd) == EINPROGRESS)
    {
        printf("第%d次\n",++couter);
    }
    //獲取異步讀返回值
    ret = aio_return(&rd);


    printf("\n\n返回值為:%d",ret);



    return 0;
}

注意：編譯上述程序時必須在編譯時再加一個-lrt

4.異步寫aio_write

/*
aio_write和aio_read函數類似，當該函數返回成功時，說明該寫請求以進行排隊(成功0，失敗-1)
其和aio_read調用時的區別是就是我們如果在打開文件是，flags設置了O_APPEND則我們在填充aiocb時不需要填充它的偏移量了
*/
int aio_write(struct aiocb *paiocb);

例程

#include<stdio.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<assert.h>
#include<unistd.h>
#include<stdlib.h>
#include<errno.h>
#include<string.h>
#include<sys/types.h>
#include<fcntl.h>
#include<aio.h>


#define BUFFER_SIZE 1025


int main(int argc,char **argv)
{
    //定義aio控制塊結構體
    struct aiocb wr;


    int ret,fd;


    char str[20] = {"hello,world"};


    //置零wr結構體
    bzero(&wr,sizeof(wr));


    fd = open("test.txt",O_WRONLY | O_APPEND);
    if(fd < 0)
    {
        perror("test.txt");
    }


    //為aio.buf申請空間
    wr.aio_buf = (char *)malloc(BUFFER_SIZE);
    if(wr.aio_buf == NULL)
    {
        perror("buf");
    }


    wr.aio_buf = str;


    //填充aiocb結構
    wr.aio_fildes = fd;
    wr.aio_nbytes = 1024;


    //異步寫操作
    ret = aio_write(&wr);
    if(ret < 0)
    {
        perror("aio_write");
    }


    //等待異步寫完成
    while(aio_error(&wr) == EINPROGRESS)
    {
        printf("hello,world\n");
    }


    //獲得異步寫的返回值
    ret = aio_return(&wr);
    printf("\n\n\n返回值為:%d\n",ret);


    return 0;
}

5. aio_suspend & aio_cancel

/*
函數來掛起（或阻塞）調用進程，直到異步請求完成為止，此時會產生一個信號，或者發生其他超時操作。調用者提供了一個aiocb引用列表，其中任何一個完成都會導致 aio_suspend 返回。
*/
int aio_suspend( const struct aiocb *const cblist[],
                  int n, const struct timespec *timeout );

/*
要取消對某個給定文件描述符的所有請求，我們需要提供這個文件的描述符，以及一個對 aiocbp 的 NULL 引用。如果所有的請求都取消了，這個函數就會返回 AIO_CANCELED；如果至少有一個請求沒有被取消，那么這個函數就會返回 AIO_NOT_CANCELED；如果沒有一個請求可以被取消，那么這個函數就會返回 AIO_ALLDONE。我們然后可以使用 aio_error 來驗證每個 AIO 請求。如果這個請求已經被取消了，那么 aio_error 就會返回 -1，並且 errno 會被設置為 ECANCELED。
*/
int aio_cancel( int fd, struct aiocb *aiocbp );

6. lio_listio

/*
這個函數非常重要，因為這意味着我們可以在一個系統調用（一次內核上下文切換）中啟動大量的 I/O 操作.ode 參數可以是 LIO_WAIT 或 LIO_NOWAIT。LIO_WAIT 會阻塞這個調用，直到所有的 I/O 都完成為止。在操作進行排隊之后，LIO_NOWAIT 就會返回。list 是一個 aiocb 引用的列表，最大元素的個數是由 nent 定義的。注意 list 的元素可以為 NULL，lio_listio 會將其忽略。sigevent 引用定義了在所有 I/O 操作都完成時產生信號的方法。


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

例程

#include<stdio.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<assert.h>
#include<unistd.h>
#include<stdlib.h>
#include<errno.h>
#include<string.h>
#include<sys/types.h>
#include<fcntl.h>
#include<aio.h>


#define BUFFER_SIZE 1025


int MAX_LIST = 2;



int main(int argc,char **argv)
{
    struct aiocb *listio[2];
    struct aiocb rd,wr;
    int fd,ret;


    //異步讀事件
    fd = open("test1.txt",O_RDONLY);
    if(fd < 0)
    {
        perror("test1.txt");
    }


    bzero(&rd,sizeof(rd));


    rd.aio_buf = (char *)malloc(BUFFER_SIZE);
    if(rd.aio_buf == NULL)
    {
        perror("aio_buf");
    }


    rd.aio_fildes = fd;
    rd.aio_nbytes = 1024;
    rd.aio_offset = 0;
    rd.aio_lio_opcode = LIO_READ;   ///lio操作類型為異步讀


    //將異步讀事件添加到list中
    listio[0] = &rd;



    //異步些事件
    fd = open("test2.txt",O_WRONLY | O_APPEND);
    if(fd < 0)
    {
        perror("test2.txt");
    }


    bzero(&wr,sizeof(wr));


    wr.aio_buf = (char *)malloc(BUFFER_SIZE);
    if(wr.aio_buf == NULL)
    {
        perror("aio_buf");
    }


    wr.aio_fildes = fd;
    wr.aio_nbytes = 1024;


    wr.aio_lio_opcode = LIO_WRITE;   ///lio操作類型為異步寫


    //將異步寫事件添加到list中
    listio[1] = &wr;


    //使用lio_listio發起一系列請求
    ret = lio_listio(LIO_WAIT,listio,MAX_LIST,NULL);


    //當異步讀寫都完成時獲取他們的返回值


    ret = aio_return(&rd);
    printf("\n讀返回值:%d",ret);


    ret = aio_return(&wr);
    printf("\n寫返回值:%d",ret);




    return 0;
}

7. IO 完成時異步通知

/*
使用回調進行異步通知,該種通知方式使用一個系統回調函數來通知應用程序，要想完成此功能，我們必須在aiocb中設置我們想要進行異步回調的aiocb指針，以用來回調之后表示其自身
*/

例程

#include<stdio.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<assert.h>
#include<unistd.h>
#include<stdlib.h>
#include<errno.h>
#include<string.h>
#include<sys/types.h>
#include<fcntl.h>
#include<aio.h>
#include<unistd.h>


#define BUFFER_SIZE 1025



void aio_completion_handler(sigval_t sigval)
{
    //用來獲取讀aiocb結構的指針
    struct aiocb *prd;
    int ret;


    prd = (struct aiocb *)sigval.sival_ptr;


    printf("hello\n");


    //判斷請求是否成功
    if(aio_error(prd) == 0)
    {
        //獲取返回值
        ret = aio_return(prd);
        printf("讀返回值為:%d\n",ret);
    }
}


int main(int argc,char **argv)
{
    int fd,ret;
    struct aiocb rd;


    fd = open("test.txt",O_RDONLY);
    if(fd < 0)
    {
        perror("test.txt");
    }




    //填充aiocb的基本內容
    bzero(&rd,sizeof(rd));


    rd.aio_fildes = fd;
    rd.aio_buf = (char *)malloc(sizeof(BUFFER_SIZE + 1));
    rd.aio_nbytes = BUFFER_SIZE;
    rd.aio_offset = 0;


    //填充aiocb中有關回調通知的結構體sigevent
    rd.aio_sigevent.sigev_notify = SIGEV_THREAD;//使用線程回調通知
    rd.aio_sigevent.sigev_notify_function = aio_completion_handler;//設置回調函數
    rd.aio_sigevent.sigev_notify_attributes = NULL;//使用默認屬性
    rd.aio_sigevent.sigev_value.sival_ptr = &rd;//在aiocb控制塊中加入自己的引用


    //異步讀取文件
    ret = aio_read(&rd);
    if(ret < 0)
    {
        perror("aio_read");
    }


    printf("異步讀以開始\n");
    sleep(1);
    printf("異步讀結束\n");




    return 0;
}