[系统运维]网络编程之IO、 poll、 epoll模型

网络编程的目的是为了解决机器间的通信问题。

1.常规API实现的基础IO

    int listenfd, connfd, n;
    struct sockaddr_in servaddr;
    char buff[MAXLNE];

    if( (listenfd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        printf("create socket failed error: %s(errno:%d)\n", strerror(errno), errno);
        return 0;
    }

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(9999);

    if(bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1) {
        printf("bind socket error: %s(errorno:%d)\n", strerror(errno), errno);
        return 0;
    }

    if(listen(listenfd, 10) == -1) {
        printf("listen socket error:%s(errno:%d)\n", strerror(errno), errno);
        return 0;
    }
   

    struct sockaddr_in client;
    socklen_t len = sizeof(client);
    if((connfd = accept(listenfd, (struct sockaddr*)&client, &len)) == -1)
    {
        printf("accept socket error:%s(errno:%d)\n", strerror(errno), errno);
        return 0;
    }

    printf("=========waitting for client's request============\n");
    while (1)
    {
        n = recv(connfd, buff, MAXLNE, 0);

        if(n > 0)
        {
            buff[n] = '\0';
            printf("recv msg from client:%s\n", buff);

            send(connfd, buff, n, 0);
        }
        else if(n == 0)
        {
            // client close
            close(connfd);
            printf("client closed....\n");
            break;
        }
    }

这种方式只适用于1对1的CS模型，多台client发起连接是无法进行正常交互通信的。但可以正常发起连接，因为连接的成功不由API的执行结果决定， 在服务端进入listen状态之后，三次握手的过程由协议栈完成，在应用层是通过accept取出完成三次握手后产生的tcb后对clientfd再进行操作。所以只要服务端处于开启状态，连接都能正常发起，但无法进行通讯。因为此时无法通过accpet取出tcb，只会在recv和send之间切换状态。

2.一连接一线程

void *client_routine(void* arg)
{
    int fd = *(int*)arg;

    char buff[MAXLNE];

    while (1)
    {
        /* code */
        int n = recv(fd, buff, MAXLNE, 0);
        if(n > 0)
        {
            buff[n] = '\0';
            printf("recv msg from client: %s\n", buff);

            send(fd, buff, n, 0);
        }        
        else if(0 == n)
        {
            close(fd);
            break;
        }
    }
    
    return NULL;
}

。。。 

while(1)
    {
        struct sockaddr_in client;
        socklen_t len = sizeof(client);
        if((connfd = accept(listenfd, (struct sockaddr*)&client, &len)) == -1)
        {
            printf("accept socket error:%s(errno:%d)\n", strerror(errno), errno);
            return 0;
        }

        pthread_t myThreadID;
        pthread_create(&myThreadID, NULL, client_routine, (void *)&connfd);
    }

在上面实现方式的基础上可以针对其无法响应accpet的缺陷进行修改，方式是进程中只对listenfd进行监听，成功accept后，即获取到了可以进行IO通信的fd（文件描述符,windows也称为套接字）。以此fd为参数创建一个线程，该线程中对指定的IO进行读写操作。

在客户端数量较少的情况下，这种方式是比较适合的。但是由于线程本身是需要占用资源的，所以这种这种方式最大的弊端就是无法突破内存的限制，即C10K 问题， 如果一个线程占用4m资源， 对于进程拥有的4G内存来说，4 * 1024 / 4 = 1024个线程。 继续创建线程的话会导致宕机。

3.多路复用组件-select

/*****select*****

int select(int maxfdp1,fd_set *readset,fd_set *writeset,fd_set *exceptset,const struct timeval *timeout);

【参数说明】
int maxfdp1
指定待测试的文件描述字个数，它的值是待测试的最大描述字加1。(最大描述符即最新产生的fd)

fd_set
fd_set可以理解为一个集合，这个集合中存放的是文件描述符(file descriptor)，即文件句柄。中间的三个参数指定我们要让内核测试读、写和异常条件的文件描述符集合。如果对某一个的条件不感兴趣，就可以把它设为空指针。

timeout
timeout告知内核等待所指定文件描述符集合中的任何一个就绪可花多少时间。其timeval结构用于指定这段时间的秒数和微秒数。
【返回值】
int 若有就绪描述符返回其数目，若超时则为0，若出错则为-1

FD_SET(connfd, &rfds);
FD_SET(connfd, &wfds);
FD_CLR(i, &rfds);

*****************/

    // 采用轮询的方式处理fd ， posix中定义 __FD_SETSIZE = 1024 表示一个select 能处理1024个fd

    fd_set rfds, rset, wfds, wset;

    FD_ZERO(&rfds);                 // clear set
    FD_SET(listenfd, &rfds);        // add listenfd to set(rfds)

    FD_ZERO(&wfds);

    int max_fd = listenfd;

    while (1)
    {
        rset = rfds;
        wset = wfds;

        int nready = select(max_fd+1, &rset, &wset, NULL, NULL);
     
        if(FD_ISSET(listenfd, &rset))         
        {
            struct sockaddr_in client;
            socklen_t len = sizeof(client);

            if((connfd = accept(listenfd, (struct sockaddr *)&client, &len)) == -1)
            {
                printf("accept socket error:%s(errno:%d)\n", strerror(errno), errno);
                return 0;
            }

            FD_SET(connfd, &rfds);        

            if(connfd > max_fd) max_fd = connfd;
            
            if(--nready == 0) continue;
        }

        int i = 0;
        for(i = listenfd+1; i <= max_fd; i++)            
        {
            if(FD_ISSET(i, &rset))                 
            {
                n = recv(i, buff, MAXLNE, 0);
                if(n > 0)
                {
                    buff[n] = '\0';

                    printf("recv msg fron client: %s\n", buff);

                    FD_SET(i, &wfds);
                }
                else if(0 == n)
                {
                    FD_CLR(i, &rfds);
                    printf("client closed.....\n");
                    close(i);
                }

                if(--nready == 0)   break;
            }
            else if(FD_ISSET(i, &wset))        
            {
                send(i, buff, n, 0);
                FD_SET(i, &rfds);
            }
        }
    
    }
    

根据其定义可以了解到，select组件是通过对指定fd的集合进行监听来实现IO复用。通过将某一个fd移动到可读描述符集合或者可写描述符集合，来实现对应的可读状态和可写状态的监听。select的源码定义中限定一个select对象可处理1024个fd，可以通过重新编译源码进行改写，但不建议这样做，建议通过多线程创建多个select对象，或者创建多个进程，通过这两种方式，可以直接突破1024的限制，这种方式也解决了单线程单链接所存在的C10K问题。

4.poll组件

struct pollfd fds[POLL_SIZE] = {0};
    fds[0].fd = listenfd;
    fds[0].events = POLLIN;

    int max_fd = listenfd;            // 将最新创建的fd作为max_fd
    int i = 0;

    for(i = 1; i < POLL_SIZE; i++)
    {
        fds[i].fd = -1;
    }   

    while(1)
    {
        int nready = poll(fds, max_fd+1, -1);     // poll(fds, POLL_SIZE, TIME_DELAY);       
       
        if(fds[0].revents & POLLIN)
        {
            struct sockaddr_in client;
            socklen_t len = sizeof(client);

            if((connfd = accept(listenfd, (struct sockaddr*)&client, &len)) == -1)
            {
                printf("accept socket error:%s(errno%d)\n", strerror(errno), errno);
                return 0;
            }

            printf("accept a client....");

            fds[connfd].fd = connfd;
            fds[connfd].events = POLLIN;
            
            if(connfd > max_fd) max_fd = connfd;

            if(--nready == 0)   continue;

        }

        for(i = listenfd + 1; i <= max_fd; i++)
        {
            if(fds[i].revents * POLLIN)
            {
                // 可读
                n = recv(i, buff, MAXLNE, 0);
                if(n > 0)
                {
                    buff[n] = '\0';
                    printf("recv msg from client: %s\n", buff);

                    send(i, buff, n, 0);
                }    
                else if(0 == n)
                {
                    fds[i].fd = -1;
                    close(i);
                }

                if(--nready == 0)   break;
            }
        }

    }

select组件对于一个fd来讲，只能指定监听其某一个状态，需要根据不同的状态进行 切换监听。poll组件对此进行了改进，poll在对整个描述符集合监听时，会返回每个fd的实际状态，即各自是否可读或可写， 用户只需要对fd进行遍历，若该fd处于触发状态，直接进行业务处理即可。

5.epoll

 int  epollfd = epoll_create(1);     // 无效参数，兼容旧接口；

    struct epoll_event events[POLL_SIZE] = {0}; 

    struct epoll_event ev;
    ev.data.fd=listenfd;
    ev.events=EPOLLIN|EPOLLET;              

    epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &ev);  

