/*
* 我们先来看一下事件的创建
*/
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);
//调用assign 给ev 赋值
if (event_assign(ev, base, fd, events, cb, arg) < 0) {
mm_free(ev);
return (NULL);
}
return (ev);
}
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->ev_base = base;
ev->ev_callback = callback;
ev->ev_arg = arg;
ev->ev_fd = fd;
ev->ev_events = events;
ev->ev_res = 0;
ev->ev_flags = EVLIST_INIT;
ev->ev_ncalls = 0;
ev->ev_pncalls = NULL;
if (events & EV_SIGNAL) {
//信号事件和io事件冲突
if ((events & (EV_READ|EV_WRITE|EV_CLOSED)) != 0) {
event_warnx("%s: EV_SIGNAL is not compatible with "
"EV_READ, EV_WRITE or EV_CLOSED", __func__);
return -1;
}
ev->ev_closure = EV_CLOSURE_EVENT_SIGNAL;
} else {
if (events & EV_PERSIST) {
//如果事件是persist 清理记0
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);// 初始化最小堆索引为 -1
//void min_heap_elem_init_(struct event* e) { e->ev_timeout_pos.min_heap_idx = -1; }
if (base != NULL) {
/* by default, we put new events into the middle priority */
ev->ev_pri = base->nactivequeues / 2;
}
event_debug_note_setup_(ev);
return 0;
}
libevent事件管理结构
IO 事件管理
反应堆中event_base 中
{
........
struct event_io_map io;
struct event_signal_map sigmap;
........
}
________________
event_base->io event_io_map ------> | evmap_io |--->| io |----->| io |---||------ event_dlist events.
———————————————— 此文件描述符关心的所有事件链表
| |
————————————————
| |
-----------------
struct evmap_io {
struct event_dlist events; //链表管理一个文件描述符所有关心的io事件
ev_uint16_t nread;
ev_uint16_t nwrite;
ev_uint16_t nclose;
};
struct evmap_signal {
struct event_dlist events;//链表管理所有的信号事件
};
Linux 下的事件的hashmap io与signal相同。
#define event_io_map event_signal_map
struct event_signal_map {
/* An array of evmap_io * or of evmap_signal *; empty entries are
* set to NULL. */
void **entries; //entries 管理的是evmap_io evmap_signal
/* The number of entries available in entries */
int nentries;
};
下面是添加一个io事件的过程：
int evmap_io_add_(struct event_base *base, evutil_socket_t fd, struct event *ev)
{
const struct eventop *evsel = base->evsel;
struct event_io_map *io = &base->io;
struct evmap_io *ctx = NULL;
int nread, nwrite, nclose, retval = 0;
short res = 0, old = 0;
struct event *old_ev;
EVUTIL_ASSERT(fd == ev->ev_fd);
if (fd < 0)
return 0;
// Linux use it
#ifndef EVMAP_USE_HT
if (fd >= io->nentries) {
//分配空间
if (evmap_make_space(io, fd, sizeof(struct evmap_io *)) == -1)
return (-1);
}
#endif
//初始化一个evmap_io 并将ctx指针 指向这个结构体 linux下文件描述符即为key。
GET_IO_SLOT_AND_CTOR(ctx, io, fd, evmap_io, evmap_io_init,
evsel->fdinfo_len);
/*
#define GET_SIGNAL_SLOT_AND_CTOR(x, map, slot, type, ctor, fdinfo_len)
do {
if ((map)->entries[slot] == NULL) {
(map)->entries[slot] =
mm_calloc(1,sizeof(struct type)+fdinfo_len);
if (EVUTIL_UNLIKELY((map)->entries[slot] == NULL))
return (-1);
(ctor)((struct type *)(map)->entries[slot]);
}
(x) = (struct type *)((map)->entries[slot]);
} while (0)
*/
//一个event 可能对应多个事件
// 这里就是上一个的事件类型
nread = ctx->nread;
nwrite = ctx->nwrite;
nclose = ctx->nclose;
if (nread)
old |= EV_READ;
if (nwrite)
old |= EV_WRITE;
if (nclose)
old |= EV_CLOSED;
//新的事件加入 若之前存在就不再设置res 但是会将各个事件个数加一
if (ev->ev_events & EV_READ) {
if (++nread == 1)
res |= EV_READ;
}
if (ev->ev_events & EV_WRITE) {
if (++nwrite == 1)
res |= EV_WRITE;
}
if (ev->ev_events & EV_CLOSED) {
if (++nclose == 1)
res |= EV_CLOSED;
}
//一个描述符添加的事件个数不能超过oxffff.
if (EVUTIL_UNLIKELY(nread > 0xffff || nwrite > 0xffff || nclose > 0xffff)) {
event_warnx("Too many events reading or writing on fd %d",
(int)fd);
return -1;
}
ctx->nread = (ev_uint16_t) nread;
ctx->nwrite = (ev_uint16_t) nwrite;
ctx->nclose = (ev_uint16_t) nclose;
//将新事件加入链表中
LIST_INSERT_HEAD(&ctx->events, ev, ev_io_next);
return (retval);
}
定时则是由min_heap最小堆的数据结构管理，每次取时间最短的事件。
添加事件：
int
event_add_nolock_(struct event *ev, const struct timeval *tv,
int tv_is_absolute)
{
struct event_base *base = ev->ev_base;
int res = 0;
int notify = 0;
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 (-1);
}
/*
* 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,
1 + min_heap_size_(&base->timeheap)) == -1)
return (-1); /* 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))) {
//如果事件是未被加入的 那么就先加入map中 上面所述
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 != -1)
event_queue_insert_inserted(base, ev); //此处并没有将event加入到queue 只是将事件中的flag指为EVLIST_INSERTED (base)->event_count += 1
if (res == 1) {
/* evmap says we need to notify the main thread. */
notify = 1;
res = 0;
}
}
/*
* we should change the timeout state only if the previous event
* addition succeeded.
*/
//添加定时器事件
if (res != -1 && 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. */
//如果此事件已经被激活并且返回了timeout事件就将事件终止， 并删除被激活的定时器事件 进而重新调度定时器
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 = 0;
}
}
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 = 1;
else if ((top = min_heap_top_(&base->timeheap)) != NULL &&
evutil_timercmp(&top->ev_timeout, &now, <))
notify = 1;
}
}
/* if we are not in the right thread, we need to wake up the loop */
//若为多线程 则需要唤醒主线程进行防止主线程仍阻塞在select/poll/epoll_wait. pipe/eventfd
if (res != -1 && notify && EVBASE_NEED_NOTIFY(base))
evthread_notify_base(base);
event_debug_note_add_(ev);
return (res);
}