首页 > 代码库 > Libevent源码分析—event_base_dispatch()
Libevent源码分析—event_base_dispatch()
我们知道libevent是一个Reactor模式的事件驱动的网络库。
到目前为止，我们已经看了核心的event和event_base结构体的源码，看了初始化这两个结构体的源码，看了注册event的源码，也将event注册到I/O多路复用监听的事件上了。现在准备工作都做好了，下面就是看运行时的主循环了，在这个主循环中，是如何检测事件、分发事件、调用事件的回调函数的。这一步就是libevent的核心框架流程了。
Reactor模式中的Event、Event Handler、Reactor目前都完成了，下面就剩Event Demultiplexer了。
这一步通过event_base_dispatch()完成
intevent_base_dispatch(struct event_base *event_base){ return (event_base_loop(event_base, 0)); //调用event_base_loop()}
可以看到，该函数只是做了调用event_base_loop()这一个动作，所以工作实际是在函数event_base_loop()内完成的。
event_base_loop()
该函数完成以下工作：
1.信号标记被设置，则调用信号的回调函数
2.根据定时器最小时间，设置I/O多路复用的最大等待时间，这样即使没有I/O事件发生，也能在最小定时器超时时返回。
3.调用I/O多路复用，监听事件，将活跃事件添加到活跃事件链表中
4.检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中
5.对event_base的活跃事件链表中的事件，调用event_process_active(）函数，在该函数内调用event的回调函数，优先级高的event先处理。
该函数内部调用了eventop.dispatch()监听事件，event_sigcb函数指针处理信号事件，timeout_process()将超时的定时事件加入到活跃事件链表中，event_process_active()处理活跃事件链表中的事件，调用相应的回调函数。
intevent_base_loop(struct event_base *base, int flags){ const struct eventop *evsel = base->evsel; void *evbase = base->evbase; //event_base的I/O多路复用 struct timeval tv; struct timeval *tv_p; int res, done; /* clear time cache */ //清空时间缓存 base->tv_cache.tv_sec = 0; //处理Signal事件时，指定信号所属的event_base if (base->sig.ev_signal_added) evsignal_base = base; done = 0; while (!done) { //进入事件主循环 /* Terminate the loop if we have been asked to */ //设置event_base的标记，以表明是否需要跳出循环 if (base->event_gotterm) { //event_loopexit_cb()可设置 base->event_gotterm = 0; break; } if (base->event_break) { //event_base_loopbreak()可设置 base->event_break = 0; break; } /* You cannot use this interface for multi-threaded apps */ //当event_gotsig被设置时，则event_sigcb就是信号处理的回调函数 while (event_gotsig) { event_gotsig = 0; if (event_sigcb) { res = (*event_sigcb)(); //调用信号处理的回调函数 if (res == -1) { errno = EINTR; return (-1); } } } timeout_correct(base, &tv); //校准时间 tv_p = &tv; //根据定时器堆中最小超时时间计算I/O多路复用的最大等待时间tv_p if (!base->event_count_active && !(flags & EVLOOP_NONBLOCK)) { timeout_next(base, &tv_p); } else { /* * if we have active events, we just poll new events * without waiting. */ evutil_timerclear(&tv); } /* If we have no events, we just exit */ //没有注册事件，则退出 if (!event_haveevents(base)) { event_debug(("%s: no events registered.", __func__)); return (1); } /* update last old time */ gettime(base, &base->event_tv); /* clear time cache */ base->tv_cache.tv_sec = 0; //调用I/O多路复用，监听事件 res = evsel->dispatch(base, evbase, tv_p); if (res == -1) return (-1); //将time cache赋值为当前系统时间 gettime(base, &base->tv_cache); //检查定时事件，将就绪的定时事件从小根堆中删除，插入到活跃事件链表中 timeout_process(base); if (base->event_count_active) { //处理event_base的活跃链表中的事件 //调用event的回调函数，优先级高的event先处理 event_process_active(base); if (!base->event_count_active && (flags & EVLOOP_ONCE)) done = 1; } else if (flags & EVLOOP_NONBLOCK) done = 1; } /* clear time cache */ //循环结束，清空时间缓存 base->tv_cache.tv_sec = 0; event_debug(("%s: asked to terminate loop.", __func__)); return (0);}
epoll_dispatch()
在上面我们看到，event_base_loop()中通过I/O多路复用的dispatch()函数完成监听事件功能。在之前的event_init()中我们看到，通过遍历eventops数组，从中选择一个I/O多路复用机制，所以不同的I/O多路复用机制有不同的eventop结构体，相应的也就有不同的dispatch()函数。下面，再次看下eventop结构体(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;};
在event_add()中通过add()成员函数注册event到监听事件中，现在在event_base_loop()中通过dispatch()成员函数监听事件。
libevent支持多种I/O多路复用机制，下面先看下epoll的eventop结构体(epoll.c)
const struct eventop epollops = { "epoll", epoll_init, epoll_add, epoll_del, epoll_dispatch, epoll_dealloc, 1 /* need reinit */};
然后看下epoll的dispatch()函数(epoll.c)
从下面源码可见，epoll_dispatch()的工作主要有：
1.调用epoll_wait()监听事件
2.如果有信号发生，调用evsignal_process()处理信号
3.将活跃的event根据其活跃的类型注册到活跃事件链表上
4.如果events数组大小不够，则重新分配为原来2倍大小
static intepoll_dispatch(struct event_base *base, void *arg, struct timeval *tv){ struct epollop *epollop = arg; struct epoll_event *events = epollop->events; struct evepoll *evep; int i, res, timeout = -1; if (tv != NULL) timeout = tv->tv_sec * 1000 + (tv->tv_usec + 999) / 1000; //转换为微米 if (timeout > MAX_EPOLL_TIMEOUT_MSEC) { //设置最大超时时间 /* Linux kernels can wait forever if the timeout is too big; * see comment on MAX_EPOLL_TIMEOUT_MSEC. */ timeout = MAX_EPOLL_TIMEOUT_MSEC; } 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事件发生中断，处理Signal事件 return (0); } else if (base->sig.evsignal_caught) { evsignal_process(base); //有Signal事件发生，处理Signal事件 } event_debug(("%s: epoll_wait reports %d", __func__, res)); for (i = 0; i < res; i++) { //处理活跃事件 int what = events[i].events; //活跃类型 struct event *evread = NULL, *evwrite = NULL; int fd = events[i].data.fd; //event的文件描述符 if (fd < 0 || fd >= epollop->nfds) continue; evep = &epollop->fds[fd]; if (what & (EPOLLHUP|EPOLLERR)) { //判断epoll的events类型，并找到注册的event evread = evep->evread; evwrite = evep->evwrite; } else { if (what & EPOLLIN) { evread = evep->evread; } if (what & EPOLLOUT) { evwrite = evep->evwrite; } } if (!(evread||evwrite)) continue; //添加event到活跃事件链表中 if (evread != NULL) event_active(evread, EV_READ, 1); if (evwrite != NULL) event_active(evwrite, EV_WRITE, 1); } //如果注册的事件全部变为活跃，则增大events数组为原来两倍 if (res == epollop->nevents && epollop->nevents < MAX_NEVENTS) { /* We used all of the event space this time. We should be ready for more events next time. */ int new_nevents = epollop->nevents * 2; struct epoll_event *new_events; new_events = realloc(epollop->events, new_nevents * sizeof(struct epoll_event)); if (new_events) { epollop->events = new_events; epollop->nevents = new_nevents; } } return (0);}
event_process_active()
好了，现在活跃的I/O事件、定时器事件已经全部添加到活跃事件链表中了。下面就开始调用这些event的回调函数进行处理了，这步是在event_base_loop()中调用event_process_active()来完成的。
该函数从event_base的activequeueus链表数组上取出一个链表；对该链表上的event调用回调函数；优先调用优先级值最小的event
/* * Active events are stored in priority queues. Lower priorities are always * process before higher priorities. Low priority events can starve high * priority ones. */static voidevent_process_active(struct event_base *base){ struct event *ev; struct event_list *activeq = NULL; int i; short ncalls; for (i = 0; i < base->nactivequeues; ++i) { //取出第一个活跃链表 if (TAILQ_FIRST(base->activequeues[i]) != NULL) { activeq = base->activequeues[i]; break; } } assert(activeq != NULL); //优先处理优先级值最小的event for (ev = TAILQ_FIRST(activeq); ev; ev = TAILQ_FIRST(activeq)) { if (ev->ev_events & EV_PERSIST) event_queue_remove(base, ev, EVLIST_ACTIVE); //是持久事件，则从活跃链表移除 else event_del(ev); //不是持久事件，则直接删除该事件 /* Allows deletes to work */ ncalls = ev->ev_ncalls; ev->ev_pncalls = &ncalls; while (ncalls) { ncalls--; ev->ev_ncalls = ncalls; //调用该event的回调函数，event.ev_res保存返回值 (*ev->ev_callback)((int)ev->ev_fd, ev->ev_res, ev->ev_arg); if (event_gotsig || base->event_break) { ev->ev_pncalls = NULL; return; } } ev->ev_pncalls = NULL; }}
Libevent源码分析—event_base_dispatch()
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。