zoukankan      html  css  js  c++  java
  • c++ 网络编程(八) LINUX-epoll/windows-IOCP下 socket opoll函数用法 优于select方法的epoll 以及windows下IOCP 解决多进程服务端创建进程资源浪费问题

    原文作者:aircraft

    原文链接:https://www.cnblogs.com/DOMLX/p/9622548.html

    锲子:关于并发服务器中的I/O复用实现方式,前面在网络编程系列四还是五来着????我们讲过select的方式,但select的性能比较低,当连接数量超过几百个的时候就很慢了,并不适合以Web服务器端开发为主流的现代开发环境。因此就有了Linux下的epoll,BSD的kqueue,Solaris的/dev/poll和Windows的IOCP等复用技术。本章就来讲讲Linux下的epoll技术和Windows下的IOCP模型。

    一:IOCP和Epoll之间的异同。
    异:
    1:IOCP是WINDOWS系统下使用。Epoll是Linux系统下使用。
    2:IOCP是IO操作完毕之后,通过Get函数获得一个完成的事件通知。
    Epoll是当你希望进行一个IO操作时,向Epoll查询是否可读或者可写,若处于可读或可写状态后,Epoll会通过epoll_wait进行通知。
    3:IOCP封装了异步的消息事件的通知机制,同时封装了部分IO操作。但Epoll仅仅封装了一个异步事件的通知机制,并不负责IO读写操作。Epoll保持了事件通知和IO操作间的独立性,更加简单灵活。
    4:基于上面的描述,我们可以知道Epoll不负责IO操作,所以它只告诉你当前可读可写了,并且将协议读写缓冲填充,由用户去读写控制,此时我们可以做出额外的许多操作。IOCP则直接将IO通道里的读写操作都做完了才通知用户,当IO通道里发生了堵塞等状况我们是无法控制的。

    同:
    1:它们都是异步的事件驱动的网络模型。
    2:它们都可以向底层进行指针数据传递,当返回事件时,除可通知事件类型外,还可以通知事件相关数据。

     

    二:Epoll理解与应用。

     

    1、epoll是什么?

     

    epoll是当前在Linux下开发大规模并发网络程序的热门人选,epoll Linux2.6内核中正式引入,和select相似,都是I/O多路复用(IO multiplexing)技术

     

    Linux下设计并发网络程序,常用的模型有:

     

          Apache模型(Process Per Connection,简称PPC

     

         TPCThread PerConnection)模型

     

         select模型和poll模型。

     

         epoll模型

     

    2、epoll与select对比优化:

     

      • 基于select的I/O复用技术速度慢的原因:
        1,调用select函数后常见的针对所有文件描述符的循环语句。它每次事件发生需要遍历所有文件描述符,找出发生变化的文件描述符。(以前写的示例没加循环)

        2,每次调用select函数时都需要向该函数传递监视对象信息。即每次调用select函数时向操作系统传递监视对象信息,至于为什么要传?是因为我们监视的套接字变化的函数,而套接字是操作系统管理的。(这个才是最耗效率的)

        注释:基于这样的原因并不是说select就没用了,在这样的情况下就适合选用select:1,服务端接入者少 2,程序应具有兼容性。

      • epoll是怎么优化select问题的:
        1,每次发生事件它不需要循环遍历所有文件描述符,它把发生变化的文件描述符单独集中到了一起。

        2,仅向操作系统传递1次监视对象信息,监视范围或内容发生变化时只通知发生变化的事项。

      • 实现epoll时必要的函数和结构体

        函数:
        epoll_create:创建保存epoll文件描述符的空间,该函数也会返回文件描述符,所以终止时,也要调用close函数。(创建内存空间)

        epoll_ctl:向空间注册,添加或修改文件描述符。(注册监听事件)

        epoll_wait:与select函数类似,等待文件描述符发生变化。(监听事件回调)

        结构体:
        struct epoll_event
        {
        __uint32_t events;
        epoll_data_t data;
        }

        typedef union epoll_data
        {
        void *ptr;
        int fd;
        __uinit32_t u32;
        __uint64_t u64;
        } epoll_data_t;

    epoll的几个函数的介绍:

    1、epoll_create函数:

    /**  
     * @brief    该函数生成一个epoll专用的文件描述符。它其实是在内核申请一空间,用来存放你想关注的socket fd上是否发生以及发生了什么事件。 
     *  
     * @param    size    size就是你在这个epoll fd上能关注的最大socket fd数 
     *  
     * @return   生成的文件描述符 
     */  
    int epoll_create(int size);  

    2、epoll_ctl函数:

    /**  
     * @brief    该函数用于控制某个epoll文件描述符上的事件,可以注册事件,修改事件,删除事件。 
     *  
     * @param    epfd    由 epoll_create 生成的epoll专用的文件描述符 
     * @param    op      要进行的操作例如注册事件,可能的取值EPOLL_CTL_ADD 注册、EPOLL_CTL_MOD 修 改、EPOLL_CTL_DEL 删除 
     * @param    fd      关联的文件描述符 
     * @param    event   指向epoll_event的指针 
     *  
     * @return   0       succ 
     *           -1      fail 
     */  
    int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);  

    其中用到的数据结构结构如下:
    op值:
    EPOLL_CTL_ADD:注册新的fd到epfd中;

     

    EPOLL_CTL_MOD:修改已经注册的fd的监听事件;

     

    EPOLL_CTL_DEL:从epfd中删除一个fd;

    typedef union epoll_data { 
    void *ptr;
    int fd;
    __uint32_t u32;
    __uint64_t u64;
    } epoll_data_t; 
    struct epoll_event {
    __uint32_t events; /* Epoll events */
    epoll_data_t data; /* User data variable */
    };

    常用的事件类型:
    EPOLLIN :表示对应的文件描述符可以读;
    EPOLLOUT:表示对应的文件描述符可以写;
    EPOLLPRI:表示对应的文件描述符有紧急的数据可读
    EPOLLERR:表示对应的文件描述符发生错误;
    EPOLLHUP:表示对应的文件描述符被挂断;
    EPOLLET: 表示对应的文件描述符有事件发生;

    例:

        <code class="language-cpp">struct epoll_event ev;    
        //设置与要处理的事件相关的文件描述符    
        ev.data.fd=listenfd;    
        //设置要处理的事件类型    
        ev.events=EPOLLIN|EPOLLET;    
        //注册epoll事件    
        epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev); </code>  

    3、epoll_wait函数:

    /**  
     * @brief    该函数用于轮询I/O事件的发生 
     *  
     * @param    epfd        由epoll_create 生成的epoll专用的文件描述符 
     * @param    events      用于回传代处理事件的数组 
     * @param    maxevents   每次能处理的事件数 
     * @param    timeout     等待I/O事件发生的超时值;-1相当于阻塞,0相当于非阻塞。一般用-1即可 
     *  
     * @return   >=0         返回发生事件数 
     *           -1          错误 
     */  
    int epoll_wait(int epfd,struct epoll_event * events,int maxevents,int timeout);  

    用改良的epoll实现回声服务端代码:

    #include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <unistd.h>
    #include <arpa/inet.h>
    #include <sys/socket.h>
    #include <sys/epoll.h>
    
    #define BUF_SIZE 100
    #define EPOLL_SIZE 50
    void error_handling(char *buf);
    
    int main(int argc, const char * argv[]) {
        int serv_sock, clnt_sock;
        struct sockaddr_in serv_adr, clnt_adr;
        socklen_t adr_sz;
        int str_len, i;
        char buf[BUF_SIZE];
    
        //类似select的fd_set变量查看监视对象的状态变化,epoll_event结构体将发生变化的文件描述符单独集中到一起
        struct epoll_event *ep_events;
        struct epoll_event event;
        int epfd, event_cnt;
    
        if(argc != 2)
        {
            printf("Usage: %s <port> 
    ", argv[0]);
            exit(1);
        }
    
        serv_sock = socket(PF_INET, SOCK_STREAM, 0);
        if(serv_sock == -1)
            error_handling("socket() error");
    
        memset(&serv_adr, 0, sizeof(serv_adr));
        serv_adr.sin_family = AF_INET;
        serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);
        serv_adr.sin_port = htons(atoi(argv[1]));
    
        if(bind(serv_sock, (struct sockaddr *) &serv_adr, sizeof(serv_adr)) == -1)
            error_handling("bind() error");
    
        if(listen(serv_sock, 5) == -1)
            error_handling("listen() error");
    
        //创建文件描述符的保存空间称为“epoll例程”
        epfd = epoll_create(EPOLL_SIZE);
        ep_events = malloc(sizeof(struct epoll_event) *EPOLL_SIZE);
    
        //添加读取事件的监视(注册事件)
        event.events = EPOLLIN;  //读取数据事件
        event.data.fd = serv_sock;
        epoll_ctl(epdf, EPOLL_CTL_ADD, serv_sock, &event);
    
        while (1)
        {
            //响应事件,返回发生事件的文件描述符数
            event_cnt = epoll_wait(epfd, ep_events, EPOLL_SIZE, -1);  //传-1时,一直等待直到事件发生
            if(event_cnt == -1)
            {
                puts("epoll_wait() error");
                break;
            }
    
            //服务端套接字和客服端套接字
            for (i = 0; i < event_cnt; i++) {
                if(ep_events[i].data.fd == serv_sock)//服务端与客服端建立连接
                {
                    adr_sz = sizeof(clnt_adr);
                    clnt_sock = accept(serv_sock, (struct sockaddr *)&clnt_adr, &adr_sz);
                    event.events = EPOLLIN;
                    event.data.fd = clnt_sock;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, clnt_sock, &event);
                    printf("connected client: %d 
    ", clnt_sock);
                }
                else  //连接之后传递数据
                {
                    str_len = read(ep_events[i].data.fd, buf, BUF_SIZE);
                    if(str_len == 0)
                    {
                        //删除事件
                        epoll_ctl(epfd, EPOLL_CTL_DEL, ep_events[i].data.fd, NULL);
                        close(ep_events[i].data.fd);
                        printf("closed client: %d 
    ", ep_events[i].data.fd);
                    }
                    else
                    {
                        write(ep_events[i].data.fd, buf, str_len);
                    }
                }
            }
        }
    
        close(serv_sock);
        close(epfd);
        return 0;
    }
    
    void error_handling(char *message)
    {
        fputs(message, stderr);
        fputc('
    ', stderr);
        exit(1);
    }

    epoll客户端代码:

    #define _GNU_SOURCE
    #include "sysutil.h"
    #include "buffer.h"
    #include <sys/epoll.h>
    
    int main(int argc, char const *argv[])
    {
        //创建client套接字
        int sockfd = tcp_client(0);
        //调用非阻塞connect函数
        int ret = nonblocking_connect(sockfd, "localhost", 9981, 5000);
        if(ret == -1)
        {
            perror("Connect Timeout .");
            exit(EXIT_FAILURE);
        }
    
        //将三个fd设置为Non-Blocking
        activate_nonblock(sockfd);
        activate_nonblock(STDIN_FILENO);
        activate_nonblock(STDOUT_FILENO);
    
    
        buffer_t recvbuf; //sockfd -> Buffer -> stdout
        buffer_t sendbuf; //stdin -> Buffer -> sockfd
    
        //初始化缓冲区
        buffer_init(&recvbuf);
        buffer_init(&sendbuf);
    
    
        //创建epoll
        int epollfd = epoll_create1(0);
        if(epollfd == -1)
            ERR_EXIT("create epoll");
        struct epoll_event events[1024];
    
        uint32_t sockfd_event = 0;
        uint32_t stdin_event = 0;
        uint32_t stdout_event = 0;
    
        epoll_add_fd(epollfd, sockfd, sockfd_event);
        epoll_add_fd(epollfd, STDIN_FILENO, stdin_event);
        epoll_add_fd(epollfd, STDOUT_FILENO, stdout_event);
    
    
        while(1)
        {
            //重新装填epoll事件
            sockfd_event = 0;
            stdin_event = 0;
            stdout_event = 0;
            //epoll无法每次都重新装填,所以给每个fd添加一个空事件
            
            if(buffer_is_readable(&sendbuf))
            {
                sockfd_event |= kWriteEvent;
            }
            if(buffer_is_writeable(&sendbuf))
            {
                stdin_event |= kReadEvent;
            }
            if(buffer_is_readable(&recvbuf))
            {
                stdout_event |= kWriteEvent;
            }
            if(buffer_is_writeable(&recvbuf))
            {
                sockfd_event |= kReadEvent;
            }
    
            epoll_mod_fd(epollfd, sockfd, sockfd_event);
            epoll_mod_fd(epollfd, STDIN_FILENO, stdin_event);
            epoll_mod_fd(epollfd, STDOUT_FILENO, stdout_event);
    
    
            //监听fd数组
            int nready = epoll_wait(epollfd, events, 1024, 5000);
            if(nready == -1)
                ERR_EXIT("epoll wait");
            else if(nready == 0)
            {
                printf("epoll timeout.
    ");
                continue;
            }
            else
            {
                int i;
                for(i = 0; i < nready; ++i)
                {
                    int peerfd = events[i].data.fd;
                    int revents = events[i].events;
                    if(peerfd == sockfd && revents & kReadREvent)
                    {
                        //从sockfd接收数据到recvbuf
                        if(buffer_read(&recvbuf, peerfd) == 0)
                        {
                            fprintf(stderr, "server close.
    ");
                            exit(EXIT_SUCCESS);
                        } 
                    }
                        
                    if(peerfd == sockfd && revents & kWriteREvent)
                    {
                        buffer_write(&sendbuf, peerfd); //将sendbuf中的数据写入sockfd
                    }
    
                    if(peerfd == STDIN_FILENO && revents & kReadREvent)
                    {
                        //从stdin接收数据写入sendbuf
                        if(buffer_read(&sendbuf, peerfd) == 0)
                        {
                            fprintf(stderr, "exit.
    ");
                            exit(EXIT_SUCCESS);
                        } 
                    }
    
                    if(peerfd == STDOUT_FILENO && revents & kWriteREvent)
                    {
                        buffer_write(&recvbuf, peerfd); //将recvbuf中的数据输出至stdout
                    }
                }
            }
    
        }
    
    }

    条件触发和边缘触发

      • 什么是条件触发和边缘触发?它们是指事件响应的方式,epoll默认是条件触发的方式。条件触发是指:只要输入缓冲中有数据就会一直通知该事件,循环响应epoll_wait。而边缘触发是指:输入缓冲收到数据时仅注册1次该事件,即使输入缓冲中还留有数据,也不会再进行注册,只响应一次。

      • 边缘触发相对条件触发的优点:可以分离接收数据和处理数据的时间点,从实现模型的角度看,边缘触发更有可能带来高性能。

      • 将上面epoll实例改为边缘触发:
        1,首先改写 event.events = EPOLLIN | EPOLLET; (EPOLLIN:读取数据事件 EPOLLET:边缘触发方式)

        2,边缘触发只响应一次接收数据事件,所以要一次性全部读取输入缓冲中的数据,那么就需要判断什么时候数据读取完了?Linux声明了一个全局的变量:int errno; (error.h中),它能记录发生错误时提供额外的信息。这里就可以用它来判断是否读取完数据:

    str_len = read(...);
    if(str_len < 0)
    {
        if(errno == EAGAIN) //读取输入缓冲中的全部数据的标志
            break;
    }

    3,边缘触发方式下,以阻塞方式工作的read&write有可能会引起服务端的长时间停顿。所以边缘触发一定要采用非阻塞的套接字数据传输形式。那么怎么将套接字的read,write数据传输形式修改为非阻塞模式呢?

    //fd套接字文件描述符,将此套接字数据传输模式修改为非阻塞
    void setnonblockingmode(int fd)
    {
        int flag = fcntl(fd, F_GETFL,0); //得到套接字原来属性
        fcntl(fd, F_SETFL, flag | O_NONBLOCK);//在原有属性基础上设置添加非阻塞模式
    }

    三.IOCP理解与应用。

    扯远点。首先传统服务器的网络IO流程如下:
    接到一个客户端连接->创建一个线程负责这个连接的IO操作->持续对新线程进行数据处理->全部数据处理完毕->终止线程。
    但是这样的设计代价是:

    • 1:每个连接创建一个线程,将导致过多的线程。
      • 2:维护线程所消耗的堆栈内存过大。
        • 3:操作系统创建和销毁线程过大。
          • 4:线程之间切换的上下文代价过大。


    此时我们可以考虑使用线程池解决其中3和4的问题。这种传统的服务器网络结构称之为会话模型。
    后来我们为防止大量线程的维护,创建了I/O模型,它被希望要求可以:
    1:允许一个线程在不同时刻给多个客户端进行服务。
    2:允许一个客户端在不同时间被多个线程服务。


    这样做的话,我们的线程则会大幅度减少,这就要求以下两点:
    1:客户端状态的分离,之前会话模式我们可以通过线程状态得知客户端状态,但现在客户端状态要通过其他方式获取。
    2:I/O请求的分离。一个线程不再服务于一个客户端会话,则要求客户端对这个线程提交I/O处理请求。

    那么就产生了这样一个模式,分为三部分:

    • 1:会话状态管理模块。它负责接收到一个客户端连接,就创建一个会话状态。
      • 2:当会话状态发生改变,例如断掉连接,接收到网络消息,就发送一个I/O请求给 I/O工作模块进行处理。
        • 3:I/O工作模块接收到一个I/O请求后,从线程池里唤醒一个工作线程,让该工作线程处理这个I/O请求,处理完毕后,该工作线程继续挂起。

    上面的做法,则将网络连接 和I/O工作线程分离为三个部分,相互通讯仅依靠 I/O请求。此时可知有以下一些建议:

    • 1:在进行I/O请求处理的工作线程是被唤醒的工作线程,一个CPU对应一个的话,可以最大化利用CPU。所以 活跃线程的个数 建议等于 硬件CPU个数。
      • 2:工作线程我们开始创建了线程池,免除创建和销毁线程的代价。因为线程是对I/O进行操作的,且一一对应,那么当I/O全部并行时,工作线程必须满足I/O并行操作需求,所以 线程池内最大工作线程个数 建议大于或者等于 I/O并行个数。
        • 3:但是我们可知CPU个数又限制了活跃的线程个数,那么线程池过大意义很低,所以按常规建议 线程池大小 等于 CPU个数*2 左右为佳。例如,8核服务器建议创建16个工作线程的线程池。 上面描述的依然是I/O模型并非IOCP,那么IOCP是什么呢,全称 IO完成端口。

    它是一种WIN32的网络I/O模型,既包括了网络连接部分,也负责了部分的I/O操作功能,用于方便我们控制有并发性的网络I/O操作。它有如下特点:

    • 1:它是一个WIN32内核对象,所以无法运行于Linux.
      • 2:它自己负责维护了工作线程池,同时也负责了I/O通道的内存池。
        • 3:它自己实现了线程的管理以及I/O请求通知,最小化的做到了线程的上下文切换。
          • 4:它自己实现了线程的优化调度,提高了CPU和内存缓冲的使用率。

    使用IOCP的基本步骤很简单:

    • 1:创建IOCP对象,由它负责管理多个Socket和I/O请求。CreateIoCompletionPort需要将IOCP对象和IOCP句柄绑定。
      • 2:创建一个工作线程池,以便Socket发送I/O请求给IOCP对象后,由这些工作线程进行I/O操作。注意,创建这些线程的时候,将这些线程绑定到IOCP上。
        • 3:创建一个监听的socket。
          • 4:轮询,当接收到了新的连接后,将socket和完成端口进行关联并且投递给IOCP一个I/O请求。注意:将Socket和IOCP进行关联的函数和创建IOCP的函数一样,都是CreateIoCompletionPort,不过注意传参必然是不同的。
            • 5:因为是异步的,我们可以去做其他,等待IOCP将I/O操作完成会回馈我们一个消息,我们再进行处理。
              • 其中需要知道的是:I/O请求被放在一个I/O请求队列里面,对,是队列,LIFO机制。当一个设备处理完I/O请求后,将会将这个完成后的I/O请求丢回IOCP的I/O完成队列。
                • 我们应用程序则需要在GetQueuedCompletionStatus去询问IOCP,该I/O请求是否完成。
                  • 其中有一些特殊的事情要说明一下,我们有时有需要人工的去投递一些I/O请求,则需要使用PostQueuedCompletionStatus函数向IOCP投递一个I/O请求到它的请求队列中。

     

     

    最后说一句啦。本网络编程入门系列博客是连载学习的,有兴趣的可以看我博客其他篇。。。。c++ 网络编程课设入门超详细教程 ---目录

     

    参考博客:http://blog.csdn.net/penzo/article/details/5986574

    参考博客:https://blog.csdn.net/educast/article/details/15500349

    参考博客:https://blog.csdn.net/u010223072/article/details/49276415

    参考书籍:《TCP/IP网络编程--尹圣雨》

     

    若有兴趣交流分享技术,可关注本人公众号,里面会不定期的分享各种编程教程,和共享源码,诸如研究分享关于c/c++,python,前端,后端,opencv,halcon,opengl,机器学习深度学习之类有关于基础编程,图像处理和机器视觉开发的知识

  • 相关阅读:
    hdu5728 PowMod
    CF1156E Special Segments of Permutation
    CF1182E Product Oriented Recurrence
    CF1082E Increasing Frequency
    CF623B Array GCD
    CF1168B Good Triple
    CF1175E Minimal Segment Cover
    php 正则
    windows 下安装composer
    windows apache "The requested operation has failed" 启动失败
  • 原文地址:https://www.cnblogs.com/DOMLX/p/9622548.html
Copyright © 2011-2022 走看看