IO复用之epoll

知识地图

IO复用之epoll 图1

1.epoll基础简介

1.1 相关函数介绍

epoll_create函数

epoll_create函数用于创建epoll文件描述符，该文件描述符用于后续的epoll操作，参数size目前还没有实际用处，我们只要填一个大于0的数就行。

# include <sys/epoll.h>
int epoll_create(int size);

参数：
size:目前内核还没有实际使用，只要大于0就行
返回值：
返回epoll文件描述符

IO复用之epoll 图2

图 1

epoll_ctl函数

epoll_ctl函数用于增加，删除，修改epoll事件，epoll事件会存储于内核epoll结构体红黑树中。

#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

参数：
epfd：epoll文件描述符
op：操作码
EPOLL_CTL_ADD:插入事件
EPOLL_CTL_DEL:删除事件
EPOLL_CTL_MOD:修改事件
fd：事件绑定的套接字文件描述符
events：事件结构体
返回值：
成功：返回0
失败：返回-1

IO复用之epoll 图3

图 2

struct epoll_event结构体 epoll_event事件结构体

#include <sys/epoll.h>
struct epoll_event{  uint32_t events; //epoll事件，参考事件列表
epoll_data_t data;
} ;
typedef union epoll_data {
    void *ptr;
    int fd;  //套接字文件描述符
uint32_t u32;
    uint64_t u64;
} epoll_data_t;

epoll事件列表：

 头文件：<sys/epoll.h>
enum EPOLL_EVENTS{
    EPOLLIN = 0x001, //读事件
EPOLLPRI = 0x002,
EPOLLOUT = 0x004, //写事件
EPOLLRDNORM = 0x040,
EPOLLRDBAND = 0x080,
EPOLLWRNORM = 0x100,
EPOLLWRBAND = 0x200,
EPOLLMSG = 0x400,
EPOLLERR = 0x008, //出错事件
EPOLLHUP = 0x010, //出错事件
EPOLLRDHUP = 0x2000,
EPOLLEXCLUSIVE = 1u << 28,
EPOLLWAKEUP = 1u << 29,
EPOLLONESHOT = 1u << 30,
EPOLLET = 1u << 31 //边缘触发
};

epoll_wait函数 epoll_wait用于监听套接字事件，可以通过设置超时时间timeout来控制监听的行为为阻塞模式还是超时模式。

#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event *events,
int maxevents, int timeout);

参数：
epfd：epoll文件描述符
events：epoll事件数组
maxevents：epoll事件数组长度
timeout：超时时间
小于0：一直等待
等于0：立即返回
大于0：等待超时时间返回，单位毫秒
返回值：
小于0：出错
等于0：超时
大于0：返回就绪事件个数

IO复用之epoll 图4

图 3

2.epoll软件架构

IO复用之epoll 图5

图 4

## 3.LT模式和ET模式 ### 3.1 LT模式：水平触发

socket读触发：socket接收缓冲区有数据，会一直触发epoll_wait EPOLLIN事件，直到数据被用户读取完。
socket写触发：socket可写，会一直触发epoll_wait EPOLLOUT事件。

3.2 ET模式：边缘触发

socket读触发：socket数据从无到有，会触发epoll_wait EPOLLIN事件，只会触发一次EPOLLIN事件，用户检测到事件后，需一次性把socket接收缓冲区数据全部读取完，读取完的标志为recv返回-1，errno为EAGAIN。
socket写触发：socket可写，会触发一次epoll_wait EPOLLOUT事件。边缘触发读取数据示例代码：

