Timer support in event driven network framework

发表于 | 阅读次数

0x1 Introduction

Typically there is one event loop in event driven asynchronous network framework, it waits on the file descriptors to become ready then to perform specific operations, typically the event loop is executed in one thread. And such network framework also supports timer, one type of timer is to process some operation after some time, another type of timer is to process some operation at some intervals.

Next we will look into the detail timer’s support in serveral network framework.

0x2 Nginx’s implementation

In Nginx, ngx_process_events_and_timers() will handle timer, it detects the expired timers then call its handler to do further operation.

From the following code, there is one variable ngx_timer_resolution, it is used to handle timer in two different ways.

If ngx_timer_resolution is not zero, the timeout parameter is set as -1, then it will pass to ngx_process_events(), if timeout parameter is -1, epoll_wait() will wait infinitely until some file descriptors become ready, so the timer seems not accurate at that time. And ngx_timer_resolution is configured in nginx’s configuration file.

If ngx_timer_resolution is zero, it will call ngx_event_find_timer() to get the delta time between the next expired time and current time, the next expired time is sotred in the rbtree.

In ngx_process_events_and_timers(), it will call ngx_event_expire_timers() to process the expired timers, and call the timer’handler.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
void ngx_process_events_and_timers(ngx_cycle_t *cycle)
{
    ngx_uint_t  flags;
    ngx_msec_t  timer, delta;
    if (ngx_timer_resolution) {
        timer = NGX_TIMER_INFINITE;
        flags = 0;
    } else {
        timer = ngx_event_find_timer();
        flags = NGX_UPDATE_TIME;
    }
    ...
    delta = ngx_current_msec;
    (void) ngx_process_events(cycle, timer, flags);
    delta = ngx_current_msec - delta;
    ngx_event_process_posted(cycle, &ngx_posted_accept_events);
    ...
    if (delta) {
        ngx_event_expire_timers();
    }
    ngx_event_process_posted(cycle, &ngx_posted_events);
}
static ngx_int_t ngx_epoll_process_events(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags)
{
    ...
    events = epoll_wait(ep, event_list, (int) nevents, timer);
    ...
    if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
        ngx_time_update();
    }
    ...
    for (i = 0; i < events; i++) {
        c = event_list[i].data.ptr;
        ...
        revents = event_list[i].events;
        ...        
        wev = c->write;
        ...
        if (flags & NGX_POST_EVENTS) {
                ngx_post_event(wev, &ngx_posted_events);
            } else {
                wev->handler(wev);
        }
        ...
}

In ngx_epoll_process_events(), if will call ngx_time_update() to update time, and ngx_time_update() will call the system function gettimeofday(), and the system performance will be impacted if we call gettimeofday() frequently.

1
2
3
4
5
6
7
8
9
void ngx_time_update(void)
{
    ...
    ngx_gettimeofday(&tv);
    ...
    sec = tv.tv_sec;
    msec = tv.tv_usec / 1000;
    ...
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
void ngx_event_expire_timers(void)
{
    ...
    for ( ;; ) {
        root = ngx_event_timer_rbtree.root;
        ...
        if (root == sentinel) {
            return;
        }
        ...
        node = ngx_rbtree_min(root, sentinel);
        ...
        if ((ngx_msec_int_t) (node->key - ngx_current_msec) > 0) {
            return;
        }
        ev = (ngx_event_t *) ((char *) node - offsetof(ngx_event_t, timer));
        ngx_rbtree_delete(&ngx_event_timer_rbtree, &ev->timer);
        ev->timer_set = 0;
        ev->timedout = 1;
        ev->handler(ev);
    }
}

0x3 Redis’s implementation

Here are the functions of Redis to add/delete timer, the timer’s user can use those functions to get timer service from the event loop thread. In Redis, the serverCron() is register as the timer’s callback when calling aeCreateTimeEvent(), then serverCron() will be called at some interval to do resource and status checking of Redis Server.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
        aeTimeProc *proc, void *clientData,
        aeEventFinalizerProc *finalizerProc)
{
    long long id = eventLoop->timeEventNextId++;
    aeTimeEvent *te;
    te = zmalloc(sizeof(*te));
    if (te == NULL) return AE_ERR;
    te->id = id;
    aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);
    te->timeProc = proc;
    te->finalizerProc = finalizerProc;
    te->clientData = clientData;
    te->next = eventLoop->timeEventHead;
    eventLoop->timeEventHead = te;
    return id;
}
int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id)
{
    aeTimeEvent *te = eventLoop->timeEventHead;
    while(te) {
        if (te->id == id) {
            te->id = AE_DELETED_EVENT_ID;
            return AE_OK;
        }
        te = te->next;
    }
    return AE_ERR;
}

