Linux基础(09)aio高级编程

1.出于安全性 Linux有一个机制 应用层和内核层是无法互相直接读取内存的, 他们要互相读取数据是有一个拷贝过程的, 

　　如: 应用层要读取内核层的数据就调用read(), 内核就会先把数据copy到一个buff并返回给read()

　但, 如果这个过程很大很频繁那么同步机制的效率就非常低, 因为不读完主线程就会一直阻塞,会耽误接下来的操作,导致服务器延迟高,效率低

　所以出现了异步IO这个概念:

　　如:应用层要读取一段数据, 那么发起申请后,内核层开始进行拷贝无论是否完成都会立刻返回, 但内核还会继续拷贝操作直到完成而应用层的主线程则继续往下操作.

　而如何获知内核是否完成操作有两个方式　1.应用层主动发起询问,如果内核层完成返回一个完成状态,反之返回一个未完成状态 ,但是时机很难把握

　　　　　　　　　　　　　　　　　　　　2.应用层挂接(挂接后内核也是立刻返回的)一个回调函数callback

　　　　　(callback不是内核执行的, 是系统开一个线程执行的, 不会占用主线程的时间), 应用层继续其他操作 ,等到内核拷贝完后会自动调用callback

---

2.aio异步机制概述

　　aio是一种异步非阻塞IO , 例如:读取一个流媒体视频时,不可能一直阻塞在读写操作上,  所以有了aio,让aio继续读取或其他操作, 而cpu则处理其他操作

　　同步阻塞IO模型

　　 2.1 aio异步读写是在Linux内核2.6之后才正式纳入其标准。之所以会增加此模块，是因为众所周知我们计算机CPU的执行速度远大于I/O读写的执行速度，如果我们用传统的阻塞式或非阻塞式来操作I/O的话，那么我们在同一个程序中(不用多线程或多进程)就不能同时操作俩个以上的文件I/O，每次只能对一个文件进行I/O操作，很明显这样效率很低下(因为CPU速度远大于I/O操作的速度，所以当执行I/O时，CPU其实还可以做更多的事)。因此就诞生了相对高效的异步I/O

　　同步非阻塞 I/O

　　2.2 同步阻塞 I/O 模型。在这个模型中，用户空间的应用程序执行一个系统调用，这会导致应用程序阻塞。这意味着应用程序会一直阻塞，直到系统调用完成为止（数据传输完成或发生错误）。调用应用程序处于一种不再消费 CPU 而只是简单等待响应的状态，因此从处理的角度来看，这是非常有效的。

　　2.3 同步阻塞 I/O 的一种效率稍低的变种是同步非阻塞 I/O。在这种模型中，设备是以非阻塞的形式打开的。这意味着 I/O 操作不会立即完成，read操作可能会返回一个错误代码，说明这个命令不能立即满足（EAGAIN 或 EWOULDBLOCK）。

　　异步阻塞 I/O

　　2.4 IO多路复用(复用的select线程)。I/O复用模型会用到select、poll、epoll函数，这几个函数也会使进程阻塞，但是和阻塞I/O所不同的的，这两个函数可以同时阻塞多个I/O操作。而且可以同时对多个读操作，多个写操作的I/O函数进行检测，直到有数据可读或可写时，才真正调用I/O操作函数。对于每个提示符来说，我们可以获取这个描述符可以写数据、有读数据可用以及是否发生错误的通知。

　　2.5 epoll支持水平触发和边缘触发，最大的特点在于边缘触发，它只告诉进程哪些fd刚刚变为就绪状态，并且只会通知一次。还有一个特点是，epoll使用“事件”的就绪通知方式，通过epoll_ctl注册fd，一旦该fd就绪，内核就会采用类似callback的回调机制来激活该fd，epoll_wait便可以收到通知.

　　2.6 epoll的优点

　　　　* 没有最大并发连接的限制，能打开的FD的上限远大于1024（1G的内存上能监听约10万个端口）。

　　　　* 效率提升，不是轮询的方式,只管你“活跃”的连接，不会随着FD数目的增加效率下降。只有活跃可用的FD才会调用callback函数。

　　　　* 内存拷贝，利用mmap()文件映射内存加速与内核空间的消息传递；即epoll使用mmap减少复制开销。

　　表面上看epoll的性能最好，但是在连接数少并且连接都十分活跃的情况下，select和poll的性能可能比epoll好，毕竟epoll的通知机制需要很多函数回调。

　　异步非阻塞 I/O

　　2.7 异步非阻塞 I/O 模型是一种CPU处理与 I/O 重叠进行的模型。读请求会立即返回，说明 read 请求已经成功发起了。在后台完成读操作时，应用程序然后会执行其他处理操作。当 read 的响应到达时，就会产生一个信号或执行一个基于线程的回调函数来完成这次 I/O 处理过程。