/* 边缘触发：一次性把接收缓冲区数据全部读完 */
memset(recv_buf, 0, BUF_SIZE);
unsigned int len = 0;
while(1) {
    ret = recv(fd, recv_buf + len, BUF_SIZE, 0);
    if (ret == 0) {
        printf("remove fd:%d\n", fd);
        epoll_ctl(efd, EPOLL_CTL_DEL, fd, NULL);
        close(fd);
        break;
    } else if ((ret == -1) && ((errno == EINTR) || (errno == EAGAIN) || (errno == EWOULDBLOCK))) {
        printf("fd:%d recv errno:%d done\n", fd, errno);
        break;
    } else if ((ret == -1) && !((errno == EINTR) || (errno == EAGAIN) || (errno == EWOULDBLOCK))) {
        printf("remove fd:%d errno:%d\n", fd, errno);
        epoll_ctl(efd, EPOLL_CTL_DEL, fd, NULL);
        close(fd);
        break;
    } else {
        len += ret;
    }
    printf("recv fd:%d, len:%d, %s\n", fd, ret, recv_buf);
}

[/code]

4.阻塞和非阻塞

讨论epoll阻塞和非阻塞得从两方面讨论：epoll阻塞和epoll监听套接字阻塞。

epoll阻塞：epoll自身是阻塞的，我们可以通过epoll_wait超时参数设置epoll阻塞行为。
epoll监听套接字阻塞：epoll监听套接字阻塞是指插入epoll监听事件的套接字设置为阻塞模式。

epoll监听套接字设置成阻塞还是非阻塞？这个问题可以肯定的回答是非阻塞，因为epoll是为高并发设计的，任何的其他阻塞行为，都会影响epoll高效运行。

5.epoll为什么高效？

红黑树红黑树提高epoll事件增删查改效率。
回调通知机制

当epoll监听套接字有数据读或者写时，会通过注册到socket的回调函数通知epoll，epoll检测到事件后，将事件存储在就绪队列（rdllist）。

就绪队列

epoll_wait返回成功后，会将所有就绪事件存储在事件数组，用户不需要进行无效的轮询，从而提高了效率。

6.epoll示例程序

6.1 服务端程序

 
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <fcntl.h>
#include <errno.h>
#include <stdbool.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#define LISTEN_BACKLOG (5)
#define BUF_SIZE (2048)
#define ONCE_READ_SIZE (1500)
#define EPOLL_SIZE (100);
#define MAX_EVENTS (10)
void usage(void) {
    printf("*********************************\n");
    printf("./server 本端ip 本端端口\n");
    printf("*********************************\n");
}
void setnonblocking(int fd) {
    int flag = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}
