epoll使用的简单例子

Posted on | In

前言

实习完了回家歇了一个多星期，也没闲着，在计划着做做自己的项目，由于自己主要学的是c++，所以更倾向于做网络以及服务器开发方面的项目，因此先研究了一番epoll，学习了一下libevent这个高性能事件驱动库，不过这些库学起来还是相对比较复杂的，所以打算先从epoll这个最基础的组件入手，自己慢慢写，然后看看这些库到底解决了哪些痛点

epoll介绍

这里其实介绍太多也没啥意思，简单介绍一下：epoll是Linux内核为处理大批量句柄而作了改进的poll，是Linux下多路复用IO接口select/poll的增强版本，它能显著减少程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。epoll的优点有以下几个：

  • 支持一个进程打开大数目的socket描述符。
  • IO效率不随FD数目增加而线性下降。
  • 内核支持，因此效率相对更高

epoll底层是一个红黑树和一个就绪文件描述符的链表，这里就不再细说，感兴趣的同学可以自行去了解一下，本文的目的不是介绍这个，而且要介绍底层的话又得带上源码，可能占用大量篇幅，这里就不再细说了。

epoll的接口函数

不同于之前的select和poll，epoll的接口是一组函数，下面我来简单介绍下epoll的接口函数

epoll_create

函数原型为int epoll_create(int size);
这里在较新的内核之后，size已经不再起作用，但是为了兼容性，还是要求size这个参数必须是正数，而不能是负数或者0，这个函数的作用也就是告诉内核创建一个epoll句柄，值得一提的是这个函数返回的也是一个文件描述符，因此需要注意的是在使用完之后，需要调用close来关闭这个描述符，否则可能导致文件描述符被耗尽。
返回值：返回一个正数表示创建成功，正数代表的也就是epoll对应的文件描述符，作为创建好的句柄。如果返回-1，代表创建失败，errno会相应被置为错误码。

epoll_ctl

函数原型为int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
这个函数就是主要进行操作的函数了，其中epfd就是我们在上一步中使用epoll_create获得的文件描述符，op代表操作选项，有以下三个可选项：

  • EPOLL_CTL_ADD：注册新的fd到epfd中；
  • EPOLL_CTL_MOD：修改已经注册的fd的监听事件；
  • EPOLL_CTL_DEL：从epfd中删除一个fd；

fd是指需要进行操作的文件描述符，epoll_event的结构如下：

1
2
3
4
5
6
7
8
9
10
11
typedef union epoll_data {  
   void *ptr;  
   int fd;  
   __uint32_t u32;  
   __uint64_t u64;  
} epoll_data_t;  
  
struct epoll_event {  
   __uint32_t events; /* Epoll events */  
   epoll_data_t data; /* User data variable */  
};

events可以是以下几个宏的集合：

  • EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
  • EPOLLOUT：表示对应的文件描述符可以写；
  • EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
  • EPOLLERR：表示对应的文件描述符发生错误；
  • EPOLLHUP：表示对应的文件描述符被挂断；
  • EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。
  • EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

值得说一下的是epoll_data的类型是一个union，所以也就是说实际上一个时刻，只有一种类型的数据会储存在epoll_data里面，不过这里面有个void *的指针，这个就够了，这就是很多库让用户提供回调函数，然后在用户指定的事件发生的时候，调用这个回调函数的原理，使用一个结构体储存一些必要信息，然后获得指向这个结构体的指针，强制类型转换为void *之后，传给epoll_data，再事件发生的时候，epoll会讲我们传递的这个指针再返回给我们，我们再强制类型转换回去，就能获得我们之前的数据

epoll_wait

函数原型为：int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
参数epfd是epoll的句柄，events则是一个指向数组首元素的指针，这个数组里面就是内核返回给我们的已经发生的事件集合，maxevents很简单，就是事件最大有多长，timeout的单位是毫秒，代表我们愿意等待的超时时间。
返回值：成功时，返回需要处理的事件数目。调用失败时，返回0，表示等待超时。