---

3.aio编程方法

　　　　　　　　　　　　　　　　　　makefile编译 aio的程序时在末尾加 -lrt 因为要用到aio的第三方库

//aio对象
struct aiocb
{
  int aio_fildes;               // 要操作的文件描述符
  int aio_lio_opcode;           // 下面两个要操作码 ,批量发起aio时使用的
  volatile void *aio_buf;       // 易变的缓冲区 , 随时会因为线程而改变
  size_t aio_nbytes;            // aio读取数据的个数 , 具体看需求
  struct sigevent aio_sigevent; // 绑定的回调函数
};
AIO_READ AIO_WRITE 

volatile: 提醒编译器它后面所定义的变量随时都有可能改变
         https://www.cnblogs.com/yc_sunniwell/archive/2010/06/24/1764231.html

　　3.1  int aio_read(struct aiocb *aiocbp) ... int aio_write(struct aiocb *aiocbp)写和读差不多

　　　　aio_read 函数请求对一个有效的文件描述符进行异步读操作。这个文件描述符可以表示一个文件、套接字甚至管道。

　　　　aio_read 函数在请求进行排队之后会立即返回。如果执行成功，返回值就为 0；如果出现错误，返回值就为 -1，并设置 errno 的值。

　　　　如果使用aio_read 那么aio_lio_opcode就不用再赋值了,因为已经是read状态了

　　3.2 int aio_error( struct aiocb *aiocbp );

　　　　EINPROGRESS，说明请求尚未完成

　　　　ECANCELLED，说明请求被应用程序取消了

　　　　-1 说明发生了错误，具体错误原因可以查阅 errno

　　　　0 ,说明完成当前请求

　　　　aio_error 函数被用来主动确定aio_read/ aio_write的返回状态

　　3.3 ssize_t aio_return( struct aiocb *aiocbp );

　　　　结束或接收当前read的返回状态后才能进行aio_buf的读取

　　　　异步 I/O 和标准块 I/O 之间的另外一个区别是我们不能立即访问这个函数的返回状态，因为我们并没有阻塞在 read 调用上。

　　　　在标准的 read 调用中，返回状态是在该函数返回时提供的。但是在异步 I/O 中，我们要使用 aio_return 函数。

#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;    //文件的偏移, 0从文件的第一个字符开始读

    //进行异步读操作
    ret = aio_read(&rd);
    if(ret < 0)
    {
        perror("aio_read");
        exit(1);
    }
    //do other things
    
    couter = 0;