Here is the code about how timer is handled in the event loop, it searches the nearnest timer in the timer list, then calculates the timout parameter for the aeApiPoll(), then aeApiPoll() will wait on the file descriptors within the timout setting, after aeApiPoll(), it uses processTimeEvents() to process the timers and call its callback of the specific timer.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
        ...
        aeTimeEvent *shortest = NULL;
        struct timeval tv, *tvp;
        if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
            shortest = aeSearchNearestTimer(eventLoop);
        if (shortest) {
            long now_sec, now_ms;
            aeGetTime(&now_sec, &now_ms);
            tvp = &tv;
            long long ms =
                (shortest->when_sec - now_sec)*1000 +
                shortest->when_ms - now_ms;
            if (ms > 0) {
                tvp->tv_sec = ms/1000;
                tvp->tv_usec = (ms % 1000)*1000;
            } else {
                tvp->tv_sec = 0;
                tvp->tv_usec = 0;
            }
        } else {
            if (flags & AE_DONT_WAIT) {
                tv.tv_sec = tv.tv_usec = 0;
                tvp = &tv;
            } else {
                tvp = NULL; /* wait forever */
            }
        }
        ...
        numevents = aeApiPoll(eventLoop, tvp);
        ...
    }
    ...
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);
    ...
}
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;
    retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
            tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
    if (retval > 0) {
        int j;
        ...
        numevents = retval;
        for (j = 0; j < numevents; j++) {
            int mask = 0;
            struct epoll_event *e = state->events+j;
            ...
            if (e->events & EPOLLIN) mask |= AE_READABLE;
            if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
            if (e->events & EPOLLERR) mask |= AE_WRITABLE;
            if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    }
    return numevents;
}
static int processTimeEvents(aeEventLoop *eventLoop) {
    int processed = 0;
    aeTimeEvent *te, *prev;
    long long maxId;
    time_t now = time(NULL);
    ...
    if (now < eventLoop->lastTime) {
        te = eventLoop->timeEventHead;
        while(te) {
            te->when_sec = 0;
            te = te->next;
        }
    }
    eventLoop->lastTime = now;
    ...
    prev = NULL;
    te = eventLoop->timeEventHead;
    maxId = eventLoop->timeEventNextId-1;
    while(te) {
        long now_sec, now_ms;
        long long id;
        ...
        aeGetTime(&now_sec, &now_ms);
        if (now_sec > te->when_sec ||
            (now_sec == te->when_sec && now_ms >= te->when_ms))
        {
            int retval;
            id = te->id;
            retval = te->timeProc(eventLoop, id, te->clientData);
            processed++;
            if (retval != AE_NOMORE) {
                aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
            } else {
                te->id = AE_DELETED_EVENT_ID;
            }
        }
        prev = te;
        te = te->next;
    }
    return processed;
}

0x4 Libevent’s implementation

libevent uses timerfd to handle timer when USING_TIMERFD is set, here is the code, it uses timerfd_create() to create timefd.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
static void * epoll_init(struct event_base *base)
{
#ifdef USING_TIMERFD
	if ((base->flags & EVENT_BASE_FLAG_PRECISE_TIMER) &&
	    base->monotonic_timer.monotonic_clock == CLOCK_MONOTONIC) {
		int fd;
		fd = epollop->timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK|TFD_CLOEXEC);
		if (epollop->timerfd >= 0) {
			struct epoll_event epev;
			memset(&epev, 0, sizeof(epev));
			epev.data.fd = epollop->timerfd;
			epev.events = EPOLLIN;
			if (epoll_ctl(epollop->epfd, EPOLL_CTL_ADD, fd, &epev) < 0) {
				event_warn("epoll_ctl(timerfd)");
				close(fd);
				epollop->timerfd = -1;
			}
        } 
        ...
	}
#endif
}

Here is the code to set the timeout for timerfd, it call timerfd_settime() to set the timeout. When timeout happens, it will trigger the timerfd which is waiting on epoll_wait(), then the event loop can process the timer handler.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
static int epoll_dispatch(struct event_base *base, struct timeval *tv)
{
#ifdef USING_TIMERFD
	if (epollop->timerfd >= 0) {
		struct itimerspec is;
		is.it_interval.tv_sec = 0;
		is.it_interval.tv_nsec = 0;
		if (tv == NULL) {
			is.it_value.tv_sec = 0;
			is.it_value.tv_nsec = 0;
		} else {
			if (tv->tv_sec == 0 && tv->tv_usec == 0) {
				timeout = 0;
			}
			is.it_value.tv_sec = tv->tv_sec;
			is.it_value.tv_nsec = tv->tv_usec * 1000;
		}
		if (timerfd_settime(epollop->timerfd, 0, &is, NULL) < 0) {
			event_warn("timerfd_settime");
		}
	}
#endif
    ...
    res = epoll_wait(epollop->epfd, events, epollop->nevents, timeout);
    ...
}

Timerfd’s kernel implementation

In Linux kernel, timerfd’s implementation uses hrtimer to support timer, the corresponding code is placed in kernel/fs/timerfd.c.