c++ 網絡編程（八） LINUX-epoll/windows-IOCP下 socket opoll函數用法優於select方法的epoll 以及windows下IOCP 解決多進程服務端建立進程資

時間 2019-11-09

標籤 c++ 網絡編程 linux epoll windows iocp socket opoll 函數用法優於 select 方法以及解決進程服務建立欄目 C&C++ 简体版

原文原文鏈接

原文做者：aircrafthtml

原文連接：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模型。python

一：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通道里發生了堵塞等情況咱們是沒法控制的。c++

同：
1：它們都是異步的事件驅動的網絡模型。
2：它們均可以向底層進行指針數據傳遞，當返回事件時，除可通知事件類型外，還能夠通知事件相關數據。編程

二：Epoll理解與應用。後端

一、epoll是什麼？數組

epoll是當前在Linux下開發大規模併發網絡程序的熱門人選，epoll 在Linux2.6內核中正式引入，和select類似，都是I/O多路複用(IO multiplexing)技術。服務器

Linux下設計併發網絡程序，經常使用的模型有：網絡

Apache模型（Process Per Connection，簡稱PPC）數據結構

TPC（Thread PerConnection）模型

select模型和poll模型。

epoll模型

二、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的幾個函數的介紹：

一、epoll_create函數：

/**  
 * @brief    該函數生成一個epoll專用的文件描述符。它實際上是在內核申請一空間，用來存放你想關注的socket fd上是否發生以及發生了什麼事件。 
 *  
 * @param    size    size就是你在這個epoll fd上能關注的最大socket fd數 
 *  
 * @return   生成的文件描述符 
 */  
int epoll_create(int size);

二、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>

三、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> \n", 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 \n", 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 \n", 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('\n', 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.\n");
            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.\n");
                        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.\n");
                        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請求到它的請求隊列中。