神刀安全网

Libev事件库源码阅读笔记

Libev是一个基于Reactor模式的事件库,效率较高( Benchmark )并且代码精简(4.15版本8000多行),是学习事件驱动编程的很好的资源。

本文不会介绍Reactor模式,也不会介绍Libev的API,主要内容是我学习libev后的一些总结,介绍了Livev的设计方法和实现方法,并对一部分核心代码进行了注解。

如需更详尽的API介绍,可以参见Libev的 手册

Feature

Libev是一个用C编写的功能齐全的高性能的轻量级事件驱动库,其支持多种后台IO复用接口,并且可以注册多达十几种事件。

支持的后台IO复用接口:

select
poll
epoll
kqueue
solaris event port

支持的事件类型:

ev_io                 // IO可读可写
ev_stat // 文件属性变化
ev_signal // 信号处理
ev_timer // 相对定时器
ev_periodic // 绝对定时器
ev_child // 子进程状态变化
ev_fork // fork事件
ev_cleanup // event loop退出触发事件
ev_idle // event loop空闲触发事件
ev_embed // 嵌入另一个后台循环
ev_prepare // event loop之前事件
ev_check // event loop之后事件
ev_async // 线程间异步事件

Sample

在介绍Libev的代码结构之前,先看一个Libev手册中自带的例子,其注册了两个事件,一个timeout事件,一个io事件,任何一个事件发生后都会调用回调函数并终止主循环。这也是事件驱动编程的标准模式——注册事件后等待其触发并调用回调函数。

代码中的注释已经十分详细,我就不再赘述。要编译这段代码(evtest.c),首先在 Libev主页 下载Libev源码编译安装,然后使用 gcc evtest.c -libev 编译(记得ldconfig)。

evtest.c
// a single header file is required
#include <ev.h>

#include <stdio.h> // for puts

// every watcher type has its own typedef'd struct
// with the name ev_TYPE
ev_io stdin_watcher;
ev_timer timeout_watcher;

// all watcher callbacks have a similar signature
// this callback is called when data is readable on stdin
static void
stdin_cb (EV_P_ ev_io *w, int revents)
{

puts ("stdin ready");
// for one-shot events, one must manually stop the watcher
// with its corresponding stop function.
ev_io_stop (EV_A_ w);

// this causes all nested ev_run's to stop iterating
ev_break (EV_A_ EVBREAK_ALL);
}

// another callback, this time for a time-out
static void
timeout_cb (EV_P_ ev_timer *w, int revents)
{

puts ("timeout");
// this causes the innermost ev_run to stop iterating
ev_break (EV_A_ EVBREAK_ONE);
}

int
main (void)
{

// use the default event loop unless you have special needs
struct ev_loop *loop = EV_DEFAULT;

// initialise an io watcher, then start it
// this one will watch for stdin to become readable
ev_io_init (&stdin_watcher, stdin_cb, /*STDIN_FILENO*/ 0, EV_READ);
ev_io_start (loop, &stdin_watcher);

// initialise a timer watcher, then start it
// simple non-repeating 5.5 second timeout
ev_timer_init (&timeout_watcher, timeout_cb, 5.5, 0.);
ev_timer_start (loop, &timeout_watcher);

// now wait for events to arrive
ev_run (loop, 0);

// break was called, so exit
return 0;
}

Main Structures

Wather

事件驱动库中,很重要的一部分就是对事件的封装。在Libev中,事件被封装在Watcher结构体中,通过注册Watcher并指定对应的回调函数等参数,就可以将事件添加到主循环中。

先解释一下 EV_PEV_P_EV_AEV_A_ 这几个宏,在代码中几乎随处可见,主要是为了简化单线程模式下的函数调用的接口,这几个宏定义如下。

#if EV_MULTIPLICITY
struct ev_loop;
# define EV_P struct ev_loop *loop /* a loop as sole parameter in a declaration */
# define EV_P_ EV_P, /* a loop as first of multiple parameters */
# define EV_A loop /* a loop as sole argument to a function call */
# define EV_A_ EV_A, /* a loop as first of multiple arguments */
#else
# define EV_P void
# define EV_P_
# define EV_A
# define EV_A_
#endif