//  循环等待异步读操作结束
    while(aio_error(&rd) == EINPROGRESS)
    {
       // printf("第%d次
",++couter);
    }
    
    //获取异步读返回值
    ret = aio_return(&rd);    
    
    printf("

返回值为:%d
",ret);
    printf("%s
",rd.aio_buf);
    
    free(rd.aio_buf);
    close(fd);
    return 0;
}

#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 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");
    }


    wr.aio_buf = str;

    //填充aiocb结构
    wr.aio_fildes = fd;
    wr.aio_nbytes = sizeof(str);

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

    //等待异步写完成
    while(aio_error(&wr) == EINPROGRESS)
    {
        printf("hello,world
");
    }

    //获得异步写的返回值
    ret = aio_return(&wr);
    printf("


返回值为:%d
",ret);
    close(fd);
    return 0;
}
 

　　　　　　　　　　　　　　阻塞是不会占用cpu的, 阻塞是一种可中断的睡眠状态, 这种状态是不会参与cpu调度的

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

　　　　第一个参数是个保存了aiocb块地址的数组，我们可以向其内添加想要等待阻塞的异步事件，

　　　　第二个参数为向cblist注册的aiocb个数,　第三个参数为等待阻塞的超时时间，NULL为无限等待

　　　　有三种返回值 1.超时了 , 2.出错了error , 3.aio全部都返回了

　　　　aio_suspend 阻塞到aio集( cblist[] )里的所有aio完成后或者超时时间到了 才会返回

#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 , wd;

    int fd,ret,couter;

    //cblist链表
    struct aiocb *aiocb_list[2];



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



    //将rd结构体清空
    bzero(&rd,sizeof(rd));
    bzero(&rd,sizeof(wd));

    //为rd.aio_buf分配空间
    rd.aio_buf = malloc(BUFFER_SIZE + 1);

    //填充rd结构体
    rd.aio_fildes = fd;
    rd.aio_nbytes =  BUFFER_SIZE;
    rd.aio_offset = 0;
    
    wd.aio_buf = malloc(BUFFER_SIZE + 1);
    wd.aio_fildes = fd;
    wd.aio_nbytes =  BUFFER_SIZE;
    wd.aio_offset = 0;

    //将读fd的事件注册
    aiocb_list[0] = &rd;
    aiocb_list[1] = &wd;

    //进行异步读操作
    ret = aio_read(&rd);
    ret = aio_write(&wd);
    
    if(ret < 0)
    {
        perror("aio_read");
        exit(1);
    }

    printf("我要开始等待异步读事件完成
");
    //阻塞等待异步读事件完成
    ret = aio_suspend(aiocb_list,MAX_LIST,NULL);
    
    
    //获取异步读返回值
    ret = aio_return(&rd);
    ret = aio_return(&wd);
    printf("

返回值为:%d
",ret);


    return 0;
}

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

struct sigevent {
    int           sigev_notify;            //应用线程的通知类型　　一般使用默认选项 SIGEV_THREAD 创建一个线程
    int           sigev_signo;            //应用线程的信号
    union sigval  sigev_value;             //应用线程的值
    void         (*sigev_notify_function)(union sigval); //应用线程的回调函数
    pthread_attr_t *sigev_notify_attributes;  //应用线程的属性　　一般使用默认的 NULL

 };

　union sigval
　{
　　　int sival_int;
　　　void *sival_ptr;
　};

　　　　aio第三方的实现: 当第三方库-lrt 发现当前aio结束后, 就会启一个线程, 线程里调用设置好struct sigevent并调度回调函数

　　　　第一个参数mode可以有俩个实参，LIO_WAIT和LIO_NOWAIT，LIO_WAIT会阻塞该调用直到所有I/O都完成为止，LIO_NOWAIT则会挂入队列就返回.

　　　　aio数量多则使用lio_listio(), 只需要设置好aio_lio_opcode的状态 , lio_listio()会识别是read还write并做出相应的处理

　　　　批量发起AIO的两种方法

　　　　LIO_WAIT 阻塞发起　　阻塞等到所有发起的AIO全部完成后，才会返回

　　　　LIO_NOWAIT 非阻塞发起　　发起后立即返回，通过绑定的信号来通知

　　　　阻塞就不需要绑定回调函数*sig了 , 非阻塞当所有AIO都完成后自动调用回调函数*sig

　　　　设置回调函数就要面向sigevent这个对象了

#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>
int main( int argc , char* argv[] )
{
    struct aiocb rd , wr , *aiolist[2];
    int fd1 , fd2 ,ret;
    char *str = "
Myfriend";

    fd1 = open("./test.txt" , O_RDONLY);
    if ( fd1 < 0 )
    {
        perror("ropen");
        exit(1);
    }
    fd2 = open("./test.txt" , O_WRONLY | O_APPEND);
    if ( fd2 < 0 )
    {
        perror("wopen");
        exit(1);
    }
    
    bzero(&rd,sizeof(rd));
    bzero(&wr,sizeof(wr));

    rd.aio_fildes = fd1;
    rd.aio_lio_opcode = LIO_READ;
    rd.aio_buf = malloc(BUF_SIZE+1);
    if(rd.aio_buf == NULL)
    {
        perror("aio_buf");
        exit(1);
    }
    rd.aio_nbytes = BUF_SIZE;
    rd.aio_offset = 0;
    aiolist[1] = &rd;

    wr.aio_fildes = fd2;
    wr.aio_lio_opcode = LIO_WRITE;
    wr.aio_buf = str;
    wr.aio_nbytes = sizeof(str);
    aiolist[0] = &wr;

    ret = lio_listio(LIO_WAIT , aiolist , 2 , NULL);
    if ( ret < 0 )
    {
        perror("lio_listio
");
        exit(1);
    }
    printf("lio_return = %d
", ret);

    ret = aio_return(&rd);
    if ( ret < 0 )
    {
        perror("read_return 
");
        exit(1);
    }
    printf("read_return = %d
 str = %s
", ret , rd.aio_buf);
    
    ret = aio_return(&wr);
    if ( ret < 0 )
    {
        perror("write_return
");
        exit(1);
    }
    printf("write_return = %d
", ret);

    free(rd.aio_buf);
    close(fd1);
    close(fd2);
    return 0;
}

#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
");

    //获取返回值
    ret = aio_return(prd);

}

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("异步读以开始
");
    sleep(1);
    printf("异步读结束
");



    return 0;
}

---

 总结: AIO本质就是启用内核提供的线程来完成内核层到应用层数据的拷贝过程(内核层会拷贝到一个缓冲区再返回给应用层) ,有两种方式确定这个

　　　　　一 是主动询问aio_error()但是这种方式不好把握时机

　　　　　二 是绑定一个回调函数,当完成拷贝过程后执行回调函数