libevent源码剖析

时间 2019-11-21

标签 libevent 源码剖析繁體版

原文原文链接

　　libevent是一个使用C语言编写的，轻量级的开源高性能网络库，使用者不少，研究者也不少。因为代码简洁，设计思想简明巧妙，所以很适合用来学习，提高本身C语言的能力。数组

　　libevent有这样显著地几个亮点：缓存

　　　　1.事件驱动，高性能安全

　　　　2.轻量级，专一于网络，不如ACE那么庞大臃肿网络

　　　　3.代码精炼易读多线程

　　　　4.跨平台，支持Windows,Linux,*BSD和Mac Os;框架

　　　　5.支持多种IO多路复用技术，epoll,poll，dev/poll、select和kqueue等异步

　　　　6.支持IO，定时器和信号等事件socket

　　　　7.注册事件优先级ide

　　基于以上优势，libevent已经被普遍的应用，做为底层的网络库；好比memcached、Vomit、Nylon、Netchat等。memcached

　　下面咱们就来从程序的基本使用场景和代码的总体处理流程入手来对libevent库进行学习。

　　　　　当应用程序向libevent注册一个事件后,libevent内部是怎样处理的呢，如下是基本流程：

　　　　　　　　1)首先应用程序准备并初始化event，设置好事件类型和回调函数

　　　　　　　　2)向libevent添加该事件event。

　　　　　　　　3)程序调用event_base_dispatch()系列函数进入无线循环，等待事件.

　　　　　这只是大概的流程，如下是对于流程中的一些基本概念的讲解：

　　　　　1、事件event

　　　　　　libevent是基于事件驱动的，从名字上能够看出event是整个库的核心。首先给出event结构体的声明,它位于event.h文件中：

 1 struct event {
 2  TAILQ_ENTRY (event) ev_next;
 3  TAILQ_ENTRY (event) ev_active_next;
 4  TAILQ_ENTRY (event) ev_signal_next;
 5  unsigned int min_heap_idx; /* for managing timeouts */
 6  struct event_base *ev_base;
 7  int ev_fd;
 8  short ev_events;
 9  short ev_ncalls;
10  short *ev_pncalls; /* Allows deletes in callback */
11  struct timeval ev_timeout;
12  int ev_pri;  /* smaller numbers are higher priority */
13  void (*ev_callback)(int, short, void *arg);
14  void *ev_arg;
15  int ev_res;  /* result passed to event callback */
16  int ev_flags;
17 };

　　　　　　　　1) ev_events：event关注的事件类型，它能够是如下3种类型：

　　　　　　　　　　IO事件：EV_WRITE和EV_READ

　　　　　　　　　　定时事件：EV_TIMEOUT

　　　　　　　　　　信号：EV_SIGNAL

　　　　　　　　　　辅助选项：EV_PERSIST，代表是一个永久事件

　　　　　　　　2) ev_next,ev_active_next,ev_signal_next都是双向链表节点指针；他们是libevent对不一样事件类型和在不一样的时期，对事件的管理时使用到的字段。

　　　　　　　　　　libevent使用双向链表保存全部注册的IO和Signal事件，ev_next就是该IO事件在链表中的位置，称此链表为“已注册事件链表”；

　　　　　　　　　　一样ev_signal_next就是signal事件的signal事件链表中的位置。

　　　　　　　　　　ev_active_next:libevent将全部激活事件放入到链表active list中，而后遍历active list执行调度。

　　　　　　　　　　每当事件event转变为就绪状态时，libevent就会把它移入到active event list[priority]中，其中priority是event的优先级；接着libevent会根据本身的调度策略选择就绪事件，调用其callback()函数执行事件处理；并根据就绪的句柄和时间类型填充callback函数的参数。

　　　　　　　　3)min_heap_idx和ev_timeout：若是事件是timeout事件，他们是event在小根堆中的索引和超时值，libevent使用小根堆来管理定时事件

　　　　　　　　4)ev_base：该事件所属反应堆实例，这是一个event_base结构体。

　　　　　　　　5)ev_fd：对于IO事件，这是绑定的文件描述符，对于signal事件，是绑定的信号

　　　　　　　　6)ev_callback：event的回调函数，被ev_base调用，执行事件处理程序，这是一个函数指针，原型为：

