IO多路复用之epoll总结

使用

server端

#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <string.h>

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include <sys/epoll.h>

#define SERVER_PORT 8088
#define EPOLL_MAX_NUM 2048
#define BUFFER_MAX_LENGTH 4096

char buffer[BUFFER_MAX_LENGTH];

void str_toupper(char *str) {
    int i;
    for (i = 0; i < strlen(str); i++) {
        str[i] = toupper(str[i]);
    }
}

int main(int argc, char **argv) {
    int listen_fd, client_fd;
    struct sockaddr_in server_addr;
    struct sockaddr_in client_addr;
    socklen_t client_len;

    int epfd = 0;
    struct epoll_event event, *my_events;

    // socket: 返回一个套接字
    // af: 协议族。AF_INET: 表示IPV4
    // type: socket类型。SOCK_STREAM: 连接类型是双向流, 即TCP
    // protocol: 协议,TCP或UDP等, 等于0时表示根据type字段自动选择对应的协议
    listen_fd = socket(AF_INET, SOCK_STREAM, 0);

    // bind: 将描述符和监听地址绑定
    server_addr.sin_family = AF_INET;
    // htonl: 主机序转换为网络序, 小端转换为大端
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(SERVER_PORT);
    bind(listen_fd, (struct sockaddr *) &server_addr, sizeof(server_addr));

    // listen
    listen(listen_fd, 10);

    // create epoll
    epfd = epoll_create(EPOLL_MAX_NUM);
    if (epfd < 0) {
        perror("epoll create");
        goto END;
    }

    // 注册listen_fd到epoll
    event.events = EPOLLIN;
    event.data.fd = listen_fd;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event) < 0) {
        perror("epoll ctl add listen_fd ");
        goto END;
    }
    printf("successfully epoll_ctl listen fd\n");

    // 接收的事件
    my_events = malloc(sizeof(struct epoll_event) * EPOLL_MAX_NUM);

    // 循环接收事件
    while (1) {
        printf("begin epoll wait...\n");
        // 调用epoll_wait等待 -1表示无限阻塞
        int active_fds_cnt = epoll_wait(epfd, my_events, EPOLL_MAX_NUM, -1);
        for (int i = 0; i < active_fds_cnt; i++) {
            // 监听fd有事件, 说明有新的连接进来, 调用accept接收处理
            if (my_events[i].data.fd == listen_fd) {
                // 后面两个参数用来获取client的ip和端口信息
                client_fd = accept(listen_fd, (struct sockaddr *) &client_addr, &client_len);
                if (client_fd < 0) {
                    perror("accept");
                    continue;
                }

                printf("new connection comes\n");
                printf("IP address is: %s\n", inet_ntoa(client_addr.sin_addr));
                printf("port is: %d\n", (int) ntohs(client_addr.sin_port));

                // 注册客户端的fd到epoll
                event.events = EPOLLIN | EPOLLET;
                event.data.fd = client_fd;
                epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &event);
            } else if (my_events[i].events & EPOLLIN) {
                printf("EPOLLIN\n");
                client_fd = my_events[i].data.fd;

                // do read
                buffer[0] = '\0';
                // 固定读取最大字节数为5
                int n = read(client_fd, buffer, 5);
                if (n < 0) {
                    perror("read");
                    continue;
                } else if (n == 0) { // 连接已关闭
                    printf("close...\n");
                    epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &event);
                    close(client_fd);
                } else {
                    printf("[read]: %s\n", buffer);
                    buffer[n] = '\0';
                    str_toupper(buffer);
                    write(client_fd, buffer, strlen(buffer));
                    printf("[write]: %s \n", buffer);
                    memset(buffer, 0, BUFFER_MAX_LENGTH);
                }
            } else if (my_events[i].events & EPOLLOUT) {
                printf("EPOLLOUT\n");
            }
        }
    }

    END:
    close(epfd);
    close(listen_fd);
    return 0;
}

运行./epoll_demo_server

client端

//
// Created by c00577049 on 2022/7/27.
//

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>

#define MAX_LINE (3)
#define SERVER_PORT (8088)

void set_non_blocking(int fd) {
    int opts = 0;
    opts = fcntl(fd, F_GETFL);
    opts = opts | O_NONBLOCK;
    fcntl(fd, F_SETFL);
}