//epoll_ctl() 
//参数1： epoll_create 生成的epoll专用的文件描述符
//参数2：要进行的操作例如注册事件，可能的取值EPOLL_CTL_ADD 注册、EPOLL_CTL_MOD 修改、            EPOLL_CTL_DEL 删除
//参数3：关联的文件描述符
//参数4：指向epoll_event的指针

    while(1)
    {
        int nready = epoll_wait(epollfd, events, POLL_SIZE, 5);         

        if(nready == -1)    continue;

        int i = 0;
        for(i = 0; i < nready; i++)
        {
            int clientfd = events[i].data.fd;

            if(clientfd == listenfd)      
            {
                 struct sockaddr_in client;
                 socklen_t len = sizeof(client);

                 if((connfd = accept(listenfd, (struct sockaddr*) &client, &len)) == -1)
                 {
                     printf("accept socket error:%s(errno:%d)\n", strerror(errno), errno);
                     return 0;
                 }   
                
                // 添加新连接的fd到epollfd中
                ev.events = EPOLLIN;
				ev.data.fd = connfd;
                epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev); 
            }
            else if(events[i].events & EPOLLIN)     // 如果接收到触发事件，则对该fd进行处理
            {
                n = recv(clientfd, buff, MAXLNE, 0);
                if(n > 0)
                {
                    buff[n] = '\0';
                    printf("recv msg from client: %s\n", buff);

                    send(clientfd, buff, n, 0);
                }
                else if(n == 0)
                {
                    printf("client close...\n");

                    ev.events = EPOLLIN;
					ev.data.fd = clientfd;
					epoll_ctl(epollfd, EPOLL_CTL_DEL, clientfd, &ev);		// 删除一个fd

                    close(clientfd);
                }
            }

        }

    }

epoll是一种根据状态机模型实现的io多路复用组件，也是各类框架适应最普遍的IO底层模型。

?实例一个epoll管理对象， epoll_create() ? ? ? ? ? -----> 回调方式进行调度；事件驱动

每次调度将产生触发（LT/ET）的事件放到events集合中? ? ?--> 边缘触发和水平触发的概念

通过对事件关联的fd（events[i].data.fd）以及 所触发的事件（events[i].events）进行区分判断，进行相应的业务处理

过程中可通过epoll_ctl（）对新连接的或者需要断开连接的fd进行添加或删除，减少轮询调度产生的资源消耗