ev_loop 是主循环,而 EV_MULTIPLICITY 是一个条件编译的宏,表明是否支持有多个 ev_loop 实例存在,一般来说,每个线程中有且仅有一个 ev_loop 实例。如果整个程序是单线程的,程序中使用全局默认的 ev_loop 即可,不需要在函数中传参。而在多线程中调用函数很多时候都要指定函数操作的loop。比如启动一个io事件,调用的函数是 void ev_io_start (EV_P_ ev_io *w) ,如果没有定义 EV_MULTIPLICITY ,将会编译成 ev_io_start(io *w) ,否则会编译成 ev_io_start(struct ev_loop *loop, ev_io *w)

对于每一种事件,都有结构体 ev_TYPE 与之对应,比如 ev_ioev_timer 等。为了统一事件结构,libev在C中使用结构体布局实现了多态,可以将 ev_watcher 结构体看做所有 ev_TYPE 结构体的基类,它包含了所有 ev_TYPE 中相同的字段。

代码中对这些结构体的定义如下,为了便于理解,我对部分宏进行了还原。之所以只还原部分宏而不是全部,是因为这些宏体现了作者设计这些结构体的思路。

相关的宏

//这个宏定义了所有ev_TYPE开头的部分
#define EV_WATCHER(type) /
int active; /* private */ /
int pending; /* private */ /
int priority; /* private */ /
void *data; /* rw */ /
void (*cb)(EV_P_ struct type *w, int revents); /* private */


//这个宏在EV_WATCHER的基础上加了一个时间戳,主要用来定义和定时器有关的ev
#define EV_WATCHER_TIME(type) /
EV_WATCHER (type) /
ev_tstamp at; /* private */


//这个宏在EV_WATCHER的内容后加了一个next以构成事件链表
#define EV_WATCHER_LIST(type) /
EV_WATCHER (type) /
struct ev_watcher_list *next; /* private */

“基类”

//内容就是EV_WATCHER宏的内容,可以理解为"基类"
typedef struct ev_watcher
{
EV_WATCHER (ev_watcher)
} ev_watcher;

//内容就是EV_WATCHER_TIME宏的内容,也可以理解为"基类"
typedef struct ev_watcher_time
{
EV_WATCHER_TIME (ev_watcher_time)
} ev_watcher_time;

//可以理解为一个带有next指针的基类
typedef struct ev_watcher_list
{
EV_WATCHER_LIST (ev_watcher_list)
} ev_watcher_list;

“派生类”,这里只列举了 ev_ioev_timerev_signal ,这三种是比较常用的事件,其它事件结构的代码都差不多,具体可以见源码。

//ev_io 封装io事件的"派生类",结构体前部就是宏EV_WATCHER_LIST,fd和events是"派生类"变量
typedef struct ev_io
{
EV_WATCHER_LIST (ev_io)

int fd; /* ro */
int events; /* ro */
} ev_io;

//ev_signal 封装信号事件的"派生类",同样也有signum是"派生类"变量
typedef struct ev_signal
{
EV_WATCHER_LIST (ev_signal)

int signum; /* ro */
} ev_signal;

//ev_timer 封装相对定时器事件的"派生类",定时器用堆管理,不需要next指针
typedef struct ev_timer
{
EV_WATCHER_TIME (ev_timer)

ev_tstamp repeat; /* rw */
} ev_timer;
```
从以上代码可以看出来,每个事件结构体中的共有字段表示了这个事件的状态,优先级,参数,以及回调函数,而私有字段则是该类型事件的特有信息,比如io事件有对应的fd、定时器事件有发生时间等。

另外还有一个叫做ev_any_watcher的union可以容纳所有的事件类型。
```c
union ev_any_watcher
{
struct ev_watcher w;
struct ev_watcher_list wl;
struct ev_io io;
struct ev_timer timer;
struct ev_periodic periodic;
struct ev_signal signal;
struct ev_child child;
struct ev_stat stat;
struct ev_idle idle;
struct ev_prepare prepare;
struct ev_check check;
struct ev_fork fork;
struct ev_cleanup cleanup;
struct ev_embed embed;
struct ev_async async;
};