　　　　　　　　　　void (*ev_callback)(int fd,short events,void* arg)

　　　　　　　　　其中参数fd对应于ev_fd；events对应于events；arg对应于ev_arg；

　　　　　　　　7)ev_arg：void*，代表能够是任意类型的数据，在设置event时指定；

　　　　　　　　8)eb_flags：libevent用于标记event信息的字段，代表其当前的状态，可能的值有：

　　　　　　　　　　#define EVLIST_TIMEOUT 0x01 // event在time堆中
　　　　　　　　　　#define EVLIST_INSERTED 0x02 // event在已注册事件链表中
　　　　　　　　　　#define EVLIST_SIGNAL 0x04 // 未见使用
　　　　　　　　　　#define EVLIST_ACTIVE 0x08 // event在激活链表中
　　　　　　　　　　#define EVLIST_INTERNAL 0x10 // 内部使用标记
　　　　　　　　　　#define EVLIST_INIT 0x80 // event已被初始化

　　　　　　　　9)ev_ncalls：事件就绪执行时，调用ev_callback的次数，一般为1

　　　　　　　　10)ev_res：记录了当前激活事件的类型

　　　　　　要想向libevent添加一个事件，首先须要设置event对象，这经过调用libevent提供的函数有：event_set()，event_base_set()，event_priority_set()来完成；下面分别讲解：

　　　　　　void event_set(struct event *ev,int fd,short events,void (*callback)(int,short,void*),void *arg)

　　　　　　　　ev:执行要初始化的event对象

　　　　　　　　fd:对于信号来讲是绑定的文件描述符，对于信号来讲是绑定的signal信号

　　　　　　　　events:在该fd上关注的事件类型，它能够是EV_READ,EV_WRITE,EV_SIGNAL；

　　　　　　　　callback:这是一个函数指针，当fd上的事件event发生时，调用该函数执行处理

　　　　　　　　arg:传递给callback函数指针的参数

　　　　　　int event_base_set(struct event_base* base,struct event *ev)

　　　　　　　　设置event ev将要注册到的evnet_base;

　　　　　　int event_priority_set(struct event* ev,int pri)

　　　　　　　　设置event ev的优先级，注意，当ev正处于就绪状态时，不能设置，返回-1.

　　　　2、事件处理框架event_base

　　　　　　如下是base_event结构体的声明,它位于event-internal.h文件中：

 1 struct event_base {
 2  const struct eventop *evsel;
 3  void *evbase;　
 4  int event_count;  /* counts number of total events */
 5  int event_count_active; /* counts number of active events */
 6  int event_gotterm;  /* Set to terminate loop */
 7  int event_break;  /* Set to terminate loop immediately */
 8  /* active event management */
 9  struct event_list **activequeues;
10  int nactivequeues;
11  /* signal handling info */
12  struct evsignal_info sig;
13  struct event_list eventqueue;
14  struct timeval event_tv;
15  struct min_heap timeheap;
16  struct timeval tv_cache;
17 };

　　　　　　　　如下是结构体各字段的含义：

　　　　　　　　1)evsel和evbase这两个字段的设置可能会让人有些迷惑，这里咱们能够把evsel和evbase看作是类和静态函数的关系，好比添加事件时的调用行为：evsel->add(evbase,ev)，实际执行操做的是evbase，这至关于class::add(instance,ev)，instance就是class的一个对象实例。

　　　　　　　　2)activequeues是一个二级指针，前面讲过libevent支持事件优先级，所以你能够你把它看作是数组，其中的元素activequeues[priority]是一个链表，链表的每一个节点指向一个优先级为priority的就绪事件event。

　　　　　　　　3)eventqueue：链表，保存了全部的注册事件event的指针。

　　　　　　　　4)sig是用来管理信号的结构体

　　　　　　　　5)timeheap是管理定时事件的小根堆

　　　　　　　　6)event_tv和tv_cache是libevent用于事件管理的变量

　　　　　　咱们已经对event_base有了一个初步的了解，那么event_base如何建立和初始化的呢？

　　　　　　　　建立一个event_base对象也便是建立了一个新的libevent实例，程序须要经过调用event_init()函数来建立，该函数首先为event_base实例申请空间，而后初始化timer mini-heap，选择并初始化合适的系统多路复用机制，初始化各事件的链表；函数还检测了系统的时间设置，为后面的事件管理打下了基础

