Linux Socket 编程I/O Mutiplexing poll 和 epoll


上一篇介绍了select的基本用法（见  ），接着来学习一下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 进行完善。

  • 1
  • 2
  • 下一页

相关内容