ev_loop

ev_loop 是个十分重要也非常庞大的结构体,可以称其为事件控制器,事件的调度基本都是由它控制的。

该结构体的定义十分晦涩,从下面的代码可以看出,代码会根据 EV_MULTIPLICITY 是否定义进行条件编译,在单线程环境下,因为只有一个loop,所以所有变量直接作为全局变量使用,而在多线程模式下会有多个loop实例,因此需要将变量封装在 ev_loop 结构体中,调用函数时要指定所操作的loop。这些变量定义在 ev_vars.h 中,通过include展开。

另外,在多线程模式下,定义了 ev_loop 结构体之后,还include了 ev_wrap.h ,这个文件中对 ev_vars.h 中的所有变量定义了一堆形如 #define anfds ((loop)->anfds) 的宏,这个宏的目的是为了统一代码的编写,在未开启 EV_MULTIPLICITYanfds 表示的就是全局变量 anfds ,而在开启了 EV_MULTIPLICITY 后,函数一般会传一个 struct ev_loop *loop

anfds 也会展开成 ((loop)->anfds) 。这使得代码中不用再写一堆的 #if #else #endif ,但也让代码变的更加晦涩难懂。

#if EV_MULTIPLICITY
struct ev_loop
{
ev_tstamp ev_rt_now;
#define ev_rt_now ((loop)->ev_rt_now)
#define VAR(name,decl) decl;
#include "ev_vars.h"
#undef VAR
};
#include "ev_wrap.h"

static struct ev_loop default_loop_struct;
EV_API_DECL struct ev_loop *ev_default_loop_ptr = 0; /* needs to be initialised to make it a definition despite extern */
#else
EV_API_DECL ev_tstamp ev_rt_now = 0; /* needs to be initialised to make it a definition despite extern */
#define VAR(name,decl) static decl;
#include "ev_vars.h"
#undef VAR

static int ev_default_loop_ptr;
#endif

ANFD

在管理io事件的时候,如何根据fd快速找到与其相关的事件,是一个需要考虑的问题。Libev的方法是用一个数组来存所有fd信息的结构体,然后以fd值为索引直接找到对应的结构体,这个结构体就是下面的ANFD结构体(省略了有关Windows系统的变量)。这种方法可以在O(1)复杂度内进行索引,问题是它占的空间有多少?假如我们同时开了一百万个fd,所占空间一共是 10^6*sizeof(ANFD) ,大约12M左右,这完全是可以接受的。

结构体中字段的含义在后面介绍Libev流程时会逐渐提到。

typedef ev_watcher_list *WL;
/* file descriptor info structure */
typedef struct
{
WL head;
unsigned char events; /* the events watched for */
unsigned char reify; /* flag set when this ANFD needs reification (EV_ANFD_REIFY, EV__IOFDSET) */
unsigned char emask; /* the epoll backend stores the actual kernel mask in here */
unsigned char unused;
#if EV_USE_EPOLL
unsigned int egen; /* generation counter to counter epoll bugs */
#endif
} ANFD;

How it works?

这节主要介绍了libev的代码是怎样工作的,主要分为事件注册,事件调度与后台I/O复用三部分。

事件注册

事件注册,也就是告诉事件驱动器程序要关注某个事件的发生。这里以io事件为例来分析怎样去注册以及销毁一个事件,其它事件的代码逻辑基本上都是一样的,就不再赘述。

从上面Sample里的代码中我们可以看到,启动一个io事件调用了了以下这两个函数:

ev_io stdin_watcher;
ev_io_init (&stdin_watcher, stdin_cb, /*STDIN_FILENO*/ 0, EV_READ);
ev_io_start (loop, &stdin_watcher);