　　　　　3、事件主循环

　　　　　　　　libevent将IO事件、定时器和信号事件处理很好的结合到了一块儿，那么它是如何作到的呢？

　　　　　　　　libevent的事件主循环主要是经过event_base_loop()函数完成的，主要操做流程以下图所示，event_base_loop所作的就是持续执行下面的循环

　　　　　　　　下面是源码，能够参考

int event_base_loop(struct event_base *base, int flags)
{
    const struct eventop *evsel = base->evsel;
    void *evbase = base->evbase;
    struct timeval tv;
    struct timeval *tv_p;
    int res, done;
    // 清空时间缓存
    base->tv_cache.tv_sec = 0;
    // evsignal_base是全局变量，在处理signal时，用于指名signal所属的event_base实例
    if (base->sig.ev_signal_added)
        evsignal_base = base;
    done = 0;
    while (!done) { // 事件主循环
        // 查看是否须要跳出循环，程序能够调用event_loopexit_cb()设置event_gotterm标记
        // 调用event_base_loopbreak()设置event_break标记
        if (base->event_gotterm) {
            base->event_gotterm = 0;
            break;
        }
        if (base->event_break) {
            base->event_break = 0;
            break;
        }
        // 校订系统时间，若是系统使用的是非MONOTONIC时间，用户可能会向后调整了系统时间
        // 在timeout_correct函数里，比较last wait time和当前时间，若是当前时间< last wait time
        // 代表时间有问题，这是须要更新timer_heap中全部定时事件的超时时间。
        timeout_correct(base, &tv);
   
        // 根据timer heap中事件的最小超时时间，计算系统I/O demultiplexer的最大等待时间
        tv_p = &tv;
        if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) {
            timeout_next(base, &tv_p);
        } else {
            // 依然有未处理的就绪时间，就让I/O demultiplexer当即返回，没必要等待
            // 下面会提到，在libevent中，低优先级的就绪事件可能不能当即被处理
            evutil_timerclear(&tv);
        }
        // 若是当前没有注册事件，就退出
        if (!event_haveevents(base)) {
            event_debug(("%s: no events registered.", __func__));
            return (1);
        }
        // 更新last wait time，并清空time cache
        gettime(base, &base->event_tv);
        base->tv_cache.tv_sec = 0;
        // 调用系统I/O demultiplexer等待就绪I/O events，多是epoll_wait，或者select等；
        // 在evsel->dispatch()中，会把就绪signal event、I/O event插入到激活链表中
        res = evsel->dispatch(base, evbase, tv_p);
        if (res == -1)
            return (-1);
        // 将time cache赋值为当前系统时间
        gettime(base, &base->tv_cache);
        // 检查heap中的timer events，将就绪的timer event从heap上删除，并插入到激活链表中
        timeout_process(base);
        // 调用event_process_active()处理激活链表中的就绪event，调用其回调函数执行事件处理
        // 该函数会寻找最高优先级（priority值越小优先级越高）的激活事件链表，
        // 而后处理链表中的全部就绪事件；
        // 所以低优先级的就绪事件可能得不到及时处理；
        if (base->event_count_active) {
            event_process_active(base);
            if (!base->event_count_active && (flags & EVLOOP_ONCE))
                done = 1;
        } else if (flags & EVLOOP_NONBLOCK)
            done = 1;
    }
    // 循环结束，清空时间缓存
    base->tv_cache.tv_sec = 0;
    event_debug(("%s: asked to terminate loop.", __func__));
    return (0);
}

View Code

　　　　　　　I/O和Timer事件的统一

　　　　　　　　　libevent将Timer和Signal事件都统一到了系统的IO多路复用机制中了从上面的流程图中咱们能够看出一点端倪，那么libevent是如何作到的呢？

　　　　　　　　　首先是I/O和Timer事件的统一。系统的I/O机制向select()和epoll_wait()都容许程序制定一个最大等待时间timeout,即便没有事件发生，他们也能保证在timeout时间内返回。那么根据全部Timer事件的最小超时时间来设置系统I/O的timeout时间；当系统I/O返回时，再激活全部就绪的Timer事件就能够了，这样就能将Timer事件完美的融合到系统的I/O机制中了。

