《Linux高性能服務器編程》學習總結（九）——I/O復用

 

第九章      I/O復用

　　I/O復用技術是重要的提高服務器工作效率和性能的手段，Linux下實現I/O復用的系統調用主要有select、poll和epoll。

　　首先我們來看一下select的函數原型和常用的宏：

#include<sys/select.h>
int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout);
FD_ZERO(fd_set *fdset);                    //清除fdset所有位
FD_SET(int fd, fd_set* fdset);                    //設置fdset的位fd
FD_CLR(int fd, fd_set* fdset);                //清除fdset的位fd
int FD_ISSET(int fd, fd_set* fdset);            //測試fdset的位fd是否被設置

　　首先來看select函數原型，nfds指定了被監聽的文件描述符的總數，其值通常被設定為所有文件描述符的最大值加一，接下來的三個fd_set*類型的參數分別指向可讀可寫和異常事件對應的文件描述符集合，最后一個參數是一個微秒級的定時器，表示select阻塞這個時間后繼續執行，如果為0則立即返回，如果為NULL將一直阻塞。

　　通過觀察fd_set結構體的原型，我們發現其僅包含一個整形數組，該數組的每一位都標記了一個文件描述符，所以select有最大可監控的文件描述符的限制。后面的宏是為了簡化對於fd_set的位操作。select函數成功時返回就緒文件描述符的總數，如果在超時時間內沒有任何文件描述符就緒，則select返回0，如果在select阻塞期間程序收到信號，則select立即返回-1並置errno為EINTR。

　　select在何種情況下會認為文件描述符產生了可讀、可寫或異常情況呢？首先，當socket處於以下狀態會認為其可讀：1）socket內核接收緩沖區中的字節數大於或等於其低水位標記，此時我們可以無阻塞地讀該socket，且讀操作返回值大於0；2）socket的對端關閉連接，此時讀操作返回0；3）監聽socket上有新的請求；4）socket上有未處理的錯誤。而以下狀態會認為socket可寫：1）socket內核發送緩沖區中的可用字節數大於或等於其低水位標記，此時我們可以無阻塞地寫該socket，且寫操作返回值大於0；2）socket的寫操作被關閉，對寫操作關閉的socket執行寫操作會觸發SIGPIPE信號；3）socket使用非阻塞connect連接成功或者失敗（超時）之后；4）socket上有未處理的錯誤。而異常情況只有一種，就是產生了帶外數據。

　　我們來用一個例子看一下select程序如何來寫以及select如何同時處理普通數據和帶外數據的：

/*************************************************************************
    > File Name: 9-1.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Sat 03 Feb 2018 07:23:52 PM PST
 ************************************************************************/

#include"head.h"
using namespace std;

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi(argv[2]);

    int ret = 0;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    address.sin_port = htons(port);
    inet_pton(AF_INET, ip, &address.sin_addr);

    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(listenfd >= 0);

    ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
    assert(ret != -1);

    ret = listen(listenfd, 5);
    assert(ret != -1);

    struct sockaddr_in client_address;
    socklen_t client_addrlength = sizeof(client_address);
    int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);
    if(connfd < 0) {
        printf("errno is: %d\n", errno);
    }

    char buf[1024];
    fd_set read_fds;
    fd_set exception_fds;
    FD_ZERO(&read_fds);
    FD_ZERO(&exception_fds);

    while(1) {
        memset(buf, 0, sizeof(buf));
        FD_SET(connfd, &read_fds);
        FD_SET(connfd, &exception_fds);
        ret = select(connfd + 1, &read_fds, NULL, &exception_fds, NULL);
        if(ret < 0) {
            printf("selection failure\n");
            break;
        }
        if(FD_ISSET(connfd, &read_fds)) {
            ret = recv(connfd, buf, sizeof(buf), 0);
            if(ret <= 0) break;
            printf("get %d bytes of normal data: %s\n", ret, buf);
        }
        memset(buf, 0, sizeof(buf));
        if(FD_ISSET(connfd, &exception_fds)) {
            ret = recv(connfd, buf, sizeof(buf), MSG_OOB);
            if(ret <= 0) break;
            printf("get %d bytes of oob data: %s\n", ret, buf);
        }
    }
    close(connfd);
    close(listenfd);
    return 0;
}