int main(int argc, char *argv[]){
    struct sockaddr_in local;
    struct sockaddr_in peer;
    socklen_t addrlen = sizeof(peer);
    int sock_fd = 0, new_fd = 0;
    int ret = 0;
    char send_buf[BUF_SIZE] = {0};
    char recv_buf[BUF_SIZE] = {0};
if (argc != 3) {
    usage();
    return -1;
}
char *ip = argv[1];
unsigned short port = atoi(argv[2]);
printf("ip:port->%s:%u\n", argv[1], port);
sock_fd = socket(AF_INET, SOCK_STREAM, 0);
if (sock_fd == -1) {
    perror("socket error");
    return -1;
}
memset(&local, 0, sizeof(struct sockaddr_in));
local.sin_family = AF_INET;
local.sin_addr.s_addr = inet_addr(ip);
local.sin_port = htons(port);
ret = bind(sock_fd, (struct sockaddr *)&local, sizeof(struct sockaddr));
if (ret == -1) {
    close(sock_fd);
    perror("bind error");
    return -1;
}
ret = listen(sock_fd, LISTEN_BACKLOG);
if (ret == -1) {
    close(sock_fd);
    perror("listen error");
    return -1;
}
int epoll_size = EPOLL_SIZE;
int efd = epoll_create(epoll_size);
if (efd == -1) {
    perror("epoll create error");
    return -1;
}
struct epoll_event ev, events[MAX_EVENTS];
ev.data.fd = sock_fd;
ev.events = EPOLLIN;
if (epoll_ctl(efd, EPOLL_CTL_ADD, sock_fd, &ev) == -1) {
    perror("epoll ctl ADD error");
    return -1;
}
int timeout = 1000;
while (1) {
    int nfds = epoll_wait(efd, events, MAX_EVENTS, timeout);
    if (nfds == -1) {
        perror("epoll wait error");
        return -1;
    } else if (nfds == 0) {
        printf("epoll wait timeout\n");
        continue;
    } else {
        }
for (int i = 0; i < nfds; i++) {
    int fd = events[i].data.fd;
    printf("events[%d] events:%08x\n", i, events[i].events);
    if (fd == sock_fd) {
        new_fd = accept(sock_fd, (struct sockaddr *)&peer, &addrlen);
        if (new_fd == -1) {
            perror("accept error");
            continue;
        }
        setnonblocking(new_fd);
        ev.data.fd = new_fd;
        ev.events = EPOLLIN|EPOLLET;
        if (epoll_ctl(efd, EPOLL_CTL_ADD, new_fd, &ev) == -1) {
            perror("epoll ctl ADD new fd error");
            close(new_fd);
            continue;
        }
    } else {
        if (events[i].events & EPOLLIN) {
            printf("fd:%d is readable\n", fd);
            memset(recv_buf, 0, BUF_SIZE);
            unsigned int len = 0;
            while(1) {
                ret = recv(fd, recv_buf + len, ONCE_READ_SIZE, 0);
                if (ret ==  0) {
                    printf("remove fd:%d\n", fd);
                    epoll_ctl(efd, EPOLL_CTL_DEL, fd, NULL);
                    close(fd);
                    break;
                } else if ((ret == -1) && ((errno == EINTR) || (errno == EAGAIN) || (errno == EWOULDBLOCK))) {
                    printf("fd:%d recv errno:%d done\n", fd, errno);
                    break;
                } else if ((ret == -1) && !((errno == EINTR) || (errno == EAGAIN) || (errno == EWOULDBLOCK))) {
                    printf("remove fd:%d errno:%d\n", fd, errno);
                    epoll_ctl(efd, EPOLL_CTL_DEL, fd, NULL);
                    close(fd);
                    break;
                }else {
                    printf("once read ret:%d\n", ret);
                    len += ret;
                }
            }
            printf("recv fd:%d, len:%d, %s\n", fd, len, recv_buf);
        } if (events[i].events & EPOLLOUT) {
            printf("fd:%d is sendable\n", fd);
        } else if ((events[i].events & EPOLLERR) ||                        ((events[i].events & EPOLLHUP))) {
            printf("fd:%d error\n", fd);
        }
    }
}
}
    return 0;}

6.2 客户端程序

 
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#define LISTEN_BACKLOG (5)
#define BUF_SIZE (1500)
#define REQUEST_STR "tcp pack"
void usage(void) {
    printf("*********************************\n");
    printf("./client 对端ip 对端端口\n");
    printf("*********************************\n");
}
int main(int argc, char *argv[]){
    struct sockaddr_in client;
    struct sockaddr_in server;
    int sock_fd = 0;
    int ret = 0;
    socklen_t addrlen = 0;
    char send_buf[BUF_SIZE] = {0};
    char recv_buf[BUF_SIZE] = {0};
if (argc != 3) {
    usage();
    return -1;
}
char *ip = argv[1];
unsigned short port = atoi(argv[2]);
printf("ip:port->%s:%u\n", argv[1], port);
sock_fd = socket(AF_INET, SOCK_STREAM, 0);
if (sock_fd == -1) {
    perror("socket error");
    return -1;
}
memset(&server, 0, sizeof(struct sockaddr_in));
server.sin_family = AF_INET;
server.sin_addr.s_addr = inet_addr(ip);
server.sin_port = htons(port);
ret = connect(sock_fd, (struct sockaddr *)&server, sizeof(struct sockaddr));
if (ret == -1) {
    close(sock_fd);
    perror("connect error");
    return -1;
}
char seq = 0x31;
while(1) {
    memset(send_buf, seq, BUF_SIZE);
    send(sock_fd, send_buf, BUF_SIZE, 0);
    printf("send %s\n", send_buf);
    sleep(2);
    seq++;
}
close(sock_fd);
    return 0;}