首先看一下 ev_io_init ,与其相关的代码主要有以下几个宏。基本就是初始化了 ev_io 结构体中各个字段的值,优先级会被初始化为0,如需改变需要单独调用 ev_set_priority 。上面的 ev_io_init ,实际上就是注册了一个关注读的IO事件,相应的fd为0也就是标准输入。

#define ev_io_init(ev,cb,fd,events)          do { ev_init ((ev), (cb)); ev_io_set ((ev),(fd),(events)); } while (0)
#define ev_init(ev,cb_) do { /
((ev_watcher *)(void *)(ev))->active = /
((ev_watcher *)(void *)(ev))->pending = 0; /
ev_set_priority ((ev), 0); /
ev_set_cb ((ev), cb_); /
} while (0)

#define ev_io_set(ev,fd_,events_) do { (ev)->fd = (fd_); (ev)->events = (events_) | EV__IOFDSET; } while (0)

然后是 ev_io_start ,核心工作就是将 ev_io 添加到相关fd的事件链表上去,下面是主要代码,我在代码中给出了较为详细的注释。其中noinline等都是作者因为编译器差异定义的一些宏,而 EV_FREQUENT_CHECK 也是为了验证程序正确性添加的宏,在生产环境下不会生成任何内容,这些宏现在都可以忽略,我们只关心主要逻辑。

typedef ev_watcher *W;
typedef ev_watcher_list *WL;

void noinline
ev_io_start (EV_P_ ev_io *w) EV_THROW
{

int fd = w->fd;
//如果对一个ev_io已经调用过ev_io_start,不再重复调用
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);

//判断当前分配的fd数组是否能放该fd,若不能,分配更多的空间给fd数组(realloc)。
//这是一个宏,anfds实际上就是loop->anfds,存储了和该loop关联的ANFD数组。
//anfdmax是当前数组能放的最大fd,array_init_zero也是个宏,对新分配的空间memeset成0。
array_needsize (ANFD, anfds, anfdmax, fd + 1, array_init_zero);

//将该io事件添加到该fd对应结构体的事件链表里,使用头插法
wlist_add (&anfds[fd].head, (WL)w);

/* common bug, apparently */
assert (("libev: ev_io_start called with corrupted watcher", ((WL)w)->next != (WL)w));

//将该fd加到fdchanges数组里,fdchanges保存了事件监听状态改变的fd
//事件驱动器会在合适的时候遍历fdchanges数组,根据使用的后台IO复用机制应用更改,比如epoll_ctl
fd_change (EV_A_ fd, w->events & EV__IOFDSET | EV_ANFD_REIFY);

//start结束,取消该事件的IOFDSET标记(这个标记是在init时加的)
w->events &= ~EV__IOFDSET;

EV_FREQUENT_CHECK;
}

//设置事件状态为启动
inline_speed void
ev_start (EV_P_ W w, int active)
{

//调整优先级到合法范围
pri_adjust (EV_A_ w);
//active==1表示已经启动
w->active = active;
//增加该ev_loop的reference
ev_ref (EV_A);
}

//添加事件到对应的链表中
inline_size void
wlist_add (WL *head, WL elem)
{

elem->next = *head;
*head = elem;
}

//标记fd监听状态有所更改
inline_size void
fd_change (EV_P_ int fd, int flags)
{

unsigned char reify = anfds [fd].reify;
anfds [fd].reify |= flags;
//reify之前是0,要将fd加到fdchanges数组中
if (expect_true (!reify))
{
++fdchangecnt;
array_needsize (int, fdchanges, fdchangemax, fdchangecnt, EMPTY2);
fdchanges [fdchangecnt - 1] = fd;
}
}

最后看一下 ev_io_stop ,逻辑基本就是 ev_io_start 的反过程。