　　客戶端我們使用了前面的5-6程序發送普通數據和帶外數據。在實際測試過程中發現了一個問題，即原書上的FD_ISSET里面判斷是readfd還是exceptionfd那里是if和else if，測試時只輸出了普通數據而沒有輸出帶外數據，將客戶端發送帶外數據后sleep一下就可以正常接收，或者像我上面程序一樣使用兩個if判斷也可以正常接收。分析一下原因，如果是if和else if，則這兩個分支只能觸發一個，而對面發送的數據是既有普通數據又有帶外數據的，所以導致了這個結果，而如果客戶端sleep，則服務器端可以返回后重新讀取一次帶外數據，而改為兩個if，就需要判斷兩次，這樣就不會有處理不了帶外數據的情況了。

　　poll的系統調用和select相似，也是在一段時間內輪詢一定數量的文件描述符。當timeout值為-1時poll調用將阻塞直到某個事件發生。

　　epoll是Linux特有的I/O復用函數，其實現和select、poll有很大區別。epoll將用戶關心的文件描述符上的事件放在內核里的一個事件表中，從而無須像select和poll那樣每次調用都要重復傳入文件描述符集或事件集，但是epoll需要一個額外的文件描述符來標識內核中的這個事件表。與poll不同的是，epoll如果檢測到事件，就將所有就緒時間從內核時間表中復制到events指向的數組中，這樣就極大提高了應用程序檢索就緒文件描述符的效率，從O(n)的時間復雜度降為了O(1)。我們來看一下epoll的幾個函數：

#include<sys/epoll.h>
int epoll_create(int size);
int epoll_ctl(int epfd, int op, int fd, struct epoll_event* event);
int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);

　　創建epoll的函數size參數現在是沒有用處的，只是給內核一個提示，告訴它事件表需要多大。操作epoll的函數中op參數指定了操作類型，一共有注冊、修改和刪除三種，而event參數則描述了事件。

　　epoll對於文件描述符的操作有兩種模式：LT和ET，其中LT是默認模式，這種模式下其效率相當於一個稍微改進的poll，效率沒有顯著提高，而ET模式則是epoll的高效工作模式。對於LT模式的文件描述符，當epoll_wait檢測到其上有事件發生並將此事件通知應用程序后，因公程序可以不立即處理該事件，這樣當epoll_wait再次被觸發，還會再向應用程序通告此事件，知道該事件被處理。而對於ET模式，當epoll_wait檢測到其上有事件發生，將其通告應用程序，應用程序必須馬上處理，因為后續的epoll_wait將不再向應用程序通知這一事件。可見，ET模式降低了同一個epoll事件被重復觸發的次數，所以效率更高。我們用一個實例來看一下：

/*************************************************************************
    > File Name: 9-3.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Sat 03 Feb 2018 10:35:56 PM PST
 ************************************************************************/

#include"head.h"
using namespace std;

#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 10

//設置文件描述符為非阻塞模式
int setnonblocking(int fd) {
    int old_option = fcntl(fd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(fd, F_SETFL, new_option);
    return old_option;
}

//以兩種不同模式將事件注冊到epoll中
void addfd(int epollfd, int fd, bool enable_et) {
    epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN;
    if(enable_et) event.events |= EPOLLET;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
    setnonblocking(fd);
}

void lt(epoll_event* events, int number, int epollfd, int listenfd) {
    char buf[BUFFER_SIZE];
    for(int i = 0; i < number; i ++) {
        int sockfd = events[i].data.fd;
        if(sockfd == listenfd) {
            struct sockaddr_in client_address;
            socklen_t client_addrlength = sizeof(client_address);
            int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);
            addfd(epollfd, connfd, false);
        }
        else if(events[i].events & EPOLLIN) {
            printf("event trigger once\n");
            memset(buf, 0, sizeof(buf));
            int ret = recv(sockfd, buf, BUFFER_SIZE, 0);
            if(ret <= 0) {
                close(sockfd);
                continue;
            }
            printf("get %d bytes of content: %s\n", ret, buf);
        }
        else printf("something else happened\n");
    }
}

