五十九、linux 编程—— I/O 多路复用 fcntl

59.1 介绍

　　前面介绍的函数如，recv、send、read 和 write 等函数都是阻塞性函数，若资源没有准备好，则调用该函数的进程将进入阻塞状态。我们可以使用 I/O 多路复用来解决此问题（即解决并发）。

• I/O 多路复用的方式主要有两种实现方法
  • fcntl 函数实现（非阻塞方式）
  • select 函数实现

59.1.1 fcntl 非阻塞方式——I/O多路复用/转换

　　

59.2 例子

 59.2.1 动态数组模块

 vector_fd.c

#include <malloc.h>
#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <memory.h>
#include "vector_fd.h"

static void encapacity(vector_fd *vfd)
{
    if(vfd->counter >= vfd->max_counter){
        int *fds = (int *)calloc(vfd->counter + 5, sizeof(int));
        assert(fds != NULL);
        memcpy(fds, vfd->fd, sizeof(int) * vfd->counter);
        free(vfd->fd);
        vfd->fd = fds;
        vfd->max_counter += 5;
    }
}

static int indexof(vector_fd *vfd, int fd)
{
    int i = 0;
    for(; i < vfd->counter; i++){
        if(vfd->fd[i] == fd) return i;
    }

    return -1;
}


vector_fd *create_vector_fd(void)
{
    vector_fd *vfd = (vector_fd *)calloc(1, sizeof(vector_fd));
    assert(vfd != NULL);

    vfd->fd = (int *)calloc(5, sizeof(int));
    assert(vfd->fd != NULL);
    vfd->counter = 0;
    vfd->max_counter = 0;

    return vfd;
}


void destroy_vector_fd(vector_fd *vfd)
{
    assert(vfd != NULL);
    free(vfd->fd);
    free(vfd);
}

int get_fd(vector_fd *vfd, int index)
{
    assert(vfd != NULL);
    if(index < 0 || index > vfd->counter - 1) return 0;
    
    return vfd->fd[index];
}

void remove_fd(vector_fd *vfd, int fd)
{
    assert(vfd != NULL);
    int index = indexof(vfd, fd);
    if(index == -1) return;
    int i = index;
    for(; i < vfd->counter - 1; i++){
        vfd->fd[i] = vfd->fd[i + 1];
    }
    vfd->counter--;
}

void add_fd(vector_fd *vfd, int fd)
{
    assert(vfd != NULL);
    encapacity(vfd);
    vfd->fd[vfd->counter++] = fd;
}

vector_fd.h

#ifndef __VECTOR_FD_H__
#define __VECTOR_FD_H__

typedef struct {
    int     *fd;
    int     counter;
    int     max_counter;
}vector_fd;

extern vector_fd *create_vector_fd(void);
extern void destroy_vector_fd(vector_fd *);
extern int get_fd(vector_fd *, int index);
extern void remove_fd(vector_fd *, int fd);
extern void add_fd(vector_fd *, int fd);

#endif

　　编译成模块：gcc -o obj/vector_fd.o -Iinclude -c src/vector_fd.c

59.2.2 服务器端

　　echo_tcp_server_fcntl.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){
        /** 客户端已经关闭连接 */
        char info[] = "client closed";
        write(STDOUT_FILENO, info, sizeof(info));
        /** 从动态数组中删除对应的 fd */
        remove_fd(vfd, fd);
        /** 关闭对应客户端的 socket */
        close(fd);
    }
    else if(size > 0){
        write(STDOUT_FILENO, buff, sizeof(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);
}

void *th_fn(void *arg)
{
    int i;
    while(1){
        i = 0;
        /** 遍历动态数组中的 socket 描述符 */
        for(; i < vfd->counter; i++){
            do_service(get_fd(vfd, i));
        }
    }

    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)启动的子线程负责遍历动态数组中 socket
     *   描述符，并和对应的客户端进行双向通信(采用非阻塞方式读写)
     */
    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);

        /** 将读写修改为非阻塞方式 */
        int val;
        fcntl(fd, F_GETFL, &val);   ///< 获取原来的状态标志
        val |= O_NONBLOCK;
        fcntl(fd, F_SETFL, val);    ///< 添加新的状态标志

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

    }

    return 0;
}

　　编译：

　　gcc -o bin/echo_tcp_server_fcntl -Iinclude obj/vector_fd.o src/echo_tcp_server_fcntl.c -lpthread

59.2.3 客户端

　　echo_tcp_client_fcntl.c

#include <sys/types.h>
#include <stdlib.h>
#include <stdio.h>
#include <memory.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netdb.h>
#include <signal.h>
#include <string.h>
#include <time.h>
#include <arpa/inet.h>


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

    /** 步骤1: 创建 socket */
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if(sockfd < 0){
        perror("socket error");
        exit(1);
    }

    /** 往 serveraddr 中填入 ip、port 和地址族类型(ipv4) */
    struct sockaddr_in serveraddr;
    memset(&serveraddr, 0, sizeof(struct sockaddr_in));
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_port = htons(atoi(argv[2]));
    /** 将 ip 地址转换成网络字节序后填入 serveraddr 中  */
    inet_pton(AF_INET, argv[1], &serveraddr.sin_addr.s_addr);

    /**
     *  步骤2: 客户端调用 connect 函数连接到服务器端
     */
    if(connect(sockfd, (struct sockaddr *)&serveraddr, sizeof(struct sockaddr_in)) < 0){
        perror("connect error");
        exit(1);
    }

    /** 步骤3: 调用 IO 函数(read/write)和服务器端进行双向通信 */
    char buff[512];
    ssize_t size;
    char *prompt = "==>";
    while(1){
        memset(buff, 0, sizeof(buff));
        write(STDOUT_FILENO, prompt, 3);
        size = read(STDIN_FILENO, buff, sizeof(buff));
        if(size < 0) continue;
        buff[size - 1] = '';

        if(write(sockfd, buff, sizeof(buff)) < 0){
            perror("write msg error");
            continue;
        }
        else {
            if(read(sockfd, buff, sizeof(buff)) < 0){
                perror("read msg error");
                continue;
            }
            else {
                printf("%s
", buff);
            }
        }
    }

    /** 步骤4: 关闭 socket */
    close(sockfd);

    return 0;
}

　　编译：gcc -o bin/echo_tcp_client_fcntl src/echo_tcp_client_fcntl.c

59.2.4 测试运行