分析libevent的源代码，我的想法的是先分析各种结构体，struct event_base、struct event，然后是event_base_new函数、event_new函数、event_add函数，最后分析event_base_dispatch函数。
一、各种结构体
1、event_base
struct event_base {
/** Function pointers and other data to describe this event_base's
* backend. */
const struct eventop *evsel;
/** Pointer to backend-specific data. */
void *evbase;
/** List of changes to tell backend about at next dispatch. Only used
* by the O(1) backends. */
struct event_changelist changelist;
/** Function pointers used to describe the backend that this event_base
* uses for signals */
const struct eventop *evsigsel;
/** Data to implement the common signal handelr code. */
struct evsig_info sig;
/** Number of virtual events */
int virtual_event_count;
/** Maximum number of virtual events active */
int virtual_event_count_max;
/** Number of total events added to this event_base */
int event_count;
/** Maximum number of total events added to this event_base */
int event_count_max;
/** Number of total events active in this event_base */
int event_count_active;
/** Maximum number of total events active in this event_base */
int event_count_active_max;
/** Set if we should terminate the loop once we're done processing
* events. */
int event_gotterm;
/** Set if we should terminate the loop immediately */
int event_break;
/** Set if we should start a new instance of the loop immediately. */
int event_continue;
/** The currently running priority of events */
int event_running_priority;
/** Set if we're running the event_base_loop function, to prevent
* reentrant invocation. */
int running_loop;
/** Set to the number of deferred_cbs we've made 'active' in the
* loop. This is a hack to prevent starvation; it would be smarter
* to just use event_config_set_max_dispatch_interval's max_callbacks
* feature */
int n_deferreds_queued;
/* Active event management. */
/** An array of nactivequeues queues for active event_callbacks (ones
* that have triggered, and whose callbacks need to be called). Low
* priority numbers are more important, and stall higher ones.
*/
struct evcallback_list *activequeues;
/** The length of the activequeues array */
int nactivequeues;
/** A list of event_callbacks that should become active the next time
* we process events, but not this time. */
struct evcallback_list active_later_queue;
/* common timeout logic */
/** An array of common_timeout_list* for all of the common timeout
* values we know. */
struct common_timeout_list **common_timeout_queues;
/** The number of entries used in common_timeout_queues */
int n_common_timeouts;
/** The total size of common_timeout_queues. */
int n_common_timeouts_allocated;
/** Mapping from file descriptors to enabled (added) events */
struct event_io_map io;
/** Mapping from signal numbers to enabled (added) events. */
struct event_signal_map sigmap;
/** Priority queue of events with timeouts. */
struct min_heap timeheap;
/** Stored timeval: used to avoid calling gettimeofday/clock_gettime
* too often. */
struct timeval tv_cache;
struct evutil_monotonic_timer monotonic_timer;
/** Difference between internal time (maybe from clock_gettime) and
* gettimeofday. */
struct timeval tv_clock_diff;
/** Second in which we last updated tv_clock_diff, in monotonic time. */
time_t last_updated_clock_diff;
#ifndef EVENT__DISABLE_THREAD_SUPPORT
/* threading support */
/** The thread currently running the event_loop for this base */
unsigned long th_owner_id;
/** A lock to prevent conflicting accesses to this event_base */
void *th_base_lock;
/** A condition that gets signalled when we're done processing an
* event with waiters on it. */
void *current_event_cond;
/** Number of threads blocking on current_event_cond. */
int current_event_waiters;
#endif
/** The event whose callback is executing right now */
struct event_callback *current_event;
#ifdef _WIN32
/** IOCP support structure, if IOCP is enabled. */
struct event_iocp_port *iocp;
#endif
/** Flags that this base was configured with */
enum event_base_config_flag flags;
struct timeval max_dispatch_time;
int max_dispatch_callbacks;
int limit_callbacks_after_prio;
/* Notify main thread to wake up break, etc. */
/** True if the base already has a pending notify, and we don't need
* to add any more. */
int is_notify_pending;
/** A socketpair used by some th_notify functions to wake up the main
* thread. */
evutil_socket_t th_notify_fd[];
/** An event used by some th_notify functions to wake up the main
* thread. */
struct event th_notify;
/** A function used to wake up the main thread from another thread. */
int (*th_notify_fn)(struct event_base *base);
/** Saved seed for weak random number generator. Some backends use
* this to produce fairness among sockets. Protected by th_base_lock. */
struct evutil_weakrand_state weakrand_seed;
/** List of event_onces that have not yet fired. */
LIST_HEAD(once_event_list, event_once) once_events;
};
struct event_base结构体在event-internal.h文件中定义。
二、初始化函数
1、event_base_new函数
struct event_base *
event_base_new(void)
{
struct event_base *base = NULL;
struct event_config *cfg = event_config_new();
if (cfg) {
base = event_base_new_with_config(cfg);
event_config_free(cfg);
}
return base;
}
(1)调用event_config_new函数分配一个struct event_config结构体。
(2)如果分配成功，就调用event_base_new_with_config(cfg)分配一个struct event_base对象指针，然后将该指针返回。
总结：所以event_base_new还是调用了event_base_new_with_config函数。所以下面接着来看event_base_new_with_config函数。
2、event_base_new_with_config函数
struct event_base *
event_base_new_with_config(const struct event_config *cfg)
{
int i;
struct event_base *base;
int should_check_environment; #ifndef EVENT__DISABLE_DEBUG_MODE
event_debug_mode_too_late = ;
#endif if ((base = mm_calloc(, sizeof(struct event_base))) == NULL) {
event_warn("%s: calloc", __func__);
return NULL;
} if (cfg)
base->flags = cfg->flags; should_check_environment =
!(cfg && (cfg->flags & EVENT_BASE_FLAG_IGNORE_ENV)); {
struct timeval tmp;
int precise_time =
cfg && (cfg->flags & EVENT_BASE_FLAG_PRECISE_TIMER);
int flags;
if (should_check_environment && !precise_time) {
precise_time = evutil_getenv_("EVENT_PRECISE_TIMER") != NULL;
base->flags |= EVENT_BASE_FLAG_PRECISE_TIMER;
}
flags = precise_time ? EV_MONOT_PRECISE : ;
evutil_configure_monotonic_time_(&base->monotonic_timer, flags); gettime(base, &tmp);
} min_heap_ctor_(&base->timeheap); base->sig.ev_signal_pair[] = -;
base->sig.ev_signal_pair[] = -;
base->th_notify_fd[] = -;
base->th_notify_fd[] = -; TAILQ_INIT(&base->active_later_queue); evmap_io_initmap_(&base->io);
evmap_signal_initmap_(&base->sigmap);
event_changelist_init_(&base->changelist); base->evbase = NULL; if (cfg) {
memcpy(&base->max_dispatch_time,
&cfg->max_dispatch_interval, sizeof(struct timeval));
base->limit_callbacks_after_prio =
cfg->limit_callbacks_after_prio;
} else {
base->max_dispatch_time.tv_sec = -;
base->limit_callbacks_after_prio = ;
}
if (cfg && cfg->max_dispatch_callbacks >= ) {
base->max_dispatch_callbacks = cfg->max_dispatch_callbacks;
} else {
base->max_dispatch_callbacks = INT_MAX;
}
if (base->max_dispatch_callbacks == INT_MAX &&
base->max_dispatch_time.tv_sec == -)
base->limit_callbacks_after_prio = INT_MAX; for (i = ; eventops[i] && !base->evbase; i++) {
if (cfg != NULL) {
/* determine if this backend should be avoided */
if (event_config_is_avoided_method(cfg,
eventops[i]->name))
continue;
if ((eventops[i]->features & cfg->require_features)
!= cfg->require_features)
continue;
} /* also obey the environment variables */
if (should_check_environment &&
event_is_method_disabled(eventops[i]->name))
continue; base->evsel = eventops[i]; base->evbase = base->evsel->init(base);
} if (base->evbase == NULL) {
event_warnx("%s: no event mechanism available",
__func__);
base->evsel = NULL;
event_base_free(base);
return NULL;
} if (evutil_getenv_("EVENT_SHOW_METHOD"))
event_msgx("libevent using: %s", base->evsel->name); /* allocate a single active event queue */
if (event_base_priority_init(base, ) < ) {
event_base_free(base);
return NULL;
} /* prepare for threading */ #if !defined(EVENT__DISABLE_THREAD_SUPPORT) && !defined(EVENT__DISABLE_DEBUG_MODE)
event_debug_created_threadable_ctx_ = ;
#endif #ifndef EVENT__DISABLE_THREAD_SUPPORT
if (EVTHREAD_LOCKING_ENABLED() &&
(!cfg || !(cfg->flags & EVENT_BASE_FLAG_NOLOCK))) {
int r;
EVTHREAD_ALLOC_LOCK(base->th_base_lock, );
EVTHREAD_ALLOC_COND(base->current_event_cond);
r = evthread_make_base_notifiable(base);
if (r<) {
event_warnx("%s: Unable to make base notifiable.", __func__);
event_base_free(base);
return NULL;
}
}
#endif #ifdef _WIN32
if (cfg && (cfg->flags & EVENT_BASE_FLAG_STARTUP_IOCP))
event_base_start_iocp_(base, cfg->n_cpus_hint);
#endif return (base);
}
(1)调用mm_calloc函数分配一块大小为sizeof(struct event_base)的内存空间。
(2)如果形参cfg不为NULL，则将base.flags赋值为cfg->flags。
(3)第71-90行设置了实际使用的后端机制，for循环从遍历数组eventops，直到找到一个可用的后端为止，可以看一下eventops。
#ifdef EVENT__HAVE_EVENT_PORTS
extern const struct eventop evportops;
#endif
#ifdef EVENT__HAVE_SELECT
extern const struct eventop selectops;
#endif
#ifdef EVENT__HAVE_POLL
extern const struct eventop pollops;
#endif
#ifdef EVENT__HAVE_EPOLL
extern const struct eventop epollops;
#endif
#ifdef EVENT__HAVE_WORKING_KQUEUE
extern const struct eventop kqops;
#endif
#ifdef EVENT__HAVE_DEVPOLL
extern const struct eventop devpollops;
#endif
#ifdef _WIN32
extern const struct eventop win32ops;
#endif /* Array of backends in order of preference. */
static const struct eventop *eventops[] = {
#ifdef EVENT__HAVE_EVENT_PORTS
&evportops,
#endif
#ifdef EVENT__HAVE_WORKING_KQUEUE
&kqops,
#endif
#ifdef EVENT__HAVE_EPOLL
&epollops,
#endif
#ifdef EVENT__HAVE_DEVPOLL
&devpollops,
#endif
#ifdef EVENT__HAVE_POLL
&pollops,
#endif
#ifdef EVENT__HAVE_SELECT
&selectops,
#endif
#ifdef _WIN32
&win32ops,
#endif
NULL
};
从代码中可以看到，根据宏定义来决定某些后端机制是否存在，这样就可以找到运行机子上支持的一个可用的后端机制，而且需要注意，epool、pool、select的顺序，所以如果支持epoll就不会选择poll，如果支持poll就不会选择select，select机制是最后的选择。
/** Structure to define the backend of a given event_base. */
struct eventop {
/** The name of this backend. */
const char *name;
/** Function to set up an event_base to use this backend. It should
* create a new structure holding whatever information is needed to
* run the backend, and return it. The returned pointer will get
* stored by event_init into the event_base.evbase field. On failure,
* this function should return NULL. */
void *(*init)(struct event_base *);
/** Enable reading/writing on a given fd or signal. 'events' will be
* the events that we're trying to enable: one or more of EV_READ,
* EV_WRITE, EV_SIGNAL, and EV_ET. 'old' will be those events that
* were enabled on this fd previously. 'fdinfo' will be a structure
* associated with the fd by the evmap; its size is defined by the
* fdinfo field below. It will be set to 0 the first time the fd is
* added. The function should return 0 on success and -1 on error.
*/
int (*add)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
/** As "add", except 'events' contains the events we mean to disable. */
int (*del)(struct event_base *, evutil_socket_t fd, short old, short events, void *fdinfo);
/** Function to implement the core of an event loop. It must see which
added events are ready, and cause event_active to be called for each
active event (usually via event_io_active or such). It should
return 0 on success and -1 on error.
*/
int (*dispatch)(struct event_base *, struct timeval *);
/** Function to clean up and free our data from the event_base. */
void (*dealloc)(struct event_base *);
/** Flag: set if we need to reinitialize the event base after we fork.
*/
int need_reinit;
/** Bit-array of supported event_method_features that this backend can
* provide. */
enum event_method_feature features;
/** Length of the extra information we should record for each fd that
has one or more active events. This information is recorded
as part of the evmap entry for each fd, and passed as an argument
to the add and del functions above.
*/
size_t fdinfo_len;
};
struct eventop结构定义了后端机制的一个公共接口，至于每个后端是如何将自己的函数封装成符合这个接口的，我下面会逐个分析。
(4)然后调用init函数来初始化event_base对象。init函数的具体实现根据不同的后端机制会有所不同。
3、event_new函数
struct event *
event_new(struct event_base *base, evutil_socket_t fd, short events, void (*cb)(evutil_socket_t, short, void *), void *arg)
{
struct event *ev;
ev = mm_malloc(sizeof(struct event));
if (ev == NULL)
return (NULL);
if (event_assign(ev, base, fd, events, cb, arg) < ) {
mm_free(ev);
return (NULL);
} return (ev);
}
(1)调用mm_malloc函数分配一块大小为sizeof(struct event)的内存空间。
(2)event_new的实现类似于event_base_new函数类似，分配好空间之后，调用了event_assign函数来填充结构体。
4、event_assign函数
int
event_assign(struct event *ev, struct event_base *base, evutil_socket_t fd, short events, void (*callback)(evutil_socket_t, short, void *), void *arg)
{
if (!base)
base = current_base;
if (arg == &event_self_cbarg_ptr_)
arg = ev; event_debug_assert_not_added_(ev); ev->ev_base = base; ev->ev_callback = callback;
ev->ev_arg = arg;
ev->ev_fd = fd;
ev->ev_events = events;
ev->ev_res = ;
ev->ev_flags = EVLIST_INIT;
ev->ev_ncalls = ;
ev->ev_pncalls = NULL; if (events & EV_SIGNAL) {
if ((events & (EV_READ|EV_WRITE|EV_CLOSED)) != ) {
event_warnx("%s: EV_SIGNAL is not compatible with "
"EV_READ, EV_WRITE or EV_CLOSED", __func__);
return -;
}
ev->ev_closure = EV_CLOSURE_EVENT_SIGNAL;
} else {
if (events & EV_PERSIST) {
evutil_timerclear(&ev->ev_io_timeout);
ev->ev_closure = EV_CLOSURE_EVENT_PERSIST;
} else {
ev->ev_closure = EV_CLOSURE_EVENT;
}
} min_heap_elem_init_(ev); if (base != NULL) {
/* by default, we put new events into the middle priority */
ev->ev_pri = base->nactivequeues / ;
} event_debug_note_setup_(ev); return ;
}
(1)event_assign函数的主要操作是给形参struct event *ev的成员赋值，包括ev->ev_base、ev->ev_callback、ev->ev_arg、ev->ev_fd、ev->ev_events等
总结：event_new、event_assign函数会把传递进来的struct event_base* base保存在获取到的strut event结构体内部。
5、event_add函数
int
event_add(struct event *ev, const struct timeval *tv)
{
int res; if (EVUTIL_FAILURE_CHECK(!ev->ev_base)) {
event_warnx("%s: event has no event_base set.", __func__);
return -;
} EVBASE_ACQUIRE_LOCK(ev->ev_base, th_base_lock); res = event_add_nolock_(ev, tv, ); EVBASE_RELEASE_LOCK(ev->ev_base, th_base_lock); return (res);
}
(1)event_add函数调用了event_add_nolock_函数进行实际的操作。
6、event_add_nolock_函数
/* Implementation function to add an event. Works just like event_add,
* except: 1) it requires that we have the lock. 2) if tv_is_absolute is set,
* we treat tv as an absolute time, not as an interval to add to the current
* time */
int
event_add_nolock_(struct event *ev, const struct timeval *tv,
int tv_is_absolute)
{
struct event_base *base = ev->ev_base;
int res = ;
int notify = ; EVENT_BASE_ASSERT_LOCKED(base);
event_debug_assert_is_setup_(ev); event_debug((
"event_add: event: %p (fd "EV_SOCK_FMT"), %s%s%s%scall %p",
ev,
EV_SOCK_ARG(ev->ev_fd),
ev->ev_events & EV_READ ? "EV_READ " : " ",
ev->ev_events & EV_WRITE ? "EV_WRITE " : " ",
ev->ev_events & EV_CLOSED ? "EV_CLOSED " : " ",
tv ? "EV_TIMEOUT " : " ",
ev->ev_callback)); EVUTIL_ASSERT(!(ev->ev_flags & ~EVLIST_ALL)); if (ev->ev_flags & EVLIST_FINALIZING) {
/* XXXX debug */
return (-);
} /*
* prepare for timeout insertion further below, if we get a
* failure on any step, we should not change any state.
*/
if (tv != NULL && !(ev->ev_flags & EVLIST_TIMEOUT)) {
if (min_heap_reserve_(&base->timeheap,
+ min_heap_size_(&base->timeheap)) == -)
return (-); /* ENOMEM == errno */
} /* If the main thread is currently executing a signal event's
* callback, and we are not the main thread, then we want to wait
* until the callback is done before we mess with the event, or else
* we can race on ev_ncalls and ev_pncalls below. */
#ifndef EVENT__DISABLE_THREAD_SUPPORT
if (base->current_event == event_to_event_callback(ev) &&
(ev->ev_events & EV_SIGNAL)
&& !EVBASE_IN_THREAD(base)) {
++base->current_event_waiters;
EVTHREAD_COND_WAIT(base->current_event_cond, base->th_base_lock);
}
#endif if ((ev->ev_events & (EV_READ|EV_WRITE|EV_CLOSED|EV_SIGNAL)) &&
!(ev->ev_flags & (EVLIST_INSERTED|EVLIST_ACTIVE|EVLIST_ACTIVE_LATER))) {
if (ev->ev_events & (EV_READ|EV_WRITE|EV_CLOSED))
res = evmap_io_add_(base, ev->ev_fd, ev);
else if (ev->ev_events & EV_SIGNAL)
res = evmap_signal_add_(base, (int)ev->ev_fd, ev);
if (res != -)
event_queue_insert_inserted(base, ev);
if (res == ) {
/* evmap says we need to notify the main thread. */
notify = ;
res = ;
}
} /*
* we should change the timeout state only if the previous event
* addition succeeded.
*/
if (res != - && tv != NULL) {
struct timeval now;
int common_timeout;
#ifdef USE_REINSERT_TIMEOUT
int was_common;
int old_timeout_idx;
#endif /*
* for persistent timeout events, we remember the
* timeout value and re-add the event.
*
* If tv_is_absolute, this was already set.
*/
if (ev->ev_closure == EV_CLOSURE_EVENT_PERSIST && !tv_is_absolute)
ev->ev_io_timeout = *tv; #ifndef USE_REINSERT_TIMEOUT
if (ev->ev_flags & EVLIST_TIMEOUT) {
event_queue_remove_timeout(base, ev);
}
#endif /* Check if it is active due to a timeout. Rescheduling
* this timeout before the callback can be executed
* removes it from the active list. */
if ((ev->ev_flags & EVLIST_ACTIVE) &&
(ev->ev_res & EV_TIMEOUT)) {
if (ev->ev_events & EV_SIGNAL) {
/* See if we are just active executing
* this event in a loop
*/
if (ev->ev_ncalls && ev->ev_pncalls) {
/* Abort loop */
*ev->ev_pncalls = ;
}
} event_queue_remove_active(base, event_to_event_callback(ev));
} gettime(base, &now); common_timeout = is_common_timeout(tv, base);
#ifdef USE_REINSERT_TIMEOUT
was_common = is_common_timeout(&ev->ev_timeout, base);
old_timeout_idx = COMMON_TIMEOUT_IDX(&ev->ev_timeout);
#endif if (tv_is_absolute) {
ev->ev_timeout = *tv;
} else if (common_timeout) {
struct timeval tmp = *tv;
tmp.tv_usec &= MICROSECONDS_MASK;
evutil_timeradd(&now, &tmp, &ev->ev_timeout);
ev->ev_timeout.tv_usec |=
(tv->tv_usec & ~MICROSECONDS_MASK);
} else {
evutil_timeradd(&now, tv, &ev->ev_timeout);
} event_debug((
"event_add: event %p, timeout in %d seconds %d useconds, call %p",
ev, (int)tv->tv_sec, (int)tv->tv_usec, ev->ev_callback)); #ifdef USE_REINSERT_TIMEOUT
event_queue_reinsert_timeout(base, ev, was_common, common_timeout, old_timeout_idx);
#else
event_queue_insert_timeout(base, ev);
#endif if (common_timeout) {
struct common_timeout_list *ctl =
get_common_timeout_list(base, &ev->ev_timeout);
if (ev == TAILQ_FIRST(&ctl->events)) {
common_timeout_schedule(ctl, &now, ev);
}
} else {
struct event* top = NULL;
/* See if the earliest timeout is now earlier than it
* was before: if so, we will need to tell the main
* thread to wake up earlier than it would otherwise.
* We double check the timeout of the top element to
* handle time distortions due to system suspension.
*/
if (min_heap_elt_is_top_(ev))
notify = ;
else if ((top = min_heap_top_(&base->timeheap)) != NULL &&
evutil_timercmp(&top->ev_timeout, &now, <))
notify = ;
}
} /* if we are not in the right thread, we need to wake up the loop */
if (res != - && notify && EVBASE_NEED_NOTIFY(base))
evthread_notify_base(base); event_debug_note_add_(ev); return (res);
}
（1）
三、event_base_dispatch函数
1、event_base_dispatch函数
int
event_base_dispatch(struct event_base *event_base)
{
return (event_base_loop(event_base, ));
}
(1)可以看到，event_base_dispatch函数间接调用了 event_base_loop函数
2、event_base_loop函数
int
event_base_loop(struct event_base *base, int flags)
{
const struct eventop *evsel = base->evsel;
struct timeval tv;
struct timeval *tv_p;
int res, done, retval = ; /* Grab the lock. We will release it inside evsel.dispatch, and again
* as we invoke user callbacks. */
EVBASE_ACQUIRE_LOCK(base, th_base_lock); if (base->running_loop) {
event_warnx("%s: reentrant invocation. Only one event_base_loop"
" can run on each event_base at once.", __func__);
EVBASE_RELEASE_LOCK(base, th_base_lock);
return -;
} base->running_loop = ; clear_time_cache(base); if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)
evsig_set_base_(base); done = ; #ifndef EVENT__DISABLE_THREAD_SUPPORT
base->th_owner_id = EVTHREAD_GET_ID();
#endif base->event_gotterm = base->event_break = ; while (!done) {
base->event_continue = ;
base->n_deferreds_queued = ; /* Terminate the loop if we have been asked to */
if (base->event_gotterm) {
break;
} if (base->event_break) {
break;
} tv_p = &tv;
if (!N_ACTIVE_CALLBACKS(base) && !(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 (==(flags&EVLOOP_NO_EXIT_ON_EMPTY) &&
!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
event_debug(("%s: no events registered.", __func__));
retval = ;
goto done;
} event_queue_make_later_events_active(base); clear_time_cache(base); res = evsel->dispatch(base, tv_p); if (res == -) {
event_debug(("%s: dispatch returned unsuccessfully.",
__func__));
retval = -;
goto done;
} update_time_cache(base); timeout_process(base); if (N_ACTIVE_CALLBACKS(base)) {
int n = event_process_active(base);
if ((flags & EVLOOP_ONCE)
&& N_ACTIVE_CALLBACKS(base) ==
&& n != )
done = ;
} else if (flags & EVLOOP_NONBLOCK)
done = ;
}
event_debug(("%s: asked to terminate loop.", __func__)); done:
clear_time_cache(base);
base->running_loop = ; EVBASE_RELEASE_LOCK(base, th_base_lock); return (retval);
}
(1)event_base_loop函数的主要逻辑是就一个死循环，在循环中不断的调用由不同多路分发机制提供的后端接口。71行。
(2)调用后端接口返回后，调用event_process_active函数处理激活的事件。
3、event_process_active函数
/*
* Active events are stored in priority queues. Lower priorities are always
* process before higher priorities. Low priority events can starve high
* priority ones.
*/ static int
event_process_active(struct event_base *base)
{
/* Caller must hold th_base_lock */
struct evcallback_list *activeq = NULL;
int i, c = ;
const struct timeval *endtime;
struct timeval tv;
const int maxcb = base->max_dispatch_callbacks;
const int limit_after_prio = base->limit_callbacks_after_prio;
if (base->max_dispatch_time.tv_sec >= ) {
update_time_cache(base);
gettime(base, &tv);
evutil_timeradd(&base->max_dispatch_time, &tv, &tv);
endtime = &tv;
} else {
endtime = NULL;
} for (i = ; i < base->nactivequeues; ++i) {
if (TAILQ_FIRST(&base->activequeues[i]) != NULL) {
base->event_running_priority = i;
activeq = &base->activequeues[i];
if (i < limit_after_prio)
c = event_process_active_single_queue(base, activeq,
INT_MAX, NULL);
else
c = event_process_active_single_queue(base, activeq,
maxcb, endtime);
if (c < ) {
goto done;
} else if (c > )
break; /* Processed a real event; do not
* consider lower-priority events */
/* If we get here, all of the events we processed
* were internal. Continue. */
}
} done:
base->event_running_priority = -; return c;
}
(1)第26-44行，循环遍历激活的事件，然后调用event_process_active_single_queue函数。
4、event_process_active_single_queue函数
/*
Helper for event_process_active to process all the events in a single queue,
releasing the lock as we go. This function requires that the lock be held
when it's invoked. Returns -1 if we get a signal or an event_break that
means we should stop processing any active events now. Otherwise returns
the number of non-internal event_callbacks that we processed.
*/
static int
event_process_active_single_queue(struct event_base *base,
struct evcallback_list *activeq,
int max_to_process, const struct timeval *endtime)
{
struct event_callback *evcb;
int count = ; EVUTIL_ASSERT(activeq != NULL); for (evcb = TAILQ_FIRST(activeq); evcb; evcb = TAILQ_FIRST(activeq)) {
struct event *ev=NULL;
if (evcb->evcb_flags & EVLIST_INIT) {
ev = event_callback_to_event(evcb); if (ev->ev_events & EV_PERSIST || ev->ev_flags & EVLIST_FINALIZING)
event_queue_remove_active(base, evcb);
else
event_del_nolock_(ev, EVENT_DEL_NOBLOCK);
event_debug((
"event_process_active: event: %p, %s%s%scall %p",
ev,
ev->ev_res & EV_READ ? "EV_READ " : " ",
ev->ev_res & EV_WRITE ? "EV_WRITE " : " ",
ev->ev_res & EV_CLOSED ? "EV_CLOSED " : " ",
ev->ev_callback));
} else {
event_queue_remove_active(base, evcb);
event_debug(("event_process_active: event_callback %p, "
"closure %d, call %p",
evcb, evcb->evcb_closure, evcb->evcb_cb_union.evcb_callback));
} if (!(evcb->evcb_flags & EVLIST_INTERNAL))
++count; base->current_event = evcb;
#ifndef EVENT__DISABLE_THREAD_SUPPORT
base->current_event_waiters = ;
#endif switch (evcb->evcb_closure) {
case EV_CLOSURE_EVENT_SIGNAL:
EVUTIL_ASSERT(ev != NULL);
event_signal_closure(base, ev);
break;
case EV_CLOSURE_EVENT_PERSIST:
EVUTIL_ASSERT(ev != NULL);
event_persist_closure(base, ev);
break;
case EV_CLOSURE_EVENT: {
void (*evcb_callback)(evutil_socket_t, short, void *);
EVUTIL_ASSERT(ev != NULL);
evcb_callback = *ev->ev_callback;
EVBASE_RELEASE_LOCK(base, th_base_lock);
evcb_callback(ev->ev_fd, ev->ev_res, ev->ev_arg);
}
break;
case EV_CLOSURE_CB_SELF: {
void (*evcb_selfcb)(struct event_callback *, void *) = evcb->evcb_cb_union.evcb_selfcb;
EVBASE_RELEASE_LOCK(base, th_base_lock);
evcb_selfcb(evcb, evcb->evcb_arg);
}
break;
case EV_CLOSURE_EVENT_FINALIZE:
case EV_CLOSURE_EVENT_FINALIZE_FREE: {
void (*evcb_evfinalize)(struct event *, void *);
int evcb_closure = evcb->evcb_closure;
EVUTIL_ASSERT(ev != NULL);
base->current_event = NULL;
evcb_evfinalize = ev->ev_evcallback.evcb_cb_union.evcb_evfinalize;
EVUTIL_ASSERT((evcb->evcb_flags & EVLIST_FINALIZING));
EVBASE_RELEASE_LOCK(base, th_base_lock);
evcb_evfinalize(ev, ev->ev_arg);
event_debug_note_teardown_(ev);
if (evcb_closure == EV_CLOSURE_EVENT_FINALIZE_FREE)
mm_free(ev);
}
break;
case EV_CLOSURE_CB_FINALIZE: {
void (*evcb_cbfinalize)(struct event_callback *, void *) = evcb->evcb_cb_union.evcb_cbfinalize;
base->current_event = NULL;
EVUTIL_ASSERT((evcb->evcb_flags & EVLIST_FINALIZING));
EVBASE_RELEASE_LOCK(base, th_base_lock);
evcb_cbfinalize(evcb, evcb->evcb_arg);
}
break;
default:
EVUTIL_ASSERT();
} EVBASE_ACQUIRE_LOCK(base, th_base_lock);
base->current_event = NULL;
#ifndef EVENT__DISABLE_THREAD_SUPPORT
if (base->current_event_waiters) {
base->current_event_waiters = ;
EVTHREAD_COND_BROADCAST(base->current_event_cond);
}
#endif if (base->event_break)
return -;
if (count >= max_to_process)
return count;
if (count && endtime) {
struct timeval now;
update_time_cache(base);
gettime(base, &now);
if (evutil_timercmp(&now, endtime, >=))
return count;
}
if (base->event_continue)
break;
}
return count;
}
（1）