void et(epoll_event* events, int number, int epollfd, int listenfd) {
    char buf[BUFFER_SIZE];
    for(int i = 0; i < number; i ++) {
        int sockfd = events[i].data.fd;
        if(sockfd == listenfd) {
            struct sockaddr_in client_address;
            socklen_t client_addrlength = sizeof(client_address);
            int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);
            addfd(epollfd, connfd, true);
        }
        else if(events[i].events & EPOLLIN) {
            //這段代碼不會被重復觸發，所以我們循環讀取
            printf("event trigger once\n");
            while(1) {
                memset(buf, 0, sizeof(buf));
                int ret = recv(sockfd, buf, BUFFER_SIZE, 0);
                if(ret < 0) {
                    //非阻塞模式的I/O，當下面的條件成立表示數據已經全部取走
                    if((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
                        printf("read later\n");
                        break;
                    }
                    close(sockfd);
                    break;
                }
                else if(ret == 0) close(sockfd);
                else printf("get %d bytes of content: %s\n", ret, buf);
            }
        }
        else printf("something else happened\n");
    }
}

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }

    const char* ip = argv[1];
    int port = atoi(argv[2]);

    int ret = 0;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    address.sin_port = htons(port);
    inet_pton(AF_INET, ip, &address.sin_addr);

    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(listenfd >= 0);

    ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
    assert(ret != -1);

    ret = listen(listenfd, 5);
    assert(ret != -1);

    epoll_event events[MAX_EVENT_NUMBER];
    int epollfd = epoll_create(5);
    assert(epollfd != -1);
    addfd(epollfd, listenfd, true);

    while(1) {
        int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
        if(ret < 0) {
            printf("epoll failure\n");
            break;
        }
        //lt(events, ret, epollfd, listenfd);
        et(events, ret, epollfd, listenfd);
    }
    close(listenfd);
    return 0;
}

　　在上面的實驗中，同樣發送25個字符，第一個LT工作模式下epoll一共向應用程序通知了三次，而第二種的ET工作模式僅僅通知一次。

　　即使我們使用ET模式，那一個socket上的事件也有可能被觸發多次，比如一個線程讀取socket上的數據后開始處理這些數據，而在數據的處理過程中該socket上又有新數據可讀，此時另外一個線程來讀取這些數據，於是就出現了兩個線程同時操作一個socket的情況。而我們期望一個socket連接在任意時刻都只被一個線程處理，這一點我們可以用EPOLLONESHOT事件實現。對於注冊了EPOLLONESHOT的文件描述符，操作系統最多觸發其上注冊的一個可讀、可寫或異常事件，且只觸發一次，除非我們使用epoll_ctl重新注冊，這樣當一個線程在處理某個socket時，其他線程是不可能有機會操作這個socket的。所以當這個線程操作完畢，應該馬上注冊EPOLLONESHOT事件以確保其可以再次被觸發EPOLLIN事件。用一段代碼來看一下：

/*************************************************************************
    > File Name: 9-4.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Sun 04 Feb 2018 12:24:13 AM PST
 ************************************************************************/

#include"head.h"
using namespace std;

#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 1024
struct fds {
    int epollfd;
    int sockfd;
};

int setnonblocking(int fd) {
    int old_option = fcntl(fd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(fd, F_SETFL, new_option);
    return old_option;
}

void addfd(int epollfd, int fd, bool enable_et) {
    epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN | EPOLLET;
    if(enable_et) event.events |= EPOLLONESHOT;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
    setnonblocking(fd);
}

void reset_oneshot(int epollfd, int fd) {
    epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN | EPOLLONESHOT | EPOLLET;
    epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &event);
}