int main(int argc, char **argv) {
    int sockfd;
    char recvline[MAX_LINE + 1] = {0};

    struct sockaddr_in server_addr;

    if (argc != 2) {
        fprintf(stderr, "usage ./client <SERVER_IP>\n");
        exit(0);
    }

    // 创建socket
    if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        fprintf(stderr, "socket error");
        exit(0);
    }

    // 给server addr赋值
    bzero(&server_addr, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(SERVER_PORT);

    if (inet_pton(AF_INET, argv[1], &server_addr.sin_addr) <= 0) {
        fprintf(stderr, "inet_pton error for %s", argv[1]);
        exit(0);
    }

    // 连接服务端
    if (connect(sockfd, (struct sockaddr *) (&server_addr), sizeof(server_addr)) < 0) {
        perror("connect");
        fprintf(stderr, "connect error\n");
        exit(0);
    }

    set_non_blocking(sockfd);

    char input[100];
    int n = 0;
    int count = 0;


    // 循环从标准输入读取数据
    while (fgets(input, 100, stdin) != NULL) {
        printf("[send] %s\n", input);
        n = 0;
        // 读取的数据发送到服务器
        n = send(sockfd, input, strlen(input), 0);
        if (n < 0) {
            perror("send");
        }

        n = 0;
        count = 0;

        // 读取服务器返回的数据
        while (1) {
            n = read(sockfd, recvline + count, MAX_LINE);
            printf("n====%d\n", n);
            if (n == MAX_LINE) {
                count += n;
                printf("count1==%d\n", count);
                printf("[recv11] %s\n", recvline);
                printf("here......\n");
                continue;
            } else if (n < 0) {
                perror("read");
                break;
            } else {
                count += n;
                printf("count2==%d\n", count);
                recvline[count] = '\0';
                printf("[recv22] %s\n", recvline);
                break;
            }
        }
    }

    return 0;
}

运行./epoll_demo_client 127.0.0.1，输入字符，会返回对应字符的大写。

完整例子

主要接口函数

• int epoll_create ( int size ); 创建epoll实例，返回epoll对应的fd

• int epoll_ctl ( int epfd, int op, int fd, struct epoll_event *event ); 注册监听的事件

  事件类型op主要有以下几种

  EPOLL_CTL_ADD：往事件表中注册fd上的事件

  EPOLL_CTL_MOD：修改fd上的注册事件

  EPOLL_CTL_DEL：删除fd上的注册事件

• int epoll_wait ( int epfd, struct epoll_event* events, int maxevents, int timeout );

  函数说明：

  返回值：成功时返回就绪的文件描述符的个数，失败时返回-1并设置errno

  timeout：指定epoll的超时时间，单位是毫秒。当timeout为-1时，epoll_wait调用将永远阻塞，直到某个时间发生。当timeout为0时，epoll_wait调用将立即返回。

  events：就绪的fd对应的事件，由内核态返回赋值到用户态，然后读取这个evnents[i]进行事件处理

  maxevents：指定最多监听多少个事件

数据结构

调用epoll_create初始化epoll实例时，其实内核会创建一个eventpollfs类型的文件系统，因此我们看到该函数返回的是一个也是fd，后续的操作都围绕这个epoll实例展开。

调用epoll_ctl时会把传入的fd和监听事件包装为epitem，放入到红黑树中，这样做是为了加快查找删除fd的效率。

调用epoll_wait时，当有就绪的fd时，epoll会通过ep_poll_callback回调函数把就绪事件一个叫做rdlist的双向链表中。

总体处理流程

1. epoll_wait调用ep_poll，当rdlist为空（无就绪fd）时挂起当前进程，直到rdlist不空时进程才被唤醒，这里用到的是内核的队列唤醒等待机制。
2. 每个io设备的驱动中都注册了一个回调函数，当有数据发送获读取时会调用这个回调函数，比如键盘接收到用户的点击、网卡收到新事物数据包，这个回调函数就是 ep_ptable_queue_proc，然后调用ep_poll_callback。
3. ep_poll_callback将相应fd对应epitem加入rdlist，导致rdlist不空，进程被唤醒，epoll_wait得以继续执行。
4. ep_events_transfer函数将rdlist中的epitem拷贝到txlist中，并将rdlist清空。
5. ep_send_events函数（很关键），它扫描txlist中的每个epitem，调用其关联fd对用的poll方法。此时对poll的调用仅仅是取得fd上较新的events（防止之前events被更新），之后将取得的events和相应的fd发送到用户空间（封装在struct epoll_event，从epoll_wait返回）。

水平与边缘触发

水平触发EPOOLLT：默认模式，即fd就绪时，若fd对应的buffer数据未处理，则在下一次调用epoll_wait时会继续发送这个事件，直到就绪的fd对应的缓冲区数据被处理完，这种模式易于编程，不用担心事件数据丢失。缺点是需要不断调用epoll_wait系统调用来获取就绪事件，存在性能问题。水平触发的实现原理为：在调用完ep_send_events之后又执行了ep_reinject_items函数，该函数会把没有处理过的事件再次放到rdlist，因此下次调用epoll_wait时发现rdlist不为空就直接返回就绪事件了，因此看到的现象是会一直触发直到事件被处理。

边缘触发EPOLLET：当调用epoll_wait返回就绪事件后，若该事件未被处理，则下一次epoll_wait调用不会重复返回该事件，对程序编码要求较高，以防止事件未被处理而丢失，但该模式可以减少系统调用，提升效率。

参考

• https://www.cnblogs.com/lojunren/p/3856290.html
• http://blog.chinaunix.net/uid-28541347-id-4238524.html