bird/proto/bfd/io.c

769 lines
13 KiB
C
Raw Normal View History

/*
* BIRD -- I/O and event loop
*
* Can be freely distributed and used under the terms of the GNU GPL.
*/
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <pthread.h>
#include <time.h>
#include <sys/time.h>
#include "nest/bird.h"
#include "proto/bfd/io.h"
2013-09-10 18:09:36 +08:00
#include "lib/buffer.h"
#include "lib/heap.h"
#include "lib/lists.h"
#include "lib/resource.h"
#include "lib/event.h"
#include "lib/socket.h"
2013-09-10 18:09:36 +08:00
struct birdloop
{
pool *pool;
pthread_t thread;
pthread_mutex_t mutex;
2013-09-10 18:09:36 +08:00
btime last_time;
btime real_time;
2013-09-10 18:09:36 +08:00
u8 use_monotonic_clock;
u8 stop_called;
u8 poll_active;
u8 wakeup_masked;
int wakeup_fds[2];
2013-09-10 18:09:36 +08:00
BUFFER(timer2 *) timers;
list event_list;
list sock_list;
uint sock_num;
BUFFER(sock *) poll_sk;
BUFFER(struct pollfd) poll_fd;
u8 poll_changed;
u8 close_scheduled;
};
/*
* Current thread context
*/
static pthread_key_t current_loop_key;
static inline struct birdloop *
birdloop_current(void)
{
return pthread_getspecific(current_loop_key);
}
static inline void
birdloop_set_current(struct birdloop *loop)
{
pthread_setspecific(current_loop_key, loop);
}
static inline void
birdloop_init_current(void)
{
pthread_key_create(&current_loop_key, NULL);
}
/*
* Time clock
*/
2013-09-10 18:09:36 +08:00
static void times_update_alt(struct birdloop *loop);
static void
2013-09-10 18:09:36 +08:00
times_init(struct birdloop *loop)
{
struct timespec ts;
int rv;
rv = clock_gettime(CLOCK_MONOTONIC, &ts);
if (rv < 0)
{
log(L_WARN "Monotonic clock is missing");
2013-09-10 18:09:36 +08:00
loop->use_monotonic_clock = 0;
loop->last_time = 0;
loop->real_time = 0;
times_update_alt(loop);
return;
}
if ((ts.tv_sec < 0) || (((s64) ts.tv_sec) > ((s64) 1 << 40)))
2013-09-10 18:09:36 +08:00
log(L_WARN "Monotonic clock is crazy");
loop->use_monotonic_clock = 1;
loop->last_time = ((s64) ts.tv_sec S) + (ts.tv_nsec / 1000);
2013-09-10 18:09:36 +08:00
loop->real_time = 0;
}
static void
times_update_pri(struct birdloop *loop)
{
struct timespec ts;
int rv;
rv = clock_gettime(CLOCK_MONOTONIC, &ts);
if (rv < 0)
die("clock_gettime: %m");
btime new_time = ((s64) ts.tv_sec S) + (ts.tv_nsec / 1000);
2013-09-10 18:09:36 +08:00
if (new_time < loop->last_time)
log(L_ERR "Monotonic clock is broken");
loop->last_time = new_time;
loop->real_time = 0;
}
static void
times_update_alt(struct birdloop *loop)
{
struct timeval tv;
int rv;
rv = gettimeofday(&tv, NULL);
if (rv < 0)
die("gettimeofday: %m");
btime new_time = ((s64) tv.tv_sec S) + tv.tv_usec;
btime delta = new_time - loop->real_time;
2013-09-10 18:09:36 +08:00
if ((delta < 0) || (delta > (60 S)))
{
if (loop->real_time)
log(L_WARN "Time jump, delta %d us", (int) delta);
delta = 100 MS;
}
loop->last_time += delta;
loop->real_time = new_time;
}
static void
times_update(struct birdloop *loop)
{
if (loop->use_monotonic_clock)
times_update_pri(loop);
else
times_update_alt(loop);
}
btime
current_time(void)
{
return birdloop_current()->last_time;
}
2013-09-10 18:09:36 +08:00
/*
* Wakeup code for birdloop
*/
2013-09-10 18:09:36 +08:00
static void
pipe_new(int *pfds)
{
int rv = pipe(pfds);
2013-09-10 18:09:36 +08:00
if (rv < 0)
die("pipe: %m");
if (fcntl(pfds[0], F_SETFL, O_NONBLOCK) < 0)
die("fcntl(O_NONBLOCK): %m");
if (fcntl(pfds[1], F_SETFL, O_NONBLOCK) < 0)
die("fcntl(O_NONBLOCK): %m");
}
void
pipe_drain(int fd)
2013-09-10 18:09:36 +08:00
{
char buf[64];
int rv;
try:
rv = read(fd, buf, 64);
2013-09-10 18:09:36 +08:00
if (rv < 0)
{
if (errno == EINTR)
goto try;
if (errno == EAGAIN)
return;
die("wakeup read: %m");
}
if (rv == 64)
goto try;
}
void
pipe_kick(int fd)
2013-09-10 18:09:36 +08:00
{
u64 v = 1;
int rv;
try:
rv = write(fd, &v, sizeof(u64));
2013-09-10 18:09:36 +08:00
if (rv < 0)
{
if (errno == EINTR)
goto try;
if (errno == EAGAIN)
return;
die("wakeup write: %m");
}
}
static inline void
wakeup_init(struct birdloop *loop)
{
pipe_new(loop->wakeup_fds);
}
2013-09-10 18:09:36 +08:00
static inline void
wakeup_drain(struct birdloop *loop)
{
pipe_drain(loop->wakeup_fds[0]);
}
2013-09-10 18:09:36 +08:00
static inline void
wakeup_do_kick(struct birdloop *loop)
{
pipe_kick(loop->wakeup_fds[1]);
}
2013-09-10 18:09:36 +08:00
static inline void
wakeup_kick(struct birdloop *loop)
{
if (!loop->wakeup_masked)
wakeup_do_kick(loop);
else
loop->wakeup_masked = 2;
}
2013-09-10 18:09:36 +08:00
/*
* Events
*/
static inline uint
events_waiting(struct birdloop *loop)
{
return !EMPTY_LIST(loop->event_list);
}
static inline void
2013-09-10 18:09:36 +08:00
events_init(struct birdloop *loop)
{
init_list(&loop->event_list);
2013-09-10 18:09:36 +08:00
}
static void
events_fire(struct birdloop *loop)
{
times_update(loop);
ev_run_list(&loop->event_list);
}
void
ev2_schedule(event *e)
{
struct birdloop *loop = birdloop_current();
2013-09-10 18:09:36 +08:00
if (loop->poll_active && EMPTY_LIST(loop->event_list))
wakeup_kick(loop);
if (e->n.next)
rem_node(&e->n);
add_tail(&loop->event_list, &e->n);
}
/*
* Timers
*/
2013-09-10 18:09:36 +08:00
#define TIMER_LESS(a,b) ((a)->expires < (b)->expires)
#define TIMER_SWAP(heap,a,b,t) (t = heap[a], heap[a] = heap[b], heap[b] = t, \
heap[a]->index = (a), heap[b]->index = (b))
static inline uint timers_count(struct birdloop *loop)
{ return loop->timers.used - 1; }
static inline timer2 *timers_first(struct birdloop *loop)
{ return (loop->timers.used > 1) ? loop->timers.data[1] : NULL; }
static void
tm2_free(resource *r)
{
timer2 *t = (timer2 *) r;
tm2_stop(t);
}
static void
tm2_dump(resource *r)
{
timer2 *t = (timer2 *) r;
debug("(code %p, data %p, ", t->hook, t->data);
if (t->randomize)
debug("rand %d, ", t->randomize);
if (t->recurrent)
debug("recur %d, ", t->recurrent);
if (t->expires)
debug("expires in %d ms)\n", (t->expires - current_time()) TO_MS);
2013-09-10 18:09:36 +08:00
else
debug("inactive)\n");
}
2013-09-10 18:09:36 +08:00
static struct resclass tm2_class = {
"Timer",
sizeof(timer2),
2013-09-10 18:09:36 +08:00
tm2_free,
tm2_dump,
NULL,
NULL
};
timer2 *
tm2_new(pool *p)
{
timer2 *t = ralloc(p, &tm2_class);
t->index = -1;
return t;
}
void
tm2_set(timer2 *t, btime when)
2013-09-10 18:09:36 +08:00
{
struct birdloop *loop = birdloop_current();
2013-09-10 18:09:36 +08:00
uint tc = timers_count(loop);
if (!t->expires)
{
t->index = ++tc;
t->expires = when;
BUFFER_PUSH(loop->timers) = t;
HEAP_INSERT(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP);
}
else if (t->expires < when)
{
t->expires = when;
HEAP_INCREASE(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP, t->index);
}
else if (t->expires > when)
{
t->expires = when;
HEAP_DECREASE(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP, t->index);
}
if (loop->poll_active && (t->index == 1))
wakeup_kick(loop);
}
void
tm2_start(timer2 *t, btime after)
{
tm2_set(t, current_time() + MAX(after, 0));
}
2013-09-10 18:09:36 +08:00
void
tm2_stop(timer2 *t)
{
if (!t->expires)
return;
struct birdloop *loop = birdloop_current();
uint tc = timers_count(loop);
2013-09-10 18:09:36 +08:00
HEAP_DELETE(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP, t->index);
BUFFER_POP(loop->timers);
t->index = -1;
t->expires = 0;
}
static void
timers_init(struct birdloop *loop)
{
BUFFER_INIT(loop->timers, loop->pool, 4);
BUFFER_PUSH(loop->timers) = NULL;
}
static void
timers_fire(struct birdloop *loop)
{
btime base_time;
2013-09-10 18:09:36 +08:00
timer2 *t;
times_update(loop);
base_time = loop->last_time;
while (t = timers_first(loop))
{
if (t->expires > base_time)
return;
if (t->recurrent)
{
btime when = t->expires + t->recurrent;
2013-09-10 18:09:36 +08:00
if (when <= loop->last_time)
when = loop->last_time + t->recurrent;
2013-09-10 18:09:36 +08:00
if (t->randomize)
when += random() % (t->randomize + 1);
2013-09-10 18:09:36 +08:00
tm2_set(t, when);
2013-09-10 18:09:36 +08:00
}
else
tm2_stop(t);
t->hook(t);
}
}
/*
* Sockets
*/
2013-09-10 18:09:36 +08:00
static void
sockets_init(struct birdloop *loop)
{
init_list(&loop->sock_list);
loop->sock_num = 0;
2013-09-10 18:09:36 +08:00
BUFFER_INIT(loop->poll_sk, loop->pool, 4);
BUFFER_INIT(loop->poll_fd, loop->pool, 4);
loop->poll_changed = 1; /* add wakeup fd */
2013-09-10 18:09:36 +08:00
}
static void
sockets_add(struct birdloop *loop, sock *s)
{
add_tail(&loop->sock_list, &s->n);
loop->sock_num++;
s->index = -1;
loop->poll_changed = 1;
if (loop->poll_active)
wakeup_kick(loop);
}
void
sk_start(sock *s)
{
struct birdloop *loop = birdloop_current();
sockets_add(loop, s);
2013-09-10 18:09:36 +08:00
}
static void
sockets_remove(struct birdloop *loop, sock *s)
{
rem_node(&s->n);
loop->sock_num--;
if (s->index >= 0)
loop->poll_sk.data[s->index] = NULL;
2013-09-10 18:09:36 +08:00
s->index = -1;
loop->poll_changed = 1;
/* Wakeup moved to sk_stop() */
}
void
sk_stop(sock *s)
{
struct birdloop *loop = birdloop_current();
sockets_remove(loop, s);
2013-09-10 18:09:36 +08:00
if (loop->poll_active)
{
loop->close_scheduled = 1;
wakeup_kick(loop);
}
else
close(s->fd);
s->fd = -1;
}
static inline uint sk_want_events(sock *s)
{ return (s->rx_hook ? POLLIN : 0) | ((s->ttx != s->tpos) ? POLLOUT : 0); }
/*
FIXME: this should be called from sock code
2013-09-10 18:09:36 +08:00
static void
sockets_update(struct birdloop *loop, sock *s)
{
if (s->index >= 0)
loop->poll_fd.data[s->index].events = sk_want_events(s);
2013-09-10 18:09:36 +08:00
}
*/
2013-09-10 18:09:36 +08:00
static void
sockets_prepare(struct birdloop *loop)
{
BUFFER_SET(loop->poll_sk, loop->sock_num + 1);
BUFFER_SET(loop->poll_fd, loop->sock_num + 1);
struct pollfd *pfd = loop->poll_fd.data;
sock **psk = loop->poll_sk.data;
int i = 0;
node *n;
WALK_LIST(n, loop->sock_list)
2013-09-10 18:09:36 +08:00
{
sock *s = SKIP_BACK(sock, n, n);
ASSERT(i < loop->sock_num);
s->index = i;
*psk = s;
pfd->fd = s->fd;
pfd->events = sk_want_events(s);
pfd->revents = 0;
pfd++;
psk++;
i++;
}
ASSERT(i == loop->sock_num);
/* Add internal wakeup fd */
*psk = NULL;
pfd->fd = loop->wakeup_fds[0];
pfd->events = POLLIN;
pfd->revents = 0;
loop->poll_changed = 0;
}
static void
sockets_close_fds(struct birdloop *loop)
{
struct pollfd *pfd = loop->poll_fd.data;
sock **psk = loop->poll_sk.data;
int poll_num = loop->poll_fd.used - 1;
int i;
for (i = 0; i < poll_num; i++)
if (psk[i] == NULL)
close(pfd[i].fd);
loop->close_scheduled = 0;
}
int sk_read(sock *s, int revents);
int sk_write(sock *s);
2013-09-10 18:09:36 +08:00
static void
sockets_fire(struct birdloop *loop)
{
struct pollfd *pfd = loop->poll_fd.data;
sock **psk = loop->poll_sk.data;
int poll_num = loop->poll_fd.used - 1;
times_update(loop);
/* Last fd is internal wakeup fd */
if (pfd[poll_num].revents & POLLIN)
2013-09-10 18:09:36 +08:00
wakeup_drain(loop);
int i;
for (i = 0; i < poll_num; pfd++, psk++, i++)
{
int e = 1;
if (! pfd->revents)
continue;
if (pfd->revents & POLLNVAL)
die("poll: invalid fd %d", pfd->fd);
if (pfd->revents & POLLIN)
while (e && *psk && (*psk)->rx_hook)
e = sk_read(*psk, 0);
2013-09-10 18:09:36 +08:00
e = 1;
if (pfd->revents & POLLOUT)
while (e && *psk)
e = sk_write(*psk);
}
}
/*
* Birdloop
*/
static void * birdloop_main(void *arg);
2013-09-10 18:09:36 +08:00
struct birdloop *
birdloop_new(void)
2013-09-10 18:09:36 +08:00
{
/* FIXME: this init should be elsewhere and thread-safe */
static int init = 0;
if (!init)
{ birdloop_init_current(); init = 1; }
pool *p = rp_new(NULL, "Birdloop root");
2013-09-10 18:09:36 +08:00
struct birdloop *loop = mb_allocz(p, sizeof(struct birdloop));
loop->pool = p;
pthread_mutex_init(&loop->mutex, NULL);
2013-09-10 18:09:36 +08:00
times_init(loop);
wakeup_init(loop);
events_init(loop);
timers_init(loop);
sockets_init(loop);
return loop;
}
void
birdloop_start(struct birdloop *loop)
{
int rv = pthread_create(&loop->thread, NULL, birdloop_main, loop);
if (rv)
die("pthread_create(): %M", rv);
}
void
birdloop_stop(struct birdloop *loop)
{
pthread_mutex_lock(&loop->mutex);
loop->stop_called = 1;
wakeup_do_kick(loop);
pthread_mutex_unlock(&loop->mutex);
int rv = pthread_join(loop->thread, NULL);
if (rv)
die("pthread_join(): %M", rv);
2013-09-10 18:09:36 +08:00
}
void
birdloop_free(struct birdloop *loop)
{
rfree(loop->pool);
}
2013-09-10 18:09:36 +08:00
void
birdloop_enter(struct birdloop *loop)
{
/* TODO: these functions could save and restore old context */
pthread_mutex_lock(&loop->mutex);
birdloop_set_current(loop);
2013-09-10 18:09:36 +08:00
}
void
birdloop_leave(struct birdloop *loop)
{
/* TODO: these functions could save and restore old context */
birdloop_set_current(NULL);
pthread_mutex_unlock(&loop->mutex);
2013-09-10 18:09:36 +08:00
}
void
birdloop_mask_wakeups(struct birdloop *loop)
{
pthread_mutex_lock(&loop->mutex);
loop->wakeup_masked = 1;
pthread_mutex_unlock(&loop->mutex);
}
2013-09-10 18:09:36 +08:00
void
birdloop_unmask_wakeups(struct birdloop *loop)
2013-09-10 18:09:36 +08:00
{
pthread_mutex_lock(&loop->mutex);
if (loop->wakeup_masked == 2)
wakeup_do_kick(loop);
loop->wakeup_masked = 0;
pthread_mutex_unlock(&loop->mutex);
}
static void *
birdloop_main(void *arg)
{
struct birdloop *loop = arg;
2013-09-10 18:09:36 +08:00
timer2 *t;
int rv, timeout;
2013-09-10 18:09:36 +08:00
birdloop_set_current(loop);
pthread_mutex_lock(&loop->mutex);
2013-09-10 18:09:36 +08:00
while (1)
{
events_fire(loop);
timers_fire(loop);
times_update(loop);
if (events_waiting(loop))
timeout = 0;
else if (t = timers_first(loop))
timeout = (tm2_remains(t) TO_MS) + 1;
else
timeout = -1;
if (loop->poll_changed)
sockets_prepare(loop);
loop->poll_active = 1;
pthread_mutex_unlock(&loop->mutex);
2013-09-10 18:09:36 +08:00
try:
rv = poll(loop->poll_fd.data, loop->poll_fd.used, timeout);
if (rv < 0)
{
if (errno == EINTR || errno == EAGAIN)
goto try;
die("poll: %m");
}
pthread_mutex_lock(&loop->mutex);
2013-09-10 18:09:36 +08:00
loop->poll_active = 0;
if (loop->close_scheduled)
sockets_close_fds(loop);
if (loop->stop_called)
break;
2013-09-10 18:09:36 +08:00
if (rv)
sockets_fire(loop);
timers_fire(loop);
}
loop->stop_called = 0;
pthread_mutex_unlock(&loop->mutex);
return NULL;
2013-09-10 18:09:36 +08:00
}