void* worker(void* arg) {
    int sockfd = ((fds*)arg)->sockfd;
    int epollfd = ((fds*)arg)->epollfd;
    printf("start new thread to receive data on fd: %d\n", sockfd);
    char buf[BUFFER_SIZE];
    memset(buf, 0, sizeof(buf));
    while(1) {
        int ret = recv(sockfd, buf, BUFFER_SIZE, 0);
        if(ret == 0) {
            close(sockfd);
            printf("foreiner closed the connection\n");
            break;
        }
        else if(ret < 0) {
            if(errno == EAGAIN) {
                reset_oneshot(epollfd, sockfd);
                printf("read later\n");
                break;
            }
        }
        else {
            printf("get content: %s\n", buf);
            memset(buf, 0, sizeof(buf));
            sleep(5);
        }
    }
    printf("end thread receiving data on fd: %d\n", sockfd);
}

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }

    const char* ip = argv[1];
    int port = atoi(argv[2]);

    int ret = 0;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    address.sin_port = htons(port);
    inet_pton(AF_INET, ip, &address.sin_addr);

    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(listenfd >= 0);

    ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
    assert(ret != -1);

    ret = listen(listenfd, 5);
    assert(ret != -1);

    epoll_event events[MAX_EVENT_NUMBER];
    int epollfd = epoll_create(5);
    assert(epollfd != -1);
    addfd(epollfd, listenfd, false);
    
    while(1) {
        int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
        if(ret < 0) {
            printf("epoll failure\n");
            break;
        }
        for(int i = 0; i < ret; i ++) {
            int sockfd = events[i].data.fd;
            if(sockfd == listenfd) {
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof(client_address);
                int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);
                addfd(epollfd, connfd, true);
            }
            else if(events[i].events & EPOLLIN) {
                pthread_t thread;
                fds fds_for_new_worker;
                fds_for_new_worker.epollfd = epollfd;
                fds_for_new_worker.sockfd = sockfd;
                pthread_create(&thread, NULL, worker, (void*)&fds_for_new_worker);
            }
            else printf("something else happened\n");
        }
    }
    close(listenfd);
    return 0;
}

　　代碼中我們用sleep(5)來模擬數據處理的過程，當數據在處理時，我們又一次發送了數據，而此時服務器並沒有調用另外一個線程繼續工作，而是由原線程處理，我們看到兩次數據都是由5號線程處理的。這樣保證了連接的完整性，從而避免了很多可能的競態條件。

　　具體分析了三種I/O復用模式，我們來對比一下他們的異同與優缺點。這三個函數都能在一定時間內監聽一定數量的文件描述符，若一個或多個文件描述符有時間返回時函數返回就緒的文件描述符的個數。select的參數類型fd_set沒有把文件描述符和事件綁定，因此我們需要提供這三個集合，而且select也只能監聽這三種事件，另一方面內核對fd_set集合會在線修改，所以應用程序下次調用select需要重置三個集合。而poll將所有事件放在了一起，使編程接口更加簡潔，且內核每次操作的是pollfd中的revents成員，events並未改變。最后epoll則完全不同，它在內核維護一個內核事件表，每次epoll_wait操作都是直接從該內核事件表中取得用戶注冊的事件，而無需從用戶空間重復讀入這些事件。

　　另外，poll和epoll可以監聽的最大文件描述符數目都能達到系統允許的最大值，即65535，而select由於fd_set的本質是一個整形數組，每一位代表一個文件描述符，所以其能監聽的最大數量有限制。

　　從效率和原理上來講，select和poll都工作在低效的LT模式，而且采用輪詢的方式，而epoll則不同，他支持高效的ET模式，而且可以支持EPOLLONESHOT事件以進一步減少可讀可寫和異常事件的觸發次數，不僅如此，它采用了回調的方式，內核檢測到就緒的文件描述符就觸發回調函數，回調函數將該文件描述符上的對應事件插入內核的就緒隊列，最后將就緒隊列中的內容拷貝到用戶空間。但是當活動連接較多時，epoll的效率未必比select和poll高，所以epoll_wait適用於連接數量多但是活動連接少的情況。

　　當我們對一個非阻塞的socket調用一個connect，而連接又沒有被馬上建立，這時，我們可以調用select、poll等函數來監聽socket上的可寫事件，當函數返回時利用getsockopt讀取錯誤碼並清除錯誤，如果錯誤碼是0則表示連接成功建立，否則失敗。我們來看一下這種情況怎么實現：

/*************************************************************************
    > File Name: 9-5.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Sun 04 Feb 2018 07:47:09 PM PST
 ************************************************************************/

#include"head.h"
using namespace std;