这里我也参考了一些教程，写了一个简单的例子，这个例子相当简陋，但是能够粗略看出如果使用epoll的接口函数，以及单纯只使用这些接口函数有什么弊端，如果要开发一个完成的库需要做哪些改进等等。代码如下，自己也只做了简单的测试，也就是冒烟通过，并没有做完整测试，所以如果有问题的话，希望看客们不吝指出，感谢！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <glog/logging.h>
#include <stdio.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>
#define MAX_EVENTS 1024
#define BUFLEN 1024
#define SERV_PORT 8080

struct event_t {
  int fd;          //事件相关联的文件描述符
  int event_flags; //事件相关联的事件标志,比如EPOLLIN,EPOLLOUT等等
  void *user_arg;  //在这里都是指向自身的指针
  void (*cb)(int fd, void *args); // call_back函数
  int status; //事件的状态,标示着事件是否被添加到epoll的监听池中
  char buf[BUFLEN + 1];
  int buf_len;
  int write_flag; ///判断是否需要写数据
};

int epoll_fd = -1;
struct event_t events[MAX_EVENTS + 1];

void init_event(event_t *ev, void (*cb)(int, void *), void *arg) {
  LOG(INFO) << __func__ << " start";
  ev->cb = cb;
  ev->status = 0;
  ev->user_arg = arg;
  LOG(INFO) << __func__ << " end";
}

void add_event(int fd, event_t *ev, int event_flags) {
  LOG(INFO) << __func__ << " start";
  int op;
  struct epoll_event ep_ev;
  ev->fd = fd;
  memset(&ep_ev, 0, sizeof(ep_ev));
  ep_ev.data.ptr = ev;
  ep_ev.events = ev->event_flags = event_flags;
  if (ev->status == 0) {
    op = EPOLL_CTL_ADD;
  } else {
    op = EPOLL_CTL_MOD;
  }
  int ret = epoll_ctl(epoll_fd, op, fd, &ep_ev);
  if (ret == -1) {
    LOG(ERROR) << "epoll_ctl failed fd:" << fd;
    LOG(INFO) << __func__ << " end";
    return;
  } else {
    LOG(INFO) << "epoll_ctl succeed fd:" << fd;
    ev->status = 1;
  }
  LOG(INFO) << __func__ << " end";
}

void delete_event(int fd, event_t *ev) {
  LOG(INFO) << __func__ << " start";
  if (ev->status != 1) {
    LOG(ERROR) << "event is not in the epoll list";
    LOG(INFO) << __func__ << " end";
    return;
  }
  int op = EPOLL_CTL_DEL;
  struct epoll_event ep_ev;
  memset(&ep_ev, 0, sizeof(ep_ev));
  ep_ev.data.ptr = ev;
  int ret = epoll_ctl(epoll_fd, op, fd, &ep_ev);
  if (ret == -1) {
    LOG(ERROR) << "epoll_ctl fail to remove fd:" << fd;
    LOG(INFO) << __func__ << " end";
    return;
  } else {
    LOG(INFO) << "epoll_ctl succeed to remove fd:" << fd;
    ev->status = 0;
  }
  LOG(INFO) << __func__ << " end";
}

void send_data_cb(int, void *);

void recv_data_cb(int conn_fd, void *args) {
  LOG(INFO) << __func__ << " start";
  struct event_t *ev = (event_t *)args;
  int result = -1;
  result = recv(conn_fd, ev->buf, BUFLEN, 0);
  delete_event(conn_fd, ev);
  if (result > 0) {
    LOG(INFO) << "receive data" << ev->buf;
    ev->buf_len = result;
    ev->buf[result] = '\0';
    init_event(ev, send_data_cb, (void *)ev);
    add_event(conn_fd, ev, EPOLLOUT);
  } else {
    if (result == 0) {
      LOG(INFO) << "connection is closed";
    } else {
      if (errno == EAGAIN) {
        LOG(INFO) << __func__ << " end";
        return;
      }
      LOG(WARNING) << "receive data failed error:" << strerror(errno);
    }
    close(conn_fd);
  }
  LOG(INFO) << __func__ << " end";
}