void noinline
ev_io_stop (EV_P_ ev_io *w) EV_THROW
{

//如果该事件正在pending(等待执行的事件)中,从pending列表中移除该事件。
//这里的一个技巧是不用真的移除掉(数组删除复杂度O(n)),只要将pending列表对应位置的指针指向一个空事件就可以了。
clear_pending (EV_A_ (W)w);
if (expect_false (!ev_is_active (w)))
return;

assert (("libev: ev_io_stop called with illegal fd (must stay constant after start!)", w->fd >= 0 && w->fd < anfdmax));

EV_FREQUENT_CHECK;
//从链表中删除一个节点,这里删除节点的方法也是Linus提倡的方法,不需要记录prev指针以及是否头节点什么的。
wlist_del (&anfds[w->fd].head, (WL)w);

//取消fd的active状态
ev_stop (EV_A_ (W)w);

//将fd加到fdchanges数组中,只设置REIFY标记,表示有改动(若什么标记都不设置该fd不会放入fdchanges)
//之后事件驱动器扫描fdchanges数组会发现该fd不再监听任何事件,作出相应操作
fd_change (EV_A_ w->fd, EV_ANFD_REIFY);

EV_FREQUENT_CHECK;
}
//从链表中删除一个节点,十分经典的方法
inline_size void
wlist_del (WL *head, WL elem)
{

while (*head)
{
if (expect_true (*head == elem))
{
*head = elem->next;
break;
}
head = &(*head)->next;
}
}

Main Loop

ev_io 系列函数所做的操作基本就是填充 ev_io 结构体并将其放在对应fd的事件链表上,而将监听事件状态发生改变的fd存在fdchanges数组中,驱动控制器会根据该数组更改后台监听的事件,当事件发生时驱动控制器会自动调用相应事件的回调函数。在整个过程中,驱动控制器也就是 ev_loop 就像胶水一样将事件和后台复用机制粘在一起。

整个事件的调度过程基本都在函数 ev_run 中,整个函数较长,大概有两百多行,这里就不列出代码了,只把程序的主要逻辑写了出来。

int
ev_run (EV_P_ int flags)
{

...
//一直循环....
do
{
...
//如果这个进程是新fork出来的,执行ev_fork事件的回调
...
//执行ev_prepare回调,也就是每次poll前执行的函数
...
//执行监听有改变的事件
...
//计算poll应该等待的时间,这个时间和设置以及定时器超时时间有关
...
//调用后台I/O复用端口等待事件触发
backend_poll (EV_A_ waittime);
...
//将定时器事件放入pending数组中
...
//将ev_check事件翻入pending数组中
...
//执行pending数组中所有的回调
EV_INVOKE_PENDING;
}
while (条件成立);
}

ev_run 的逻辑可以说还是比较清晰的。程序首先会先执行一些需要在poll之前执行的回调,接着根据最先超时的计时器算出poll需要wait的时间,之后调用poll等待I/O事件发生,最后执行发生事件的回调。

具体的代码中,程序使用 queue_events 将要运行的事件放入一个叫做pending的二维数组中,其第一维是优先级,第二维是动态分配的,存放具体事件。之后程序会在适当的地方调用宏 EV_INVOKE_PENDING ,将pending数组中的事件按优先级从高到低依次执行。

I/O复用

Libev使用函数指针来实现支持多种I/O复用机制,每种复用机制要实现init, modify, poll, destroy这几个函数,也就是初始化、修改关注事件、等待事件发生、销毁这几个功能。这部分代码我只看了epoll的实现,感觉实现的还是很巧妙的,读者可以根据自己熟悉的I/O复用机制去选择看哪部分代码。

More

至此,Libev的主要设计方法和实现思路基本介绍的差不多了,限于篇幅,还有很多细节无法在一篇文章中叙述完,如果有时间的话,我会尽量完善这篇文章。

作为一个事件库,Libev的设计可以说是十分精良,代码中的各种tips也让我受益良多。但是,Libev几乎不涉及网络编程,如果要在此基础上实现网络库,还是有大量工作要做的。

转载本站任何文章请注明:转载至神刀安全网,谢谢神刀安全网 » Libev事件库源码阅读笔记

分享到:更多 ()

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址