#define BUFFER_SIZE 1024

int setnonblocking(int fd) {
    int old_option = fcntl(fd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(fd, F_SETFL, new_option);
    return old_option;
}

//非阻塞連接，如果函數成功則返回連接的socket，不成功返回-1
int unblock_connect(const char* ip, int port, int time) {
    int ret = 0;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    inet_pton(AF_INET, ip, &address.sin_addr);
    address.sin_port = htons(port);

    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    int fdopt = setnonblocking(sockfd);
    ret = connect(sockfd, (struct sockaddr*)&address, sizeof(address));
    if(ret == 0) {
        printf("connect with server immediately\n");
        fcntl(sockfd, F_SETFL, fdopt);
        return sockfd;
    }
    else if(errno != EINPROGRESS) {
        printf("unblock connect not support\n");
        return -1;
    }
    fd_set readfds;
    fd_set writefds;
    struct timeval timeout;

    FD_ZERO(&readfds);
    FD_ZERO(&writefds);

    timeout.tv_sec = time;
    timeout.tv_usec = 0;

    ret = select(sockfd + 1, NULL, &writefds, NULL, &timeout);
    if(ret <= 0) {
        printf("connection time out\n");
        close(sockfd);
        return -1;
    }
    
    if(!FD_ISSET(sockfd, &writefds)) {
        printf("no events on sockfd found\n");
        close(sockfd);
        return -1;
    }

    int error = 0;
    socklen_t length = sizeof(error);
    if(getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &length) < 0) {
        printf("get socket option failed\n");
        close(sockfd);
        return -1;
    }

    if(errno != 0) {
        printf("connection failed after select with the error: %d\n", error);
        close(sockfd);
        return -1;
    }

    printf("connection ready after select with the socket: %d\n", sockfd);
    fcntl(sockfd, F_SETFL, fdopt);
    return sockfd;
}

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi(argv[2]);
    int sockfd = unblock_connect(ip, port, 10);
    if(sockfd < 0) return 1;
    close(sockfd);
    return 0;
}

　　很多服務器要一邊處理網絡連接一邊處理用戶輸入，比如聊天室程序，這樣的就可以用I/O復用來實現，我們以一個poll實現的聊天室程序來舉例說明一下：

　　首先看下客戶端：

/*************************************************************************
    > File Name: 9-6.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Sun 04 Feb 2018 08:27:40 PM PST
 ************************************************************************/

#define _GNU_SOURCE 1
#include"head.h"
using namespace std;

#define BUFFER_SIZE 64

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi(argv[2]);

    struct sockaddr_in server_address;
    bzero(&server_address, sizeof(server_address));
    server_address.sin_family = AF_INET;
    server_address.sin_port = htons(port);
    inet_pton(AF_INET, ip, &server_address.sin_addr);

    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(sockfd >= 0);
    if(connect(sockfd, (struct sockaddr*)&server_address, sizeof(server_address)) < 0) {
        printf("connection failed\n");
        close(sockfd);
        return 1;
    }
    pollfd fds[2];
    fds[0].fd = 0;
    fds[0].events = POLLIN;
    fds[0].revents = 0;
    fds[1].fd = sockfd;
    fds[1].events = POLLIN | POLLRDHUP;
    fds[1].revents = 0;
    char read_buf[BUFFER_SIZE];
    int pipefd[2];
    int ret = pipe(pipefd);
    assert(ret != -1);

    while(1) {
        ret = poll(fds, 2, -1);
        if(ret < 0) {
            printf("poll failure\n");
            break;
        }
        
        if(fds[1].revents & POLLRDHUP) {
            printf("server close the connection\n");
            break;
        }
        else if(fds[1].revents & POLLIN) {
            memset(read_buf, 0, sizeof(read_buf));
            recv(fds[1].fd, read_buf, BUFFER_SIZE, 0);
            printf("%s\n", read_buf);
        }

        if(fds[0].revents & POLLIN) {
            ret = splice(0, NULL, pipefd[1], NULL, 32768, SPLICE_F_MORE | SPLICE_F_MOVE);
            ret = splice(pipefd[0], NULL, sockfd, NULL, 32768, SPLICE_F_MORE | SPLICE_F_MOVE);
        }
    }
    close(sockfd);
    return 0;
}