void send_data_cb(int conn_fd, void *args) {
  LOG(INFO) << __func__ << " start";
  struct event_t *ev = (event_t *)args;
  int result = -1;
  result = send(conn_fd, ev->buf, ev->buf_len, 0);
  delete_event(conn_fd, ev);
  if (result > 0) {
    LOG(INFO) << "send data:" << ev->buf;
    memset(ev->buf, 0, ev->buf_len);
    ev->buf_len = 0;
    init_event(ev, recv_data_cb, (void *)ev);
    add_event(conn_fd, ev, EPOLLIN);
  } else {
    close(conn_fd);
    LOG(WARNING) << "send data error:" << strerror(errno);
  }
  LOG(INFO) << __func__ << " end";
}

void accept_cb(int listener_fd, void *args) {
  LOG(INFO) << __func__ << " start";
  struct epoll_event *ev = (epoll_event *)args;
  struct sockaddr_in sin;
  memset(&sin, 0, sizeof(sin));
  socklen_t len = sizeof(sin);
  int conn_fd = -1;
  if ((conn_fd = accept(listener_fd, (struct sockaddr *)&sin, &len)) == -1) {
    LOG(ERROR) << "fail to accpet listen_fd:" << listener_fd;
    LOG(INFO) << __func__ << " end";
    return;
  }
  LOG(INFO) << "success to accept connection from ip:"
            << inet_ntoa(sin.sin_addr) << " port:" << ntohs(sin.sin_port);
  int event_index = 0;
  for (; event_index < MAX_EVENTS; ++event_index) {
    if (events[event_index].status == 0) {
      break;
    }
  }
  if (event_index == MAX_EVENTS) {
    close(conn_fd);
    LOG(ERROR) << "max connection limits";
    LOG(INFO) << __func__ << " end";
    return;
  }
  int ret = -1;
  if ((ret = fcntl(conn_fd, F_SETFL, O_NONBLOCK)) == -1) {
    close(conn_fd);
    LOG(ERROR) << "fail to set conn_fd:" << conn_fd
               << " non blocking error:" << strerror(errno);
    LOG(INFO) << __func__ << " end";
    return;
  }
  event_t *conn_event = &(events[event_index]);
  init_event(conn_event, recv_data_cb, (void *)ev);
  add_event(conn_fd, conn_event, EPOLLIN);
  LOG(INFO) << __func__ << " end";
}

int init_listener(int epoll_fd, int listen_port) {
  LOG(INFO) << __func__ << " start";
  int listener_fd = socket(AF_INET, SOCK_STREAM, 0);
  LOG(INFO) << "success to create new listener_fd:" << listener_fd;
  int ret = fcntl(listener_fd, F_SETFL, O_NONBLOCK);
  if (ret == -1) {
    LOG(ERROR) << "set listener_fd:" << listener_fd
               << " O_NONBLOCKING failed error" << strerror(errno);
    LOG(INFO) << __func__ << " end";
    return -1;
  }

  struct sockaddr_in addr;
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = INADDR_ANY;
  addr.sin_port = htons(listen_port);

  ret = bind(listener_fd, (struct sockaddr *)&addr, sizeof(addr));
  if (ret < 0) {
    LOG(ERROR) << "fail to bind port" << SERV_PORT
               << "error:" << strerror(errno);
    close(listener_fd);
    LOG(INFO) << __func__ << " end";
    return -1;
  }
  ret = listen(listener_fd, 20);
  if (ret < 0) {
    LOG(ERROR) << "fail to listen port error:" << strerror(errno);
    close(listener_fd);
    LOG(INFO) << __func__ << " end";
    return -1;
  }

  init_event(&events[MAX_EVENTS], accept_cb, &events[MAX_EVENTS]);
  add_event(listener_fd, &events[MAX_EVENTS], EPOLLIN);
  LOG(INFO) << __func__ << " end";
  return 0;
}

