libev实现分析
libev是一个事件驱动库，底层是基于select、epoll、kqueue等I/O复用接口。所谓事件驱动库，就是用户定义一个事件以及改事件发生时调用的函数，该库会监听该事件，并在事件发生时调用相应的函数。
libev提供了很多事件监听器(watcher)，最主要的有IO、时间以及信号监听器。当某一个文件的读事件或者写事件发生时，周期时间到了时，进程接收到某个信号时，就会调用用户定义的回调函数。
下面以IO事件为例，讲述libev的工作原理：
1、实例
1 #include<stdio.h>2 #include <ev.h>3 // every watcher type has its own typedef'd struct4 // with the name ev_TYPE5 ev_io stdin_watcher;6 ev_timer timeout_watcher;7 8 // all watcher callbacks have a similar signature9 // this callback is called when data is readable on stdin
10 static void
11 stdin_cb (EV_P_ ev_io *w, int revents)
12 {
13 puts ("stdin ready OK!");
14 // for one-shot events, one must manually stop the watcher
15 // with its corresponding stop function.
16 ev_io_stop (EV_A_ w);
17
18 // this causes all nested ev_run's to stop iterating
19 ev_break (EV_A_ EVBREAK_ALL);
20 }
21
22 // another callback, this time for a time-out
23 static void
24 timeout_cb (EV_P_ ev_timer *w, int revents)
25 {
26 puts ("timeout");
27 // this causes the innermost ev_run to stop iterating
28 ev_break (EV_A_ EVBREAK_ONE);
29 }
30
31 int
32 main (void)
33 {
34 // use the default event loop unless you have special needs
35 struct ev_loop *loop = EV_DEFAULT;
36 // initialise an io watcher, then start it
37 // this one will watch for stdin to become readable
38 ev_io_init (&stdin_watcher, stdin_cb, /*STDIN_FILENO*/ 0, EV_READ);
39 ev_io_start (loop, &stdin_watcher);
40
41 // initialise a timer watcher, then start it
42 // simple non-repeating 5.5 second timeout
43 ev_timer_init (&timeout_watcher, timeout_cb, 5.5, 0.);
44 ev_timer_start (loop, &timeout_watcher);
45
46 // now wait for events to arrive
47 ev_run (loop, 0);
48 //
49 // break was called, so exit
50 return 0;
51 }
可以看出，libev库的使用简单、方便。我们只要定义个事件的监听器对象，初始化，开始，最后调用ev_run。不同事件监听器初始化的内容也不一样，比如，IO事件监听器需要初始化监听的文件描述符，事件以及回调函数。
2、事件监听器
typedef ev_watcher *W;
typedef ev_watcher_list *WL;
typedef ev_watcher_time *WT;typedef struct ev_watcher
{int active; int pending;int priority;void *data;void (*cb)(EV_P_ struct type *w, int revents);
} ev_watcher;typedef struct ev_watcher_list
{int active; int pending;int priority;void *data;void (*cb)(EV_P_ struct ev_watcher_list *w, int revents);struct ev_watcher_list *next;
} ev_watcher_list;typedef struct ev_io
{int active; int pending;int priority;void *data;void (*cb)(EV_P_ struct ev_io *w, int revents);struct ev_watcher_list *next;int fd; /* ro */int events; /* ro */
} ev_io;
ev_watcher是一个基础监听器，包括回调函数cd；监听器列表(ev_watcher_list)是在监听器的基础上添加指向下一个监听器的指针next；IO监听器是在监听器列表的基础上加上了其特有的文件描述符和事件类型。 ev_io_init (&stdin_watcher, stdin_cb, /*STDIN_FILENO*/ 0, EV_READ);该函数就是初始化stdin_watcher这个监听器的回调函数，文件描述符，以及事假类型。 3、struct ev_loop
struct ev_loop;
# define EV_P struct ev_loop *loop
# define EV_P_ struct ev_loop *loop,
# define EV_A loop
# define EV_A_ loop, //这4个宏用于形参struct ev_loop{ //部分参数ANFD *anfdsint anfdmaxint *fdchangesint fdchangemaxint fdchangecntANPENDING *pendings [NUMPRI]int pendingmax [NUMPRI]int pendingcnt [NUMPRI]int backendint backend_fdvoid (*backend_modify)(EV_P_ int fd, int oev, int nev)void (*backend_poll)(EV_P_ ev_tstamp timeout)
}typedef struct
{WL head; //ev_watcher_list *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;unsigned int egen; /* generation counter to counter epoll bugs */SOCKET handle;OVERLAPPED or, ow;
} ANFD;typedef struct
{W w; //ev_watcher *w;int events; /* the pending event set for the given watcher */
} ANPENDING;
ev_run函数主要是一个while循环，在这个while循环中不断检测各个事件是否发生，如果发生就调用其回调函数。而这个过程中，主要用到的对象就是struct ev_loop结构体对象，检测哪些事件，回调哪个函数都存放在该对象中。 struct ev_loop结构体中的字段很多，以IO事件为例介绍几个主要的： anfds是一个数组，数组元素是结构体ANFD，ANFD有一个成员是监听器列表。数组下标是文件描述符，而列表成员是监听该文件的事件监听器。所以，anfds有点类似散列表，以文件描述符作为键，以监听器作为值，采用开链法解决散列冲突。 该字段的初始化在ev_io_start函数中，主要目的是用户定义的监听器告诉ev_loop。 fdchanges是一个int数组，也是在ev_io_start中初始化。存放的是监听了的文件描述符。这样ev_run每次循环的时候，要先从fdchanges中取出已经监听的文件描述符，再以该描述符为下标，从anfds中取出监听器对象。这样就得到文件描述符以及监听 的事件。pendings是一个二维数组，第一维是优先级，第二维是监听器。这个数组是用于执行相应的回调函数，根据优先级，遍历所有监听器，调用监听器的回调函数。3、ev_io_start
ev_io_start (EV_P_ ev_io *w) EV_THROW
{ev_start (EV_A_ (W)w, 1); //w->active = active;array_needsize (ANFD, anfds, anfdmax, fd + 1, array_init_zero); //when fd + 1 > anfdmax : 重新分配数组大小//anfds = (type *)array_realloc(sizeof (ANFD), (anfds), &(anfdmax), (fd + 1));wlist_add (&anfds[fd].head, (WL)w); // w->next = *head;*head = w;fd_change (EV_A_ fd, w->events & EV__IOFDSET | EV_ANFD_REIFY); //nfds [fd].reify |= flags;//++fdchangecnt;//array_needsize (int, fdchanges, fdchangemax, fdchangecnt, EMPTY2);//fdchanges [fdchangecnt - 1] = fd;
}array_realloc (int elem, void *base, int *cur, int cnt)
{*cur = array_nextsize (elem, *cur, cnt);return ev_realloc (base, elem * *cur);
}//分配当前数组大小的两倍内存，如果大于4096，则为4096的倍数
int array_nextsize (int elem, int cur, int cnt)
{int ncur = cur + 1;doncur <<= 1;while (cnt > ncur);/* if size is large, round to MALLOC_ROUND - 4 * longs to accommodate malloc overhead */if (elem * ncur > MALLOC_ROUND - sizeof (void *) * 4) //#define MALLOC_ROUND 4096{ncur *= elem;ncur = (ncur + elem + (MALLOC_ROUND - 1) + sizeof (void *) * 4) & ~(MALLOC_ROUND - 1);ncur = ncur - sizeof (void *) * 4;ncur /= elem;}return ncur;
}
ev_io_start函数主要就是对ev_loop的anfds和fdchanges字段操作，上面已介绍。array_needsize函数实现当数组大小不够时，要重新分配内存，分配方式与stl::vector有些类似，都是新分配的内存为当前内存的2倍，然后移动原先数据到新内存，释放旧内存。
4、ev_run
ev_run (EV_P_ int flags)
{...do{fd_reify (EV_A);backend_poll (EV_A_ waittime);if (expect_false (checkcnt))queue_events (EV_A_ (W *)checks, checkcnt, EV_CHECK);EV_INVOKE_PENDING;}while(...)...
}
ev_run函数代码比较多，以上是以IO事件为例，进行的精简。下面是以epoll作为IO多路复用的机制进行ev_run说明 1、ev_loop对象的初始化：
ev_loop对象初始化：
1、# define EV_DEFAULT ev_default_loop (0)2、
static struct ev_loop default_loop_struct;
EV_API_DECL struct ev_loop *ev_default_loop_ptr = 0;ev_default_loop (unsigned int flags) EV_THROW
{if (!ev_default_loop_ptr){EV_P = ev_default_loop_ptr = &default_loop_struct;loop_init (EV_A_ flags);}return ev_default_loop_ptr;
}3、
static void noinline ecb_cold
loop_init (EV_P_ unsigned int flags) EV_THROW
{flags = atoi (getenv ("LIBEV_FLAGS"));
#if EV_USE_IOCPif (!backend && (flags & EVBACKEND_IOCP )) backend = iocp_init (EV_A_ flags);
#endif
#if EV_USE_PORTif (!backend && (flags & EVBACKEND_PORT )) backend = port_init (EV_A_ flags);
#endif
#if EV_USE_KQUEUEif (!backend && (flags & EVBACKEND_KQUEUE)) backend = kqueue_init (EV_A_ flags);
#endif
#if EV_USE_EPOLLif (!backend && (flags & EVBACKEND_EPOLL )) backend = epoll_init (EV_A_ flags);
#endif
#if EV_USE_POLLif (!backend && (flags & EVBACKEND_POLL )) backend = poll_init (EV_A_ flags);
#endif
#if EV_USE_SELECTif (!backend && (flags & EVBACKEND_SELECT)) backend = select_init (EV_A_ flags);
}
EV_USE_EPOLL 、EV_USE_SELECT等宏是在调用./configure时，搜索sys/epoll.h sys/select.h等文件，如果文件存在，就将宏设置为1.
__cplusplus宏是g++编译器定义的4、
int inline_size
epoll_init (EV_P_ int flags)
{backend_fd = epoll_create (256);if (backend_fd < 0)return 0;fcntl (backend_fd, F_SETFD, FD_CLOEXEC);backend_mintime = 1e-3; /* epoll does sometimes return early, this is just to avoid the worst */backend_modify = epoll_modify;backend_poll = epoll_poll;epoll_eventmax = 64; /* initial number of events receivable per poll */epoll_events = (struct epoll_event *)ev_malloc (sizeof (struct epoll_event) * epoll_eventmax);return EVBACKEND_EPOLL;//即 EVBACKEND_EPOLL = 0x00000004U
}
epoll_init返回值为非0，所以不会调用后面的poll_init、select_init。
所以，在初始化时，调用了epoll_create初始化backend_fd。
2、fd_reify
fd_reify (EV_P)
{for (i = 0; i < fdchangecnt; ++i){int fd = fdchanges [i];ANFD *anfd = anfds + fd;if (o_reify & EV__IOFDSET)backend_modify (EV_A_ fd, o_events, anfd->events); //即poll_modify}fdchangecnt = 0;
}
epoll_modify (EV_P_ int fd, int oev, int nev)
{struct epoll_event ev;ev.data.u64 = (uint64_t)(uint32_t)fd| ((uint64_t)(uint32_t)++anfds [fd].egen << 32);ev.events = (nev & EV_READ ? EPOLLIN : 0)| (nev & EV_WRITE ? EPOLLOUT : 0);if (expect_true (!epoll_ctl (backend_fd, oev && oldmask != nev ? EPOLL_CTL_MOD : EPOLL_CTL_ADD, fd, &ev)))return;
}
在fd_reify中将andfs中监听的事件添加到backend_fd中。
3、backend_poll
backend_poll (EV_A_ waittime); // 即epoll_poll
epoll_poll (EV_P_ ev_tstamp timeout)
{eventcnt = epoll_wait (backend_fd, epoll_events, epoll_eventmax, timeout * 1e3);for (i = 0; i < eventcnt; ++i){struct epoll_event *ev = epoll_events + i;int fd = (uint32_t)ev->data.u64; /* mask out the lower 32 bits */int want = anfds [fd].events;int got = (ev->events & (EPOLLOUT | EPOLLERR | EPOLLHUP) ? EV_WRITE : 0)| (ev->events & (EPOLLIN | EPOLLERR | EPOLLHUP) ? EV_READ : 0);fd_event (EV_A_ fd, got); //将watcher设置到loop的pending数组中}
}
fd_event (EV_P_ int fd, int revents)
{ANFD *anfd = anfds + fd;if (expect_true (!anfd->reify))fd_event_nocheck (EV_A_ fd, revents);
}
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);}
}
ev_feed_event (EV_P_ void *w, int revents) EV_THROW
{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);pendings [pri][w_->pending - 1].w = w_;pendings [pri][w_->pending - 1].events = revents;}
}
在backend_poll中会调用epoll_wait，通过fd_event函数，将就绪的文件描述符对应的监听器添加到ev_loop对象的pendings字段中
4、EV_INVOKE_PENDING
void noinline
ev_invoke_pending (EV_P)
{pendingpri = NUMPRI;while (pendingpri) /* pendingpri possibly gets modified in the inner loop */{--pendingpri;while (pendingcnt [pendingpri]){ANPENDING *p = pendings [pendingpri] + --pendingcnt [pendingpri];p->w->pending = 0;EV_CB_INVOKE (p->w, p->events);}}
}
# define EV_CB_INVOKE(watcher,revents) (watcher)->cb (EV_A_ (watcher), (revents))
EV_INVOKE_PENDING会一次调用pendings中的监听器的回调函数。
至此，ev_run大体介绍完毕。
小结：
总的来说，对于IO事件驱动，libev是先将监听器存放在一个数组，每次遍历都将监听器监听的文件描述符添加到epoll_wait进行监听，然后将eopll_wait返回的就绪描述符对应的监听器添加到pendings，最后调用pendings中监听器的回调函数。