　　再看下服務器端：

/*************************************************************************
    > File Name: 9-7.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Sun 04 Feb 2018 10:35:16 PM PST
 ************************************************************************/

#define _GNU_SOURCE 1
#include"head.h"
using namespace std;

#define USER_LIMIT 5
#define BUFFER_SIZE 64
#define FD_LIMIT 65535

struct client_data {
    sockaddr_in address;
    char* write_buf;
    char buf[BUFFER_SIZE];
};

int setnonblocking(int fd) {
    int old_option = fcntl(fd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(fd, F_SETFL, new_option);
    return old_option;
}

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi(argv[2]);

    int ret = 0;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    address.sin_port = htons(port);
    inet_pton(AF_INET, ip, &address.sin_addr);

    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(listenfd >= 0);
    ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
    assert(ret != -1);

    ret = listen(listenfd, 5);
    assert(ret != -1);

    client_data* users = new client_data[FD_LIMIT];
    pollfd fds[USER_LIMIT + 1];
    int user_counter = 0;
    for(int i = 1; i <= USER_LIMIT; i ++) {
        fds[i].fd = -1;
        fds[i].events = 0;
    }
    fds[0].fd = listenfd;
    fds[0].events = POLLIN | POLLERR;
    fds[0].revents = 0;

    while(1) {
        ret = poll(fds, user_counter + 1, -1);
        if(ret < 0) {
            printf("poll failure\n");
            break;
        }

        for(int i = 0; i < user_counter + 1;i ++) {
            if((fds[i].fd == listenfd) && (fds[i].revents & POLLIN)) {
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof(client_address);
                int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);

                if(connfd < 0) {
                    printf("errno is: %d\n", errno);
                    continue;
                }

                if(user_counter >= USER_LIMIT) {
                    const char* info = "too many users\n";
                    printf("%s", info);
                    send(connfd, info, sizeof(info), 0);
                    close(connfd);
                    continue;
                }

                user_counter ++;
                users[connfd].address = client_address;
                setnonblocking(connfd);
                fds[user_counter].fd = connfd;
                fds[user_counter].events = POLLIN | POLLERR | POLLRDHUP;
                fds[user_counter].revents = 0;
                printf("comes a new user, now have %d users\n", user_counter);
            }
            else if(fds[i].revents & POLLERR) {
                printf("get an error from %d\n", fds[i].fd);
                char errors[100];
                memset(errors, 0, sizeof(errors));
                socklen_t length = sizeof(errors);
                if(getsockopt(fds[i].fd, SOL_SOCKET, SO_ERROR, &errors, &length) < 0) {
                    printf("get socket option failed\n");
                }
                continue;
            }
            else if(fds[i].revents &POLLRDHUP) {
                users[fds[i].fd] = users[fds[user_counter].fd];
                close(fds[i].fd);
                fds[i] = fds[user_counter];
                user_counter --;
                printf("a client left\n");
            }
            else if(fds[i].revents & POLLIN) {
                int connfd = fds[i].fd;
                memset(users[connfd].buf, '\0', BUFFER_SIZE);
                ret = recv(connfd, users[connfd].buf, BUFFER_SIZE, 0);
                if(ret < 0) {
                    if(errno != EAGAIN) {
                        close(connfd);
                        users[fds[i].fd] = users[fds[user_counter].fd];
                        fds[i] = fds[user_counter];
                        user_counter --;
                        i --;
                    }
                }
                else if(ret == 0) {}
                else {
                    for(int j = 1; j <= user_counter; j ++) {
                        if(fds[j].fd == connfd) continue;
                        fds[j].events |= ~POLLIN;
                        fds[j].events |= POLLOUT;
                        users[fds[j].fd].write_buf = users[connfd].buf;
                    }
                }
            }
            else if(fds[i].revents & POLLOUT) {
                int connfd = fds[i].fd;
                if(!users[connfd].write_buf) continue;
                ret = send(connfd, users[connfd].write_buf, strlen(users[connfd].write_buf), 0);
                users[connfd].write_buf = NULL;

                fds[i].events |= ~POLLOUT;
                fds[i].events |= POLLIN;
            }
        }
    }
    delete []users;
    close(listenfd);
    return 0;
}

