libev & libevent简介
最近开始重构定制公司的网站后台服务器，开始关注libevent 以及livev 相关 ，也欢迎相关的同学一起讨论。这两者采用相同的架构和设计思想，很多原理和代码都可以相互参考和对比理解。
简介
libev 和libevent 比较适合分布式并发系统，关于其和传统方式的比较，可以参看以下内容：
原文地址：http://www.ibm.com/developerworks/cn/aix/library/au-libev/index.html
许多服务器部署（尤其是 web 服务器部署）面对的最大问题之一是必须能够处理大量连接。无论是通过构建基于云的服务来处理网络通信流，还是把应用程序分布在 IBM Amazon EC 实例上，还是为网站提供高性能组件，都需要能够处理大量并发连接。
一个好例子是，web 应用程序最近越来越动态了，尤其是使用 AJAX 技术的应用程序。如果要部署的系统允许数千客户端直接在网页中更新信息，比如提供事件或问题实时监视的系统，那么提供信息的速度就非常重要了。在网格或云环境中，可能有来自数千客户端的持久连接同时打开着，必须能够处理每个客户端的请求并做出响应。
在讨论 libevent 和 libev 如何处理多个网络连接之前，我们先简要回顾一下处理这类连接的传统解决方案。
回页首
处理多个客户端
处理多个连接有许多不同的传统方法，但是在处理大量连接时它们往往会产生问题，因为它们使用的内存或 CPU 太多，或者达到了某个操作系统限制。
使用的主要方法如下：
- 循环：早期系统使用简单的循环选择解决方案，即循环遍历打开的网络连接的列表，判断是否有要读取的数据。这种方法既缓慢（尤其是随着连接数量增加越来越慢），又低效（因为在处理当前连接时其他连接可能正在发送请求并等待响应）。在系统循环遍历每个连接时，其他连接不得不等待。如果有 100 个连接，其中只有一个有数据，那么仍然必须处理其他 99 个连接，才能轮到真正需要处理的连接。
- poll、epoll 和变体：这是对循环方法的改进，它用一个结构保存要监视的每个连接的数组，当在网络套接字上发现数据时，通过回调机制调用处理函数。poll 的问题是这个结构会非常大，在列表中添加新的网络连接时，修改结构会增加负载并影响性能。
-
选择：
select()函数调用使用一个静态结构，它事先被硬编码为相当小的数量（1024 个连接），因此不适用于非常大的部署。
在各种平台上还有其他实现（比如 Solaris 上的 /dev/poll 或 FreeBSD/NetBSD 上的 kqueue），它们在各自的 OS 上性能可能更好，但是无法移植，也不一定能够解决处理请求的高层问题。
上面的所有解决方案都用简单的循环等待并处理请求，然后把请求分派给另一个函数以处理实际的网络交互。关键在于循环和网络套接字需要大量管理代码，这样才能监听、更新和控制不同的连接和接口。
处理许多连接的另一种方法是，利用现代内核中的多线程支持监听和处理连接，为每个连接启动一个新线程。这把责任直接交给操作系统，但是会在 RAM 和 CPU 方面增加相当大的开销，因为每个线程都需要自己的执行空间。另外，如果每个线程都忙于处理网络连接，线程之间的上下文切换会很频繁。最后，许多内核并不适于处理如此大量的活跃线程。
回页首
libevent 方法
libevent 库实际上没有更换select()、poll()或其他机制的基础。而是使用对于每个平台最高效的高性能解决方案在实现外加上一个包装器。
为了实际处理每个请求，libevent 库提供一种事件机制，它作为底层网络后端的包装器。事件系统让为连接添加处理函数变得非常简便，同时降低了底层 I/O 复杂性。这是 libevent 系统的核心。
libevent 库的其他组件提供其他功能，包括缓冲的事件系统（用于缓冲发送到客户端/从客户端接收的数据）以及 HTTP、DNS 和 RPC 系统的核心实现。
创建 libevent 服务器的基本方法是，注册当发生某一操作（比如接受来自客户端的连接）时应该执行的函数，然后调用主事件循环event_dispatch()。执行过程的控制现在由 libevent 系统处理。注册事件和将调用的函数之后，事件系统开始自治；在应用程序运行时，可以在事件队列中添加（注册）或删除（取消注册）事件。事件注册非常方便，可以通过它添加新事件以处理新打开的连接，从而构建灵活的网络处理系统。
例如，可以打开一个监听套接字，然后注册一个回调函数，每当需要调用accept()函数以打开新连接时调用这个回调函数，这样就创建了一个网络服务器。清单 1所示的代码片段说明基本过程：
系统接口介绍
地址：http://www.dirlt.com/libev.html
Table of Contents
-
1 libev
- 1.1 About The Code
- 1.2 EventLoop
- 1.3 Watcher
-
1.4 How it works
- 1.4.1 ev_run
- 1.4.2 fd_reify
- 1.4.3 backend_poll
- 1.4.4 timers_reify
- 1.4.5 EV_INVOKE_PENDING
-
1.5 Example
- 1.5.1 common.h
- 1.5.2 echo-client.cc
- 1.5.3 echo-server.cc
1libev
主页http://software.schmorp.de/pkg/libev.html
文档http://software.schmorp.de/pkg/libev.html
libev所实现的功能就是一个强大的reactor,可能notify事件主要包括下面这些：
- ev_io // IO可读可写
- ev_stat // 文件属性变化
- ev_async // 激活线程
- ev_signal // 信号处理
- ev_timer // 定时器
- ev_periodic // 周期任务
- ev_child // 子进程状态变化
- ev_fork // 开辟进程
- ev_cleanup // event loop退出触发事件
- ev_idle // 每次event loop空闲触发事件
- ev_embed // TODO(zhangyan04):I have no idea.
- ev_prepare // 每次event loop之前事件
- ev_check // 每次event loop之后事件
1.1About The Code
代码风格相当严谨而且排版也非常工整，并且从域名看出作者是德国人。但是内部使用了大量的宏造成阅读代码并不是非常方便。 并且从代码角度分析，应该是一开始支持有一个默认的event_loop,但是随着多核产生实际应用中可能会使用到多个event_loop, 猜想作者应该是为了方便的话使用了很多宏进行替换。允许使用多个event_loop的宏是EV_MULTIPLICITY.比如下面这段代码
void noinline ev_io_start (EV_P_ ev_io *w) { int fd = w->fd; if (expect_false (ev_is_active (w))) return; assert (("libev: ev_io_start called with negative fd", fd >= 0)); assert (("libev: ev_io_start called with illegal event mask", !(w->events & ~(EV__IOFDSET | EV_READ | EV_WRITE)))); EV_FREQUENT_CHECK; ev_start (EV_A_ (W)w, 1); array_needsize (ANFD, anfds, anfdmax, fd + 1, array_init_zero); wlist_add (&anfds[fd].head, (WL)w); fd_change (EV_A_ fd, w->events & EV__IOFDSET | EV_ANFD_REIFY); w->events &= ~EV__IOFDSET; EV_FREQUENT_CHECK; }
初次阅读这个代码会觉得非常难懂。
| 宏 | 说明 | 定义 |
|---|---|---|
| EV_P | event parameter | struct ev_loop *loop |
| EV_P_ | EV_P, | |
| EV_A | event argument | loop |
| EV_A_ | EV_A, |
然后很多变量只要是ev_loop成员的话都被封装成为了宏。比如代码里面的anfds,实际上的宏定义是
#define anfds ((loop)->anfds)
事实上一个ev_loop里面的字段是相当多的，不过也很正常本身就是一个强大的reactor.但是这造成一个直接后果， 就是对于想要了解ev_loop的全貌比较困难，所以想要彻底地了解libev也比较麻烦，所以我们只能够从应用层面来尝试了解它。
1.2EventLoop
首先我们关注一下reactor本身。在libev下面reactor对象称为event_loop.event_loop允许动态创建和销毁，并且允许绑定自定义数据
struct ev_loop * ev_loop_new (unsigned int flags); void ev_loop_destroy (EV_P); void ev_set_userdata (EV_P_ void *data); void *ev_userdata (EV_P);
我们这里主要关注一下flags.这里面主要是选择使用什么backend来进行poll操作，可以选择的有：
- EVBACKEND_SELECT
- EVBACKEND_POLL
- EVBACKEND_EPOLL // 通常我们选择这个
- EVBACKEND_KQUEUE
- EVBACKEND_DEVPOLL
- EVBACKEND_PORT
但是还有三个比较重要选项：
- EVFLAG_NOINOTIFY // 不适用inofity调用来使用ev_stat.这样可以减少fd使用。
- EVFLAG_SIGNALFD // 使用signalfd来检测信号是否发生，同样这样可以减少fd使用。
大部分时候我们使用EVFLAG_AUTO(0)一般就足够满足需求了，从代码角度来看如果支持epoll的话那么首先会选择epoll. 因为在watcher的回调函数里面是可以知道当前event_loop的，这样就可以获得自定义数据。然后我们看看这个event_loop如何运行和停止的
void ev_run (EV_P_ int flags); void ev_break (EV_P_ int how);
同样我们这里比较关注flags和how这两个参数。flags有下面这几个：
- 0.通常这是我们想要的，每次轮询在poll都会等待一段时间然后处理pending事件。
- EVRUN_NOWAIT.运行一次，在poll时候不会等待。这样效果相当于只是处理pending事件。
- EVRUN_ONCE.运行一次，但是在poll时候会等待，然后处理pending事件。
而how有下面这几个：
- EVBREAK_ONE.只是退出一次ev_run这个调用。通常来说使用这个就可以了。
- EVBREAK_ALL.退出所有的ev_run调用。这种情况存在于ev_run在pengding处理时候会递归调用。
在backend/epoll底层每次epoll_wait时候，libev提供了接口回调可以在epoll_wait前后调用
void ev_set_loop_release_cb (loop, void (*release)(EV_P), void (*acquire)(EV_P)) static void epoll_poll (EV_P_ ev_tstamp timeout) { /* epoll wait times cannot be larger than (LONG_MAX - 999UL) / HZ msecs, which is below */ /* the default libev max wait time, however. */ EV_RELEASE_CB; eventcnt = epoll_wait (backend_fd, epoll_events, epoll_eventmax, epoll_epermcnt ? 0 : ev_timeout_to_ms (timeout)); EV_ACQUIRE_CB; }
在event_loop里面我们还关心一件事情，就是每次event_loop轮询的时间长短。通常来说这个不会是太大问题，但是在高性能情况下面我们需要设置
void ev_set_io_collect_interval (EV_P_ ev_tstamp interval); void ev_set_timeout_collect_interval (EV_P_ ev_tstamp interval);
在ev_run里面有使用这些参数的代码比较麻烦。但是大意是这样，如果我们这是了timeout_interval的话，那么我们每次检查timeout时间的话必须 在timeout_interval，使用这段时间ev_sleep.但是这个又会影响到io_interval,所以内部做了一些换算，换算的结果作为epoll_wait超时时间。 不过同样在大部分时候我们不需要关心它，默认时候是0.0,系统会使用最快的响应方式来处理。
1.3Watcher
然后我们关心一下EventHandler.在libev下面watcher相当于EventHandler这么一个概念，通常里面会绑定fd回调函数以及我们需要关注的事件。 然后一旦触发事件之后会触发我们使用的回调函数，回调函数参数通常有reactor,watcher以及触发的事件。这里不打算重复文档里面的watcher 相关的内容和对应的API,但是对于某些内容的话可能会提到并且附带一些注释。之前我们还是看看通用过程，这里使用TYPE区分不同类型watcher.
typedef void (*)(struct ev_loop *loop, ev_TYPE *watcher, int revents) callback; // callback都是这种类型 ev_init (ev_TYPE *watcher, callback); // 初始化watcher ev_TYPE_set (ev_TYPE *watcher, [args]); // 设置watcher ev_TYPE_init (ev_TYPE *watcher, callback, [args]); // 通常使用这个函数最方便，初始化和设置都在这里 ev_TYPE_start (loop, ev_TYPE *watcher); // 注册watcher ev_TYPE_stop (loop, ev_TYPE *watcher); // 注销watcher ev_set_priority (ev_TYPE *watcher, int priority); // 设置优先级 ev_feed_event (loop, ev_TYPE *watcher, int revents); // 这个做跨线程通知非常有用，相当于触发了某个事件。 bool ev_is_active (ev_TYPE *watcher); // watcher是否active. bool ev_is_pending (ev_TYPE *watcher); // watcher是否pending. int ev_clear_pending (loop, ev_TYPE *watcher); // 清除watcher pending状态并且返回事件
wacther的状态有下面这么几种：
- initialiased.调用init函数初始化
- active.调用start进行注册
- pending.已经触发事件但是没有处理
- inactive.调用stop注销。这个状态等同于initialised这个状态。
其实关于每个watcher具体是怎么实现的没有太多意思，因为大部分现有代码都差不多。会在下一节说说内部数据结构是怎么安排的， 了解内部数据结构以及过程之后很多问题就可以避免了，比如"The special problem of disappearing file descriptors"这类问题。
1.4How it works
1.4.1ev_run
最主要的还是看看ev_run这个部分代码。我们不打算仔细阅读只是看看梗概然后大体分析一下数据结构应该怎么样的
void ev_run (EV_P_ int flags) { assert (("libev: ev_loop recursion during release detected", loop_done != EVBREAK_RECURSE)); loop_done = EVBREAK_CANCEL; EV_INVOKE_PENDING; /* in case we recurse, ensure ordering stays nice and clean */ do { if (expect_false (loop_done)) break; /* update fd-related kernel structures */ fd_reify (EV_A); /* calculate blocking time */ { ev_tstamp waittime = 0.; ev_tstamp sleeptime = 0.; /* remember old timestamp for io_blocktime calculation */ ev_tstamp prev_mn_now = mn_now; /* update time to cancel out callback processing overhead */ time_update (EV_A_ 1e100); if (expect_true (!(flags & EVRUN_NOWAIT || idleall || !activecnt))) { waittime = MAX_BLOCKTIME; if (timercnt) { ev_tstamp to = ANHE_at (timers [HEAP0]) - mn_now + backend_fudge; if (waittime > to) waittime = to; } /* don't let timeouts decrease the waittime below timeout_blocktime */ if (expect_false (waittime < timeout_blocktime)) waittime = timeout_blocktime; /* extra check because io_blocktime is commonly 0 */ if (expect_false (io_blocktime)) { sleeptime = io_blocktime - (mn_now - prev_mn_now); if (sleeptime > waittime - backend_fudge) sleeptime = waittime - backend_fudge; if (expect_true (sleeptime > 0.)) { ev_sleep (sleeptime); waittime -= sleeptime; } } } assert ((loop_done = EVBREAK_RECURSE, 1)); /* assert for side effect */ backend_poll (EV_A_ waittime); assert ((loop_done = EVBREAK_CANCEL, 1)); /* assert for side effect */ /* update ev_rt_now, do magic */ time_update (EV_A_ waittime + sleeptime); } /* queue pending timers and reschedule them */ timers_reify (EV_A); /* relative timers called last */ EV_INVOKE_PENDING; } while (expect_true ( activecnt && !loop_done && !(flags & (EVRUN_ONCE | EVRUN_NOWAIT)) )); if (loop_done == EVBREAK_ONE) loop_done = EVBREAK_CANCEL; }
我们可以总结一下大致步骤，其实和大部分的event loop写出来差不多。
- 首先触发那些已经pending的watchers.
- 判断是否loop_done
- fd_reify.这个后面会单独说。
- 计算出waittime并且进行必要的sleep.
- backend_poll开始轮询,并且整理好pending事件
- timers_reify.这个和fd_reify不同
- 调用EV_INVOKE_PENDING来触发pending的io事件
非常简单。接下来我们看看fd_reify,backend_poll,timers_reify以及EV_INVOKE_PENDING.
1.4.2fd_reify
下面是fd_reify代码片段.可以看出，这个部分就是在修改fd关注的events。
inline_size void fd_reify (EV_P) { int i; for (i = 0; i < fdchangecnt; ++i) { int fd = fdchanges [i]; ANFD *anfd = anfds + fd; ev_io *w; unsigned char o_events = anfd->events; unsigned char o_reify = anfd->reify; anfd->reify = 0; /*if (expect_true (o_reify & EV_ANFD_REIFY)) probably a deoptimisation */ { anfd->events = 0; for (w = (ev_io *)anfd->head; w; w = (ev_io *)((WL)w)->next) anfd->events |= (unsigned char)w->events; if (o_events != anfd->events) o_reify = EV__IOFDSET; /* actually |= */ } if (o_reify & EV__IOFDSET) backend_modify (EV_A_ fd, o_events, anfd->events); } fdchangecnt = 0; }
而这个fdchanges这个是在哪里调用的呢。我们可以看到就是在ev_io_start这个部分。也就是说如果我们想要修改 fd关注事件的话，我们必须显示地ev_io_stop掉然后修正之后重新ev_io_start.底层调用fd_change的话底层维护 数组fdchanges来保存发生events变动的fd.
void noinline ev_io_start (EV_P_ ev_io *w) { int fd = w->fd; if (expect_false (ev_is_active (w))) return; assert (("libev: ev_io_start called with negative fd", fd >= 0)); assert (("libev: ev_io_start called with illegal event mask", !(w->events & ~(EV__IOFDSET | EV_READ | EV_WRITE)))); EV_FREQUENT_CHECK; ev_start (EV_A_ (W)w, 1); array_needsize (ANFD, anfds, anfdmax, fd + 1, array_init_zero); wlist_add (&anfds[fd].head, (WL)w); fd_change (EV_A_ fd, w->events & EV__IOFDSET | EV_ANFD_REIFY); w->events &= ~EV__IOFDSET; EV_FREQUENT_CHECK; } inline_size void fd_change (EV_P_ int fd, int flags) { unsigned char reify = anfds [fd].reify; anfds [fd].reify |= flags; if (expect_true (!reify)) { ++fdchangecnt; array_needsize (int, fdchanges, fdchangemax, fdchangecnt, EMPTY2); fdchanges [fdchangecnt - 1] = fd; } }
1.4.3backend_poll
backend_poll底层支持很多poll实现，我们这里仅仅看ev_epoll.c就可以.代码在这里面我们不列举了， 如果某个fd触发事件的话那么最终会调用fd_event(EV_A_,fd,event)来进行通知。所以我们看看fd_event.
inline_speed void fd_event_nocheck (EV_P_ int fd, int revents) { ANFD *anfd = anfds + fd; ev_io *w; for (w = (ev_io *)anfd->head; w; w = (ev_io *)((WL)w)->next) { int ev = w->events & revents; if (ev) ev_feed_event (EV_A_ (W)w, ev); } } void noinline ev_feed_event (EV_P_ void *w, int revents) { W w_ = (W)w; int pri = ABSPRI (w_); if (expect_false (w_->pending)) pendings [pri][w_->pending - 1].events |= revents; else { w_->pending = ++pendingcnt [pri]; array_needsize (ANPENDING, pendings [pri], pendingmax [pri], w_->pending, EMPTY2); // set the watcher and revents. pendings [pri][w_->pending - 1].w = w_; pendings [pri][w_->pending - 1].events = revents; } }
可以看到底层是一个ANFD的数组，根据fd进行偏移。如果fd过大的话似乎会影响性能没有hpserver里面的demuxtable实现方式好。 然后得到这个fd下面所有的watcher,然后在loop->pendings里面记录所有这些触发的watcher.
1.4.4timers_reify
其中HEAP0就是最小堆下标。如果repeat的话说明需要重复发生，那么就会重新调整时间戳，如果不是repeat的话， 那么内部会调用ev_timer_stop这个方法将这个计时器移除。所有的定时任务都通过feed_reverse添加。feed_reverse 内部是维护一个动态数组来保存所有的定时器任务，然后在feed_reverse_done里面遍历这些任务来触发这些定时器任务。
inline_size void timers_reify (EV_P) { EV_FREQUENT_CHECK; if (timercnt && ANHE_at (timers [HEAP0]) < mn_now) { do { ev_timer *w = (ev_timer *)ANHE_w (timers [HEAP0]); /*assert (("libev: inactive timer on timer heap detected", ev_is_active (w)));*/ /* first reschedule or stop timer */ if (w->repeat) { ev_at (w) += w->repeat; if (ev_at (w) < mn_now) ev_at (w) = mn_now; assert (("libev: negative ev_timer repeat value found while processing timers", w->repeat > 0.)); ANHE_at_cache (timers [HEAP0]); downheap (timers, timercnt, HEAP0); } else ev_timer_stop (EV_A_ w); /* nonrepeating: stop timer */ EV_FREQUENT_CHECK; feed_reverse (EV_A_ (W)w); } while (timercnt && ANHE_at (timers [HEAP0]) < mn_now); feed_reverse_done (EV_A_ EV_TIMER); } }
1.4.5EV_INVOKE_PENDING
这个宏最终调用的函数就是下面这个，遍历所有的pendings事件并且逐一触发。
void noinline ev_invoke_pending (EV_P) { int pri; for (pri = NUMPRI; pri--; ) while (pendingcnt [pri]) { ANPENDING *p = pendings [pri] + --pendingcnt [pri]; p->w->pending = 0; EV_CB_INVOKE (p->w, p->events); EV_FREQUENT_CHECK; } }
1.5Example
尝试编写一个简单的带有超时的echo-server和echo-client就发现其实还有非常多的其他的工作量，比如buffer的管理状态机实现等。 所以我没有写出一个完整的example,只是简单地写了假设echo-client连接上server的话就简单地打印链接信息并且关闭。
1.5.1common.h
#ifndef _COMMON_H_ #define _COMMON_H_ #include <unistd.h> #include <fcntl.h> #include <sys/types.h> #include <sys/socket.h> #include <arpa/inet.h> #include <strings.h> #include <cstdlib> #include <cstdio> #include <cstddef> #include <string> namespace common{ #define D(exp,fmt,...) do { \ if(!(exp)){ \ fprintf(stderr,fmt,##__VA_ARGS__); \ abort(); \ } \ }while(0) static void setnonblock(int fd){ fcntl(fd,F_SETFL,fcntl(fd,F_GETFL) | O_NONBLOCK); } static void setreuseaddr(int fd){ int ok=1; setsockopt(fd,SOL_SOCKET,SO_REUSEADDR,&ok,sizeof(ok)); } static void setaddress(const char* ip,int port,struct sockaddr_in* addr){ bzero(addr,sizeof(*addr)); addr->sin_family=AF_INET; inet_pton(AF_INET,ip,&(addr->sin_addr)); addr->sin_port=htons(port); } static std::string address_to_string(struct sockaddr_in* addr){ char ip[128]; inet_ntop(AF_INET,&(addr->sin_addr),ip,sizeof(ip)); char port[32]; snprintf(port,sizeof(port),"%d",ntohs(addr->sin_port)); std::string r; r=r+"("+ip+":"+port+")"; return r; } static int new_tcp_server(int port){ int fd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP); D(fd>0,"socket failed(%m)\n"); setnonblock(fd); setreuseaddr(fd); sockaddr_in addr; setaddress("0.0.0.0",port,&addr); bind(fd,(struct sockaddr*)&addr,sizeof(addr)); listen(fd,64); // backlog = 64 return fd; } static int new_tcp_client(const char* ip,int port){ int fd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP); setnonblock(fd); sockaddr_in addr; setaddress(ip,port,&addr); connect(fd,(struct sockaddr*)(&addr),sizeof(addr)); return fd; } }; // namespace common #endif // _COMMON_H_
1.5.2echo-client.cc
#include "ev.h" #include "common.h" static void do_connected(struct ev_loop* reactor,ev_io* w,int events){ close(w->fd); ev_break(reactor,EVBREAK_ALL); } int main(){ struct ev_loop* reactor=ev_loop_new(EVFLAG_AUTO); int fd=common::new_tcp_client("127.0.0.1",34567); ev_io io; ev_io_init(&io,&do_connected,fd,EV_WRITE); ev_io_start(reactor,&io); ev_run(reactor,0); close(fd); ev_loop_destroy(reactor); return 0; }
1.5.3echo-server.cc
#include "ev.h" #include "common.h" static void do_accept(struct ev_loop* reactor,ev_io* w,int events){ struct sockaddr_in addr; socklen_t addr_size=sizeof(addr); int conn=accept(w->fd,(struct sockaddr*)&addr,&addr_size); std::string r=common::address_to_string(&addr); fprintf(stderr,"accept %s\n",r.c_str()); close(conn); } int main(){ struct ev_loop* reactor=ev_loop_new(EVFLAG_AUTO); int fd=common::new_tcp_server(34567); ev_io w; ev_io_init(&w,do_accept,fd,EV_READ); ev_io_start(reactor,&w); ev_run(reactor,0); close(fd); ev_loop_destroy(reactor); }