　　　　　　　　　libevent使用堆来管理Timer事件，其key值就是事件的超时时间，根据堆中具备最小超时值的事件和当前时间来计算等待时间。

　　　　　　　I/O和Signal事件的统一

　　　　　　　　　Signal是异步事件的经典，将Signal事件统一到系统的I/O多路复用中就不像Timer事件那么天然了。Signal事件的出现对于进程来说彻底是随机的，进程不能只是测试一个变量来判别是否发生了一个信号，而是必须告诉内核“在此信号发生时，请执行以下操做”。当Signal发生时，系统并不当即调用event的callback函数处理信号，而是设法通知系统的I/O机制，让其返回，而后再统一和I/O事件以及Timer一块儿处理。

　　　　　　　　　　那么,系统是如何设法通知系统的I/O机制呢?基本的方法就是采用“消息机制”。在libevent中是经过socket pair完成的。socket pair就是一个socket对，一个读socket，一个写socket。Socket pair建立好了以后，读socket在libevent的event_base实例上注册了一个persist的读事件。这样当写socket写入数据时，读socket就会相应的获得通知了。前面提到过,libevent会在事件主循环中检查标记，来肯定是否有触发的Signal，若是标记被设置就处理这些signal。这段代码在各个具体的I/O机制中，以epoll为例，在epoll_dispatch()函数中，代码片断以下：

　　　　　　　　　　res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);
   　　　　　　　　　 if (res == -1) {
      　　　　　　　　　　 if (errno != EINTR) {
         　　　　　　　　　   event_warn("epoll_wait");
           　　　　　　　　　 return (-1);
        　　　　　　　　}
      　　　　　　　　 evsignal_process(base);// 处理signal事件
       　　　　　　　　 return (0);
   　　　　　　　　 } else if (base->sig.evsignal_caught) {
      　　　　　　　　 evsignal_process(base);// 处理signal事件
   　　　　　　　　 }

　　　　　　　　注册signal事件是经过evsignal_add(struct event *ev)函数完成的,libevent对全部的信号注册同一个处理函数evsignal_handler()，该函数注册过程以下：

　　　　　　　　1.取得ev要注册到的信号signo；

　　　　　　　　2.若是信号signo未被注册，那么就为signo注册信号处理函数evsignal_handler();

　　　　　　　　3.若是事件ev_signal尚未注册，就注册ev_signal事件；

　　　　　　　　4.将事件ev添加到signo的event链表中

　　　　　　　　注销一个已注册的signal事件就更简单了，直接从其已注册事件的链表中移除便可。若是链表已空，那么就恢复旧有的处理函数；处理函数的evsignal_handler()函数就是记录信号的发生次数，并通知event_base有信号触发。

　　　　　3、支持多路复用

　　　　　　　　libevent的核心是事件驱动、同步非阻塞，为了达到这一目标必须采用系统提供的I/O多路复用技术，而这些复用技术在不一样的平台上却各有不一样，如何能提供统一的支持方式呢？

　　　　　　　　libevent支持多种I/O多路复用的关键就在于结构体evnetop，这个结构体前面也提到过，他的成员是一系列函数指针，定义在event-internal.h文件中：

　　　　　　　　struct eventop {
    　　　　　　　　　　const char *name;
    　　　　　　　　　　void *(*init)(struct event_base *); // 初始化
   　　　　　　　　　　 int (*add)(void *, struct event *); // 注册事件
   　　　　　　　　　　 int (*del)(void *, struct event *); // 删除事件
  　　　　　　　　　　 int (*dispatch)(struct event_base *, void *, struct timeval *); // 事件分发
    　　　　　　　　　　void (*dealloc)(struct event_base *, void *); // 注销，释放资源
   　　　　　　　　　　 /* set if we need to reinitialize the event base */
   　　　　　　　　　　 int need_reinit;
　　　　　　　　};

　　　　　　　在libevent中，每种IO多路复用技术的实现都必须提供这五种函数接口来完成自身的初始化、销毁释放；对事件的注册、注销和分发。好比对于epoll，libevent实现了5个对应的接口函数，并在初始化时经eventop的5个函数指针指向这5个函数，那么程序就可使用epoll做为IO多路复用机制了。

