Linux Socket 編程I/O Mutiplexing poll 和 epoll

上一篇介紹了select的基本用法（見 http://www.linuxidc.com/Linux/2014-03/97442.htm ），接著來學習一下poll和epoll的基本用法。首先來看poll：

#include <sys/poll.h>

int poll (struct pollfd *fds, unsigned int nfds, int timeout);

　　poll() 采用了struct pollfd 結構數組來保存關心的文件描述符，而不是像select一樣使用三個fd_set ，pollfd結構體定義如下：

struct pollfd {

    int fd; /* file descriptor */

    short events; /* requested events to watch */

    short revents; /* returned events witnessed */

};

　　每一個pollfd結構體指定了一個被監視的文件描述符，fds數組中可以存放多個pollfd結構，而且數量不會像select的FD_SETSIZE一樣被限制在1024或者2048 。數組中每個pollfd結構體的events域是監視該文件描述符的事件掩碼，由用戶來設置這個域。revents域是文件描述符的操作結果事件掩碼，系統調用返回時設置這個域。events域中請求的任何事件都可能在revents域中返回。我們可以設置如下事件：

　　POLLIN：有數據可讀。

　　POLLRDNORM：有普通數據可讀。

　　POLLRDBAND：有優先數據可讀。

　　POLLPRI：有緊迫數據可讀。

　　------------------------------------------------------------

　　POLLOUT：寫數據不會導致阻塞。

　　POLLWRNORM：寫普通數據不會導致阻塞。

　　POLLWRBAND：寫優先數據不會導致阻塞。

　　此外，revents域中還可能返回下列事件：

　　POLLERR：指定的文件描述符發生錯誤。

　　POLLHUP：指定的文件描述符掛起事件。

　　POLLNVAL：指定的文件描述符非法。

　　注意：只能作為描述字的返回結果存儲在revents中，而不能作為測試條件用於events中。

　　其中POLLIN | POLLPRI等價於select()的讀事件，POLLOUT | POLLWRBAND等價於select()的寫事件。POLLIN等價於POLLRDNORM | POLLRDBAND，而POLLOUT則等價於POLLWRNORM。假如，要同時監視一個文件描述符是否可讀和可寫，我們可以設置events為POLLIN | POLLOUT。在poll返回時，我們可以檢查revents中的標志，對應於文件描述符請求的events結構體。如果POLLIN事件被設置，則文件描述符可以被讀取而不阻塞。如果POLLOUT被設置，則文件描述符可以寫入而不導致阻塞。這些標志並不是互斥的：它們可能被同時設置，表示這個文件描述符的讀取和寫入操作都會正常返回而不阻塞。

　　timeout參數指定等待的毫秒數，無論I/O是否准備好，超時時間一到poll都會返回。timeout指定為負數值表示無限超時，UNPv1 中使用的INFTIM 宏貌似現在已經廢棄，因此如果要設置無限等待，直接將timeout賦值為-1；timeout為0指示poll調用立即返回並列出准備好I/O的文件描述符，但並不等待其它的事件。

　　成功時，poll()返回結構體中revents域不為0的文件描述符個數；如果在超時前沒有任何事件發生，poll()返回0；失敗時，poll()返回-1。

//pollEcho.cpp
#include <stdio.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <vector>
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <errno.h>
#include <poll.h>
#include <stropts.h>
#include <netdb.h>

#define PORT 1314
#define MAX_LINE_LEN 1024

int main()
{
    struct sockaddr_in cli_addr, server_addr;
    socklen_t addr_len;
    int one,flags,nrcv,nwrite,nready;
    
    int listenfd,connfd;
    char buf[MAX_LINE_LEN],addr_str[INET_ADDRSTRLEN];
    
    std::vector<struct pollfd> pollfdArray;
    struct pollfd pfd;
    
    bzero(&server_addr, sizeof server_addr);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(PORT);
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    
    if( listenfd < 0)
    {
        printf("listen error: %s \n", strerror(errno));
        exit(1);
    }
       
    one = 1;
    setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR, &one, sizeof one);
    
    flags = fcntl(listenfd,F_GETFL,0);
    fcntl(listenfd, F_SETFL, flags | O_NONBLOCK);
    
    if(bind(listenfd,reinterpret_cast<struct sockaddr *>(&server_addr),sizeof(server_addr)) < 0)
    {
        printf("bind error: %s \n", strerror(errno));
        exit(1);
    }
    
    listen(listenfd, 100);
    
    pfd.fd = listenfd;
    pfd.events = POLLIN;
    
    pollfdArray.push_back(pfd);
    
    while(1)
    {
        nready = poll(&(*pollfdArray.begin()), pollfdArray.size(), -1);
        
        if( nready < 0)
        {
            printf("poll error: %s \n", strerror(errno));
        }
        
        if( pollfdArray[0].revents & POLLIN)
        {
            addr_len = sizeof cli_addr;
            connfd = accept(listenfd, reinterpret_cast<struct sockaddr *>(&cli_addr), &addr_len);
            
            if( connfd < 0)
            {
                if( errno != ECONNABORTED || errno != EWOULDBLOCK || errno != EINTR)
                {
                    printf("accept error: %s \n", strerror(errno));
                    continue;
                }
            }
            
            printf("recieve from : %s at port %d\n", inet_ntop(AF_INET,&cli_addr.sin_addr,addr_str,INET_ADDRSTRLEN),cli_addr.sin_port);
            
            flags = fcntl(connfd, F_GETFL, 0);
            fcntl(connfd,F_SETFL, flags | O_NONBLOCK);
            
            bzero(&pfd, sizeof pfd);
            
            pfd.fd = connfd;
            pfd.events = POLLIN;
            
            pollfdArray.push_back(pfd);
            
            if(--nready < 0)
            {
                continue;
            }
            
        }
        
        for( unsigned int i = 1; i < pollfdArray.size(); i++) // i from 1 not 0
        {
            pfd = pollfdArray[i];
            
            if(pfd.revents & (POLLIN | POLLERR))
            {
                memset(buf, 0, MAX_LINE_LEN);
                if( (nrcv = read(pfd.fd, buf, MAX_LINE_LEN)) < 0)
                {
                    if(errno != EWOULDBLOCK || errno != EAGAIN || errno != EINTR)
                    {
                        printf("read error: %s\n",strerror(errno));
                    }
                }
                else if( 0 == nrcv)
                {
                    close(pfd.fd);
                    pollfdArray.erase(pollfdArray.begin() + i);
                }
                else
                {
                    printf("nrcv: %s\n",buf);
                    nwrite = write(pfd.fd, buf, nrcv);
                    if( nwrite < 0)
                    {
                        if(errno != EAGAIN || errno != EWOULDBLOCK)
                            printf("write error: %s\n",strerror(errno));
                    }
                    printf("nwrite = %d\n",nwrite);
                }
                
            }
        }
    }
    return 0;
}

　　以上代碼操作的文件描述符都設置成為了非阻塞的狀態，這也是為了更好的配合I/O multiplexing 的執行，試想如果read 或者 write 阻塞在某個描述符上，I/O multiplexing 就失去了真正的意義了，因為此時select/poll 函數就無法處理其它描述符產生的事件了。但是只要設置為非阻塞就夠了嗎？ 這顯然是還不夠的，後面會專門寫一篇文章對非阻塞的I/O multiplexing 進行完善。