1、数据结构
#define ev_at(w) ((WT)(w))->at
#define ev_active(w) ((W)(w))->active
typedef ev_watcher_time *WT;
struct ev_loop
{
ev_tstamp mn_now
ANHE * timers
int timermax
int timercnt
ev_watcher * rfeeds
}
/* Heap Entry */ //是否缓存时间监控器中的at字段。
#if EV_HEAP_CACHE_AT
/* a heap element */
typedef struct {
ev_tstamp at;
WT w;
} ANHE;
#define ANHE_w(he) (he).w /* access watcher, read-write */
#define ANHE_at(he) (he).at /* access cached at, read-only */
#define ANHE_at_cache(he) (he).at = (he).w->at /* update at from watcher */
#else
/* a heap element */
typedef WT ANHE;
#define ANHE_w(he) (he)
#define ANHE_at(he) (he)->at
#define ANHE_at_cache(he)
#endif
2、ev_timer_start
void noinline
ev_timer_start (EV_P_ ev_timer *w) EV_THROW
{
if (expect_false (ev_is_active (w)))
return;
ev_at (w) += mn_now;
++timercnt;
ev_start (EV_A_ (W)w, timercnt + HEAP0 - 1);//w->active = timercnt + HEAP0 - 1;
array_needsize (ANHE, timers, timermax, ev_active (w) + 1, EMPTY2);
ANHE_w (timers [ev_active (w)]) = (WT)w;
ANHE_at_cache (timers [ev_active (w)]);
upheap (timers, ev_active (w));
EV_FREQUENT_CHECK;
/*assert (("libev: internal timer heap corruption", timers [ev_active (w)] == (WT)w));*/
}
inline_speed void
upheap (ANHE *heap, int k)
{
ANHE he = heap [k];
for (;;)
{
int p = HPARENT (k);
if (UPHEAP_DONE (p, k) || ANHE_at (heap [p]) <= ANHE_at (he))
break;
heap [k] = heap [p];
ev_active (ANHE_w (heap [k])) = k;
k = p;
}
heap [k] = he;
ev_active (ANHE_w (he)) = k;
}
3、timers_reify
inline_size void
timers_reify (EV_P)
{
EV_FREQUENT_CHECK;
if (timercnt && ANHE_at (timers [HEAP0]) < mn_now)
{
do
{
ev_timer *w = (ev_timer *)ANHE_w (timers [HEAP0]);
/*assert (("libev: inactive timer on timer heap detected", ev_is_active (w)));*/
/* first reschedule or stop timer */
if (w->repeat)
{
ev_at (w) += w->repeat;
if (ev_at (w) < mn_now)
ev_at (w) = mn_now;
assert (("libev: negative ev_timer repeat value found while processing timers", w->repeat > 0.));
ANHE_at_cache (timers [HEAP0]);
downheap (timers, timercnt, HEAP0);
}
else
ev_timer_stop (EV_A_ w); /* nonrepeating: stop timer */
EV_FREQUENT_CHECK;
feed_reverse (EV_A_ (W)w);
}
while (timercnt && ANHE_at (timers [HEAP0]) < mn_now);
feed_reverse_done (EV_A_ EV_TIMER);
}
}
#if EV_USE_4HEAP
#define DHEAP 4
#define HEAP0 (DHEAP - 1) /* index of first element in heap */
#define HPARENT(k) ((((k) - HEAP0 - 1) / DHEAP) + HEAP0)
#define UPHEAP_DONE(p,k) ((p) == (k))
/* away from the root */
inline_speed void
downheap (ANHE *heap, int N, int k)
{
ANHE he = heap [k];
ANHE *E = heap + N + HEAP0;
for (;;)
{
ev_tstamp minat;
ANHE *minpos;
ANHE *pos = heap + DHEAP * (k - HEAP0) + HEAP0 + 1;
/* find minimum child */
if (expect_true (pos + DHEAP - 1 < E))
{
/* fast path */ (minpos = pos + 0), (minat = ANHE_at (*minpos));
if ( ANHE_at (pos [1]) < minat) (minpos = pos + 1), (minat = ANHE_at (*minpos));
if ( ANHE_at (pos [2]) < minat) (minpos = pos + 2), (minat = ANHE_at (*minpos));
if ( ANHE_at (pos [3]) < minat) (minpos = pos + 3), (minat = ANHE_at (*minpos));
}
else if (pos < E)
{
/* slow path */ (minpos = pos + 0), (minat = ANHE_at (*minpos));
if (pos + 1 < E && ANHE_at (pos [1]) < minat) (minpos = pos + 1), (minat = ANHE_at (*minpos));
if (pos + 2 < E && ANHE_at (pos [2]) < minat) (minpos = pos + 2), (minat = ANHE_at (*minpos));
if (pos + 3 < E && ANHE_at (pos [3]) < minat) (minpos = pos + 3), (minat = ANHE_at (*minpos));
}
else
break;
if (ANHE_at (he) <= minat)
break;
heap [k] = *minpos;
ev_active (ANHE_w (*minpos)) = k;
k = minpos - heap;
}
heap [k] = he;
ev_active (ANHE_w (he)) = k;
}
#else /* 4HEAP */
#define HEAP0 1
#define HPARENT(k) ((k) >> 1)
#define UPHEAP_DONE(p,k) (!(p))
/* away from the root */
inline_speed void
downheap (ANHE *heap, int N, int k)
{
ANHE he = heap [k];
for (;;)
{
int c = k << 1;
if (c >= N + HEAP0)
break;
c += c + 1 < N + HEAP0 && ANHE_at (heap [c]) > ANHE_at (heap [c + 1])
? 1 : 0;
if (ANHE_at (he) <= ANHE_at (heap [c]))
break;
heap [k] = heap [c];
ev_active (ANHE_w (heap [k])) = k;
k = c;
}
heap [k] = he;
ev_active (ANHE_w (he)) = k;
}
#endif
inline_speed void
feed_reverse (EV_P_ W w)
{
array_needsize (W, rfeeds, rfeedmax, rfeedcnt + 1, EMPTY2);
rfeeds [rfeedcnt++] = w;
}
inline_size void
feed_reverse_done (EV_P_ int revents)
{
do
ev_feed_event (EV_A_ rfeeds [--rfeedcnt], revents);
while (rfeedcnt);
}
void noinline
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;
}
pendingpri = NUMPRI - 1;
}
4、ev_run
int ev_run (EV_P_ int flags)
{
waittime = MAX_BLOCKTIME;
if (timercnt)
{
ev_tstamp to = ANHE_at (timers [HEAP0]) - mn_now;
if (waittime > to) waittime = to;
。。。。。
}
timers_reify(EV_A);
EV_INVOKE_PENDING;
}
Libev中在管理定时器时，使用了堆这种结构存储ev_timer，除了最小2叉堆之外，还有4叉堆，可用通过宏定义来设置使用哪个。
对于n叉堆来说，使用数组进行存储时，下标为x的元素，其孩子节点的下标范围是[nx+1, nx+n]。比如2叉堆，下标为x的元素，其孩子节点的下标为2x+1和2x+2.
根据定理，对于4叉堆而言，下标为x的元素，其孩子节点的下标范围是[4x+1, 4x+4]。还可以得出，其父节点的下标是(x-1)/4。然而在Libev的代码中，使用数组a存储堆时，4叉堆的第一个元素存放在a[3]，2叉堆的第一个元素存放在a[1]。
所以，对于Libev中的4叉堆实现而言，下标为k的元素（对应在正常实现中的下标是k-3），其孩子节点的下标范围是[4(k-3)+1+3, 4(k-3)+4+3]；其父节点的下标是((k-3-1)/4)+3。
对于Libev中的2叉堆实现而言，下标为k的元素（对应在正常实现中，其下标是k-1），其孩子节点的下标范围是[2(k-1)+1+1, 2(k-1)+2+1]，也就是[2k, 2k+1]；其父节点的下标是((k-1-1)/2)+1，也就是k/2。
downheap和upheap函数就是使用以上原则，不断与子结点或者父结点比较，交换，最终形成堆。
首先，ev_timer_start 将时间监控器添加到timers中(通过upheap)，loop->timer是一个数组形式的最小堆。根据timer->at做的比较，即堆顶为时间最小的监控器，timer->active是数组的下标。
在ev_run中，先计算超时时间，使其不大于最小的时间。
最后，timers_reify 中取出到时的时间监听器，添加到pendings队列。如果是repeat的话，则更新下一次触发时间，调用downheap操作将这个节点下移至合适的位置；否则直接删除该watcher。
ev_prepare, ev_check, ev_idle
从角色上来看，这三个类型的watcher其实都是事件循环的一个扩展点。通过这三个watcher，开发者可以在事件循环的一些特殊时刻获得回调的机会。
- ev_prepare 在事件循环发生阻塞前会被触发。
- ev_check 在事件循环阻塞结束后会被触发。ev_check的触发是按优先级划分的。可以保证，ev_check是同一个优先级上阻塞结束后最先被触发的watcher。所以，如果要保证ev_check是最先被执行的，可以把它的优先级设成最高。
- ev_idle 当没有其他watcher被触发时被触发。ev_idle也是按优先级划分的。它的语义是，在当前优先级以及更高的优先级上没有watcher被触发，那么它就会被触发，无论之后在较低优先级上是否有其他watcher被触发。
这三类watcher给外部的开发者提供了非常便利的扩展机制，在这个基础上，开发者可以做很多有意思的事情，也对事件循环有了更多的控制权。具体到底能做些什么，做到什么程度，那就要看开发者们的想象力和创造力了：）