　　　　　4、时间管理

　　　　　　　　为了支持定时器，libevent必须需和系统时间打交道主要涉及到时间的加减辅助函数、时间缓存、时间校订和定时器堆的时间值调整等。

　　　　　　　　libevent在初始化时会检测系统时间的类型，经过调用detect_monotonic()完成，它经过clock_gettime()来检测系统是否支持monotonic时钟类型

　　　　　　　　　　static void detect_monotonic(void)
　　　　　　　　　　{
　　　　　　　　　　　　#if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC)
   　　　　　　　　　　　 struct timespec    ts;
　　　　　　　　　　　   if (clock_gettime(CLOCK_MONOTONIC, &ts) == 0)
     　　　　　　　　　　   use_monotonic = 1; // 系统支持monotonic时间
　　　　　　　　　　　　#endif
　　　　　　　　　　}

　　　　　　　　monotonic事件值得是系统从boot后到如今的时间，若是系统支持monotonic事件就将全局变量use_monotonic设置为1。

　　　　　　　　1.时间缓存

　　　　　　　　　　结构体event_base中的tv_cache，用来记录时间缓存。若是tv_cache已经设置，那么就直接使用缓存的时间，不然须要再次

执行系统调用获取系统时间

　　　　　　　　2.时间校订

　　　　　　　　　　若是系统支持monotonic时间，该时间是从boot后到如今所通过的时间，所以不须要执行校订。若是系统不支持monotonic时间，用户可能会手动调整时间，校订由函数timeout_correct()完成。

 1 static void timeout_correct(struct event_base *base, struct timeval *tv)
 2 {
 3     struct event **pev;
 4     unsigned int size;
 5     struct timeval off;
 6     if (use_monotonic) // monotonic时间就直接返回，无需调整
 7         return;
 8     gettime(base, tv); // tv <---tv_cache
 9     // 根据前面的分析能够知道event_tv应该小于tv_cache
10     // 若是tv < event_tv代表用户向前调整时间了，须要校订时间
11     if (evutil_timercmp(tv, &base->event_tv, >=)) {
12         base->event_tv = *tv;
13         return;
14     }
15     // 计算时间差值
16     evutil_timersub(&base->event_tv, tv, &off);
17     // 调整定时事件小根堆
18     pev = base->timeheap.p;
19     size = base->timeheap.n;
20     for (; size-- > 0; ++pev) {
21         struct timeval *ev_tv = &(**pev).ev_timeout;
22         evutil_timersub(ev_tv, &off, ev_tv);
23     }
24     base->event_tv = *tv; // 更新event_tv为tv_cache
25 }

　　　　　　　　在调整小根堆时，由于全部定时事件的时间值都会被减去相同的值，所以虽然堆中元素的时间键值被改变了，可是相对关系没有改变，不会改变堆的总体结构。所以只要遍历堆中的全部元素，将每一个元素的时间键值减去相同的值便可完成调整，不须要从新调整堆的结构。调整完后，要将event_tv值从新设置为tv_cache值。

　　　　5、libevent支持多线程

　　　　　　libevent不是线程安全的，可是这并不表示libevent不支持多线程模式，其实方法在前面已经将signal事件处理时就接触到了，那就是消息通知机制。如下：

　　　　　　　　1)暴力抢占

　　　　　　　　　　中止正在执行的任务，立刻去执行新来的任务。好处是消息能够当即获得处理，须要注意的是必须处理好线程切换问题。

　　　　　　　　2)纯粹的消息通知机制

　　　　　　　　　　执行完正在执行的任务，再执行新任务。经过消息通知，切换问题省心了，不过消息是不能当即处理的，并且全部的内容都是经过消息发送，增长了通讯的开销。

　　　　　　　　3)消息通知+同步层

　　　　　　　　　　有个折中的办法能够减小消息通讯的开销，就是提取一个同步层，把工做安排都存放在一个工做队列中，并且可以保证“任何人把新任务扔到这个队列”和“本身取出当前第一个任务”等这些操做都可以保证不会把队列搞乱

　　　　　　　　　　工做队列实际上就是一个加锁的容器（队列，链表），这个很容易实现，而消息通知仅需一个字节，具体的任务都push到了工做队列中，减小了开销