六十、linux 编程—— I/O 多路复用 select

60.1 介绍

　　

　　

　　

　　

　　

60.2 例子

　　

　　echo_tcp_server_select.c

#include <netdb.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <memory.h>
#include <signal.h>
#include <fcntl.h>
#include <time.h>
#include <arpa/inet.h>
#include <errno.h>
#include <pthread.h>
#include "vector_fd.h"

vector_fd *vfd;
int sockfd;

void sig_handler(int signo)
{
    if(signo == SIGINT){
        printf("server close
");
        /** 步骤6: 关闭 socket */
        close(sockfd);
        /** 销毁动态数组 */
        destroy_vector_fd(vfd);
        exit(1);
    }
}

/**
 * fd 对应于某个连接的客户端，和某一个连接的客户端进行双向通信(非阻塞方式)
 */
void do_service(int fd)
{
    char buff[512];
    memset(buff, 0, sizeof(buff));

    /**
     *  因为采用非阻塞方式，若读不到数据直接返回，
     *  直接服务于下一个客户端,
     *  因此不需要判断 size < 0 的情况 */
    ssize_t size = read(fd, buff, sizeof(buff));

    if(size == 0){
        /** 客户端已经关闭连接 */
        printf("client closed
");
        /** 从动态数组中删除对应的 fd */
        remove_fd(vfd, fd);
        /** 关闭对应客户端的 socket */
        close(fd);
    }
    else if(size > 0){
        printf("%s
", buff);
        if(write(fd, buff, size) < 0){
            if(errno == EPIPE){
                /** 客户端关闭连接 */
                perror("write error");
                remove_fd(vfd, fd);
                close(fd);
            }
            perror("protocal error");
        }
    }
}

void out_addr(struct sockaddr_in *clientaddr)
{
    char ip[16];
    memset(ip, 0, sizeof(ip));
    int port = ntohs(clientaddr->sin_port);
    inet_ntop(AF_INET, &clientaddr->sin_addr.s_addr, ip, sizeof(ip));
    printf("%s(%d) connected!
", ip, port);
}

/** 遍历出动态数组中所有的描述符并加入到描述符集 set
 * 中，同时此函数返回动态数组中最大的那个描述符 */
int add_set(fd_set *set)
{
    FD_ZERO(set);   ///< 清空描述符集
    int max_fd = vfd->fd[0];
    int i = 0;
    for(; i < vfd->counter; i++){
        int fd = get_fd(vfd, i);
        if(fd > max_fd) max_fd = fd;
        FD_SET(fd, set);    ///< 将 fd 加入到描述符集中
    }

    return max_fd;
}

void *th_fn(void *arg)
{
    /** 设置超时时间 2s */
    struct timeval t;
    t.tv_sec = 2;
    t.tv_usec = 0;

    int n = 0;
    int maxfd;
    fd_set set; ///< 描述符集
    maxfd = add_set(&set);

    /**
     * 调用 select 函数会阻塞，委托内核去检查传入的描述符是否准备好，
     * 若有则返回准备好的描述符；超时则返回 0
     * 第一个参数为描述符集中的描述符的范围(最大描述符 + 1)
     */
    while((n = select(maxfd + 1, &set, NULL, NULL, &t)) >= 0){
        /** 检测哪些描述符准备好, 并和这些准备好的描述符对应的客户端进行数据的双向通信 */
        if(n > 0){
            int i = 0;
            for(; i < vfd->counter; i++){
                int fd = get_fd(vfd, i);
                if(FD_ISSET(fd, &set)){
                    do_service(fd);
                }
            }
        }
        /** 重新设置时间和清空描述符集 */
        t.tv_sec = 2;
        t.tv_usec = 0;
        /** 重新遍历动态数组中最新的描述符放置到描述符集中 */
        maxfd = add_set(&set);
    }

    return (void *)0;
}

int main(int argc, char *argv[])
{
    if(argc < 2){
        printf("usage: %s #port
", argv[0]);
        exit(1);
    }

    if(signal(SIGINT, sig_handler) == SIG_ERR){
        perror("signal sigint error");
        exit(1);
    }


    /** 步骤1: 创建 socket(套接字)
     *  注: socket 创建在内核中，是一个结构体.
     *  AF_INET: IPV4
     *  SOCK_STREAM: tcp 协议
     *  AF_INET6: IPV6
     */
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if(sockfd < 0){
        perror("socket error");
        exit(1);
    }

    /**
     * 步骤2: 调用 bind 函数将 socket 和地址(包括 ip、port)进行绑定
     */
    struct sockaddr_in  serveraddr;
    memset(&serveraddr, 0, sizeof(struct sockaddr_in));
    /** 往地址中填入 ip、port、internet 地址族类型 */
    serveraddr.sin_family = AF_INET;    ///< IPV4
    serveraddr.sin_port = htons(atoi(argv[1])); ///< 填入端口
    serveraddr.sin_addr.s_addr = INADDR_ANY; ///< 填入 IP 地址
    if(bind(sockfd, (struct sockaddr *)&serveraddr, sizeof(struct sockaddr_in))){
        perror("bind error");
        exit(1);
    }

    /**
     *  步骤3: 调用 listen 函数启动监听(指定 port 监听)
     *         通知系统去接受来自客户端的连接请求
     *         (将接受到的客户端连接请求放置到对应的队列中)
     *  第二个参数: 指定队列的长度
     */
    if(listen(sockfd, 10) < 0){
        perror("listen error");
        exit(1);
    }

    /** 创建放置套接字描述符 fd 的动态数组 */
    vfd = create_vector_fd();

    /** 设置线程的分离属性 */
    pthread_t th;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    int err;
    if((err = pthread_create(&th, &attr, th_fn, (void *)0)) != 0){
        perror("pthread create error");
        exit(1);
    }
    pthread_attr_destroy(&attr);

    /**
     * 1)主控线程获得客户端的链接，将新的 socket 描述符放置到动态数组中
     * 2) (a)启动的子线程调用 select 函数委托内核去检查传入到 select
     *       中的描述符是否准备好.
     *    (b)利用 FD_ISSET 来找出准备好的那些描述符,
     *       并和对应的客户端进行双向通信(非阻塞)
     */
    struct sockaddr_in clientaddr;
    socklen_t len = sizeof(clientaddr);

    while(1){
        /**
         *  步骤4: 调用 accept 函数从队列中获得一个客户端的请求连接, 并返回新的
         *         socket 描述符
         *  注意:  若没有客户端连接，调用此函数后会阻塞, 直到获得一个客户端的连接
         */
        /** 主控线程负责调用 accept 去获得客户端的连接 */
        int fd = accept(sockfd, (struct sockaddr *)&clientaddr, &len);
        if(fd < 0){
            perror("accept error");
            continue;
        }

        out_addr(&clientaddr);

        /** 将返回的新的 socket 描述符加入到动态数组中 */
        add_fd(vfd, fd);
    }

    return 0;
}

　　编译运行测试：