int main(int argc, char const *argv[]) {
  google::InitGoogleLogging("INFO");
  FLAGS_logtostderr = true;
  epoll_fd = epoll_create(MAX_EVENTS + 1);
  if (epoll_fd <= 0) {
    LOG(ERROR) << "failed to create epoll fd";
    return -1;
  }
  int ret = init_listener(epoll_fd, SERV_PORT);
  if (ret == -1) {
    LOG(ERROR) << "failed to init listener";
    return -1;
  }
  struct epoll_event ep_events[MAX_EVENTS + 1];
  while (1) {
    int active_fds = epoll_wait(epoll_fd, ep_events, MAX_EVENTS + 1, -1);
    if (active_fds < 0) {
      LOG(ERROR) << "erro epoll_wait";
      return -1;
    }
    for (size_t i = 0; i < active_fds; ++i) {
      struct event_t *ev = (event_t *)ep_events[i].data.ptr;
      if (ep_events[i].events | EPOLLIN) {
        LOG(INFO) << "fd:" << ev->fd << " is ready for recv data";
      } else if (ep_events[i].events | EPOLLOUT) {
        LOG(INFO) << "fd:" << ev->fd << " is ready for send data";
      }
      if ((ep_events[i].events | EPOLLIN) && (ev->event_flags | EPOLLIN)) {
        ev->cb(ev->fd, ev);
      } else if ((ep_events[i].events | EPOLLOUT) &&
                 (ep_events[i].events && EPOLLOUT)) {
        ev->cb(ev->fd, ev);
      }
    }
  }
  return 0;
}

代码解释

  1. 这里我使用了google的开源日志库glog，这里其实有点小题大做了，但是个人觉得c++自带的输出信息实在太少，自己又不想麻烦去写一个日志库，索性就直接用了google家的日志库，这个库本身也是非常好用的，引入头文件之后在main函数最前面简单初始化一下就ok了。

  2. 另外一个值得一提的是，我在收到来自己客户端的消息时，并没有去向epoll中添加一个监听套接字是否可写的事件，这里的原因很简单，因为实际上只要操作系统中，这个套接字的写缓冲里面还有空间，就是一直可写的，但是空间是否够大，这个就是不确定的了，所以我们需要做的其实很简单，直接向套接字调用send发送数据，如果send返回成功，我们就已经成功将数据写入了，只有当send返回失败同时errno为EAGAIN时，我们才需要向epoll添加一个监听套接字是否可写的事件，等待可写再进行写操作，说来惭愧，我这里没有做这一步的操作，因为这里只是单纯的熟悉一些epoll的一些接口，但是如果在开发一个库的时候是绝对不能这样做的。

  3. epoll的事件触发模式有两种，默认是LT也就是水平触发，ET是可选的，我这里压根没提ET，虽然面试的时候可能经常会问这两种模式的区别，但是我个人认为实际开发中是不应该使用了ET模式的，ET模式下数据丢失带来的后果比一点点性能提高带来的好处是更大的，所以从编程方便以及实用的角度，用默认的LT模式就好，没有必要使用ET模式，因此我这里也没做深入介绍

总结

这份代码我在我的电脑上（ubuntu16.04默认环境下）成功编译通过，需要注意的是电脑上需要安装glog，编译的时候也需要加上选项-lglog，才能够编译，否则在链接动态库的那一步是无法通过的哦，通过这份代码，我们可以熟悉一下epoll接口的基本用法，了解一下单纯只使用epoll的接口的弊端，需要作出哪些改进，这个我后面会再提到。暂时了解的不是非常深入无法很好的总结提炼出来，这里就不嫌丑了。