Libevent 事件循环(1)

// 事件的dispatch 
int
event_base_loop(struct event_base *base, int flags)
{
    //得到采用的事件模型 epoll/epoll/select
    const struct eventop *evsel = base->evsel;
    struct timeval tv;
    struct timeval *tv_p;
    int res, done, retval = 0;

    /* Grab the lock.  We will release it inside evsel.dispatch, and again
     * as we invoke user callbacks. */
    EVBASE_ACQUIRE_LOCK(base, th_base_lock);
    //判断是否loop正在running 如果由则退出
    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 -1;
    }

    base->running_loop = 1;
    //清理时间缓存
    clear_time_cache(base);

    if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)
        evsig_set_base_(base);

    done = 0;

#ifndef EVENT__DISABLE_THREAD_SUPPORT
    base->th_owner_id = EVTHREAD_GET_ID();
#endif
    //终止和中断标志至0
    base->event_gotterm = base->event_break = 0;
    //事件循环 与 平时我们自己写的epoll_wait select 等待事件一样 在一个死循环中
    while (!done) {
        base->event_continue = 0;
        base->n_deferreds_queued = 0;

        /* 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);//判断小根堆中的root是否已经超时， 如果超时 就将tv清0.  如果没有的话就将root的时间减去现在时间的结果赋值给tv(定时器触发的剩余
的时间长度)
        } else {
            /*
             * if we have active events, we just poll new events
             * without waiting.
             */
            //如果有激活事件 就将tv清空
            evutil_timerclear(&tv);
        }

        /* If we have no events, we just exit */
        //未注册事件就退出循环
        if (0==(flags&EVLOOP_NO_EXIT_ON_EMPTY) &&
            !event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
            event_debug(("%s: no events registered.", __func__));
            retval = 1;
            goto done;
        }
        
        event_queue_make_later_events_active(base);
        //清理时间缓存
        clear_time_cache(base);
        // 调用模型的epoll_wait/select/poll 等   tv_p 是刚才计算的最小时间间隔
        res = evsel->dispatch(base, tv_p); //以epoll 为例详细说明

        if (res == -1) {
            event_debug(("%s: dispatch returned unsuccessfully.",
                __func__));
            retval = -1;
            goto done;
        }
        //更新base中的时间， 下面就是调用定时事件和IO事件。
        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) == 0
                && n != 0)
                done = 1;
        } else if (flags & EVLOOP_NONBLOCK)
            done = 1;
    }
    event_debug(("%s: asked to terminate loop.", __func__));

done:
    clear_time_cache(base);
    base->running_loop = 0;

    EVBASE_RELEASE_LOCK(base, th_base_lock);

    return (retval);
}

以 epoll 模型的dispatch 看一下evsel->dispatch(base, tv_p);

static int
epoll_dispatch(struct event_base *base, struct timeval *tv)
{
    struct epollop *epollop = base->evbase;
    struct epoll_event *events = epollop->events;
    int i, res;
    long timeout = -1;
    if (tv != NULL) {
        timeout = evutil_tv_to_msec_(tv); //将tv 话为 msec传递给epoll_wait
        if (timeout < 0 || 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;
        }
    }

    epoll_apply_changes(base);
    event_changelist_remove_all_(&base->changelist, base);

    EVBASE_RELEASE_LOCK(base, th_base_lock);//多线程下防止惊群而加锁

    res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);//等待事件

    EVBASE_ACQUIRE_LOCK(base, th_base_lock);

    if (res == -1) {
        if (errno != EINTR) {
            event_warn("epoll_wait");
            return (-1);
        }
        return (0);
    }

    event_debug(("%s: epoll_wait reports %d", __func__, res));
    EVUTIL_ASSERT(res <= epollop->nevents);
    //依次处理事件
    for (i = 0; i < res; i++) {
        int what = events[i].events;
        short ev = 0;

        if (what & (EPOLLHUP|EPOLLERR)) {
            ev = EV_READ | EV_WRITE;
        } else {
            if (what & EPOLLIN)
                ev |= EV_READ;
            if (what & EPOLLOUT)
                ev |= EV_WRITE;
            if (what & EPOLLRDHUP)
                ev |= EV_CLOSED;
        }

        if (!ev)
            continue;
        //根据fd找到相应的位置  将event_callback加入到激活queue等待被调用
        evmap_io_active_(base, events[i].data.fd, ev | EV_ET);
    }
    // 拓展事件容量
    if (res == epollop->nevents && epollop->nevents < MAX_NEVENT) {
        /* 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 = mm_realloc(epollop->events,
            new_nevents * sizeof(struct epoll_event));
        if (new_events) {
            epollop->events = new_events;
            epollop->nevents = new_nevents;
        }
    }

    return (0);
}

再看一下timeout_process.

static void
timeout_process(struct event_base *base)
{
    /* Caller must hold lock. */
    struct timeval now;
    struct event *ev;
    //没有定时事件直接退出
    if (min_heap_empty_(&base->timeheap)) {
        return;
    }

    gettime(base, &now);
    //取堆顶最小
    while ((ev = min_heap_top_(&base->timeheap))) {
         //没有超时就直接退出
        if (evutil_timercmp(&ev->ev_timeout, &now, >))
            break;

        /* delete this event from the I/O queues */     
　　　　　　//有超时事件发生就将事件从io中删除
        event_del_nolock_(ev, EVENT_DEL_NOBLOCK);

        event_debug(("timeout_process: event: %p, call %p",
             ev, ev->ev_callback));
        //加入激活队列中   最终会调用  event_queue_insert_timeout，被激活的计时和IO都放在同一个queue中
        event_active_nolock_(ev, EV_TIMEOUT, 1);
    }
}