　　對於這個程序，其難點只有一個，就是當用戶退出的時候，我們將整個用戶列表的最后一項移動到退出用戶的這一項，可以保證前user_counter項都是有客戶的。但是對於執行完POLLIN事件后，我們取消注冊了POLLIN，轉而注冊了POLLOUT，在我的理解下可能有些問題。我們考慮一個極端情況，如果取消注冊了POLLIN，注冊POLLOUT后，主線程轉而處理下一個poll事件，但此時剛剛處理過的文件描述符上由發來了數據，而由於此時POLLIN事件已經被取消注冊，那么就不會在下一輪輪詢中被觸發POLLIN。 我將代碼中下一步的else if改為了if，這樣判斷完POLLIN事件后會在判斷一次POLLOUT事件，經測試沒有問題。但是仍有待商榷，暫留疑問待日后解決。

　　最后，I/O復用還可以用於同時監聽TCP和UDP服務：

/*************************************************************************
    > File Name: 9-8.cpp
    > Author: Torrance_ZHANG
    > Mail: [e-mail]
    > Created Time: Tue 06 Feb 2018 03:23:19 AM PST
 ************************************************************************/

#include"head.h"
using namespace std;

#define MAX_EVENT_NUMBER 1024
#define TCP_BUFFER_SIZE 512
#define UDP_BUFFER_SIZE 1024

int setnonblocking(int fd) {
    int old_option = fcntl(fd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(fd, F_SETFL, new_option);
    return old_option;
}

void addfd(int epollfd, int fd) {
    epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN | EPOLLET;
    epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
    setnonblocking(fd);
}

int main(int argc, char** argv) {
    if(argc <= 2) {
        printf("usage: %s ip_address port_number\n", basename(argv[0]));
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi(argv[2]);

    int ret = 0;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    inet_pton(AF_INET, ip, &address.sin_addr);
    address.sin_port = htons(port);

    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    assert(listenfd >= 0);

    ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address));
    assert(ret != -1);

    ret = listen(listenfd, 5);
    assert(ret != -1);

    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    inet_pton(AF_INET, ip, &address.sin_addr);
    address.sin_port = htons(port);
    int udpfd = socket(AF_INET, SOCK_DGRAM, 0);
    assert(udpfd >= 0);

    ret = bind(udpfd, (struct sockaddr*)&address, sizeof(address));
    assert(ret != -1);
    epoll_event events[MAX_EVENT_NUMBER];
    int epollfd = epoll_create(5);
    assert(epollfd != -1);

    addfd(epollfd, listenfd);
    addfd(epollfd, udpfd);

    while(1) {
        int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
        if(number < 0) {
            printf("epoll failure\n");
            break;
        }
        for(int i = 0; i < number; i ++) {
            int sockfd = events[i].data.fd;
            if(sockfd == listenfd) {
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof(client_address);
                int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlength);
                addfd(epollfd, connfd);
            }
            else if(sockfd == udpfd) {
                char buf[UDP_BUFFER_SIZE];
                memset(buf, 0, UDP_BUFFER_SIZE);
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof(client_address);

                ret = recvfrom(udpfd, buf, UDP_BUFFER_SIZE, 0, (struct sockaddr*)&client_address, &client_addrlength);
                if(ret > 0) {
                    sendto(udpfd, buf, UDP_BUFFER_SIZE, 0, (struct sockaddr*)&client_address, client_addrlength);
                }
            }
            else if(events[i].events & EPOLLIN) {
                char buf[TCP_BUFFER_SIZE];
                while(1) {
                    memset(buf, 0, TCP_BUFFER_SIZE);
                    ret = recv(sockfd, buf, TCP_BUFFER_SIZE, 0);
                    if(ret < 0) {
                        if((errno == EAGAIN) || (errno == EWOULDBLOCK)) break;
                        close(sockfd);
                        break;
                    }
                    else if(ret == 0) {
                        close(sockfd);
                    }
                    else send(sockfd, buf, ret, 0);
                }
            }
        }
    }
    close(listenfd);
    return 0;
}