1 files changed, 162 insertions, 51 deletions

diff --git a/sysdep/unix/io-loop.c b/sysdep/unix/io-loop.c
index c7cf4ad2..5ce2d350 100644
--- a/sysdep/unix/io-loop.c
+++ b/sysdep/unix/io-loop.c
@@ -17,7 +17,6 @@
 #include "nest/bird.h"
 
 #include "lib/buffer.h"
-#include "lib/coro.h"
 #include "lib/lists.h"
 #include "lib/resource.h"
 #include "lib/event.h"
@@ -28,6 +27,8 @@
 #include "sysdep/unix/io-loop.h"
 #include "conf/conf.h"
 
+#define THREAD_STACK_SIZE	65536	/* To be lowered in near future */
+
 /*
  *	Current thread context
  */
@@ -58,85 +59,144 @@ birdloop_inside(struct birdloop *loop)
   return 0;
 }
 
+void
+birdloop_flag(struct birdloop *loop, u32 flag)
+{
+  atomic_fetch_or_explicit(&loop->flags, flag, memory_order_acq_rel);
+  birdloop_ping(loop);
+}
+
+void
+birdloop_flag_set_handler(struct birdloop *loop, struct birdloop_flag_handler *fh)
+{
+  ASSERT_DIE(birdloop_inside(loop));
+  loop->flag_handler = fh;
+}
+
+static int
+birdloop_process_flags(struct birdloop *loop)
+{
+  if (!loop->flag_handler)
+    return 0;
+
+  u32 flags = atomic_exchange_explicit(&loop->flags, 0, memory_order_acq_rel);
+  loop->flag_handler->hook(loop->flag_handler, flags);
+  return !!flags;
+}
+
+static int
+birdloop_run_events(struct birdloop *loop)
+{
+  btime begin = current_time();
+  while (current_time() - begin < 5 MS)
+  {
+    if (!ev_run_list(&loop->event_list))
+      return 0;
+
+    times_update();
+  }
+
+  return 1;
+}
+
 /*
  *	Wakeup code for birdloop
  */
 
-static void
-pipe_new(int *pfds)
+void
+pipe_new(struct pipe *p)
 {
-  int rv = pipe(pfds);
+  int rv = pipe(p->fd);
   if (rv < 0)
     die("pipe: %m");
 
-  if (fcntl(pfds[0], F_SETFL, O_NONBLOCK) < 0)
+  if (fcntl(p->fd[0], F_SETFL, O_NONBLOCK) < 0)
     die("fcntl(O_NONBLOCK): %m");
 
-  if (fcntl(pfds[1], F_SETFL, O_NONBLOCK) < 0)
+  if (fcntl(p->fd[1], F_SETFL, O_NONBLOCK) < 0)
     die("fcntl(O_NONBLOCK): %m");
 }
 
 void
-pipe_drain(int fd)
+pipe_drain(struct pipe *p)
 {
-  char buf[64];
-  int rv;
-  
- try:
-  rv = read(fd, buf, 64);
-  if (rv < 0)
-  {
-    if (errno == EINTR)
-      goto try;
-    if (errno == EAGAIN)
+  while (1) {
+    char buf[64];
+    int rv = read(p->fd[0], buf, sizeof(buf));
+    if ((rv < 0) && (errno == EAGAIN))
       return;
-    die("wakeup read: %m");
+
+    if (rv == 0)
+      bug("wakeup read eof");
+    if ((rv < 0) && (errno != EINTR))
+      bug("wakeup read: %m");
+  }
+}
+
+int
+pipe_read_one(struct pipe *p)
+{
+  while (1) {
+    char v;
+    int rv = read(p->fd[0], &v, sizeof(v));
+    if (rv == 1)
+      return 1;
+    if ((rv < 0) && (errno == EAGAIN))
+      return 0;
+    if (rv > 1)
+      bug("wakeup read more bytes than expected: %d", rv);
+    if (rv == 0)
+      bug("wakeup read eof");
+    if (errno != EINTR)
+      bug("wakeup read: %m");
   }
-  if (rv == 64)
-    goto try;
 }
 
 void
-pipe_kick(int fd)
+pipe_kick(struct pipe *p)
 {
-  u64 v = 1;
+  char v = 1;
   int rv;
 
- try:
-  rv = write(fd, &v, sizeof(u64));
-  if (rv < 0)
-  {
-    if (errno == EINTR)
-      goto try;
-    if (errno == EAGAIN)
+  while (1) {
+    rv = write(p->fd[1], &v, sizeof(v));
+    if ((rv >= 0) || (errno == EAGAIN)) 
       return;
-    die("wakeup write: %m");
+    if (errno != EINTR)
+      bug("wakeup write: %m");
   }
 }
 
+void
+pipe_pollin(struct pipe *p, struct pollfd *pfd)
+{
+  pfd->fd = p->fd[0];
+  pfd->events = POLLIN;
+  pfd->revents = 0;
+}
+
 static inline void
 wakeup_init(struct birdloop *loop)
 {
-  pipe_new(loop->wakeup_fds);
+  pipe_new(&loop->wakeup);
 }
 
 static inline void
 wakeup_drain(struct birdloop *loop)
 {
-  pipe_drain(loop->wakeup_fds[0]);
+  pipe_drain(&loop->wakeup);
 }
 
 static inline void
 wakeup_do_kick(struct birdloop *loop)
 {
-  pipe_kick(loop->wakeup_fds[1]);
+  pipe_kick(&loop->wakeup);
 }
 
-void
-birdloop_ping(struct birdloop *loop)
+static inline void
+birdloop_do_ping(struct birdloop *loop)
 {
-  u32 ping_sent = atomic_fetch_add_explicit(&loop->ping_sent, 1, memory_order_acq_rel);
-  if (ping_sent)
+  if (atomic_fetch_add_explicit(&loop->ping_sent, 1, memory_order_acq_rel))
     return;
 
   if (loop == birdloop_wakeup_masked)
@@ -145,6 +205,15 @@ birdloop_ping(struct birdloop *loop)
     wakeup_do_kick(loop);
 }
 
+void
+birdloop_ping(struct birdloop *loop)
+{
+  if (birdloop_inside(loop) && !loop->ping_pending)
+    loop->ping_pending++;
+  else
+    birdloop_do_ping(loop);
+}
+
 
 /*
  *	Sockets
@@ -183,6 +252,9 @@ sk_start(sock *s)
 static void
 sockets_remove(struct birdloop *loop, sock *s)
 {
+  if (!enlisted(&s->n))
+    return;
+
   rem_node(&s->n);
   loop->sock_num--;
 
@@ -205,7 +277,7 @@ sk_stop(sock *s)
 }
 
 static inline uint sk_want_events(sock *s)
-{ return ((s->rx_hook && !ev_corked(s->cork)) ? POLLIN : 0) | ((s->ttx != s->tpos) ? POLLOUT : 0); }
+{ return (s->rx_hook ? POLLIN : 0) | ((s->ttx != s->tpos) ? POLLOUT : 0); }
 
 /*
 FIXME: this should be called from sock code
@@ -250,9 +322,7 @@ sockets_prepare(struct birdloop *loop)
 
   /* Add internal wakeup fd */
   *psk = NULL;
-  pfd->fd = loop->wakeup_fds[0];
-  pfd->events = POLLIN;
-  pfd->revents = 0;
+  pipe_pollin(&loop->wakeup, pfd);
 
   loop->poll_changed = 0;
 }
@@ -297,7 +367,7 @@ sockets_fire(struct birdloop *loop)
       continue;
 
     if (pfd->revents & POLLNVAL)
-      die("poll: invalid fd %d", pfd->fd);
+      bug("poll: invalid fd %d", pfd->fd);
 
     if (pfd->revents & POLLIN)
       while (e && *psk && (*psk)->rx_hook)
@@ -336,7 +406,7 @@ birdloop_init(void)
   birdloop_enter_locked(&main_birdloop);
 }
 
-static void birdloop_main(void *arg);
+static void *birdloop_main(void *arg);
 
 struct birdloop *
 birdloop_new(pool *pp, uint order, const char *name)
@@ -357,7 +427,19 @@ birdloop_new(pool *pp, uint order, const char *name)
   timers_init(&loop->time, p);
   sockets_init(loop);
 
-  loop->time.coro = coro_run(p, birdloop_main, loop);
+  int e = 0;
+
+  if (e = pthread_attr_init(&loop->thread_attr))
+    die("pthread_attr_init() failed: %M", e);
+
+  if (e = pthread_attr_setstacksize(&loop->thread_attr, THREAD_STACK_SIZE))
+    die("pthread_attr_setstacksize(%u) failed: %M", THREAD_STACK_SIZE, e);
+
+  if (e = pthread_attr_setdetachstate(&loop->thread_attr, PTHREAD_CREATE_DETACHED))
+    die("pthread_attr_setdetachstate(PTHREAD_CREATE_DETACHED) failed: %M", e);
+
+  if (e = pthread_create(&loop->thread_id, &loop->thread_attr, birdloop_main, loop))
+    die("pthread_create() failed: %M", e);
 
   birdloop_leave(loop);
 
@@ -393,6 +475,11 @@ void
 birdloop_free(struct birdloop *loop)
 {
   ASSERT_DIE(loop->links == 0);
+  ASSERT_DIE(pthread_equal(pthread_self(), loop->thread_id));
+
+  rcu_birdloop_stop(&loop->rcu);
+  pthread_attr_destroy(&loop->thread_attr);
+
   domain_free(loop->time.domain);
   rfree(loop->pool);
 }
@@ -423,6 +510,13 @@ birdloop_leave_locked(struct birdloop *loop)
   /* Check the current context */
   ASSERT_DIE(birdloop_current == loop);
 
+  /* Send pending pings */
+  if (loop->ping_pending)
+  {
+    loop->ping_pending = 0;
+    birdloop_do_ping(loop);
+  }
+
   /* Restore the old context */
   birdloop_current = loop->prev_loop;
 }
@@ -466,20 +560,24 @@ birdloop_unlink(struct birdloop *loop)
   loop->links--;
 }
 
-static void
+static void *
 birdloop_main(void *arg)
 {
   struct birdloop *loop = arg;
   timer *t;
   int rv, timeout;
 
+  rcu_birdloop_start(&loop->rcu);
+
   btime loop_begin = current_time();
 
+  tmp_init(loop->pool);
+
   birdloop_enter(loop);
   while (1)
   {
     timers_fire(&loop->time, 0);
-    if (ev_run_list(&loop->event_list))
+    if (birdloop_process_flags(loop) + birdloop_run_events(loop))
       timeout = 0;
     else if (t = timers_first(&loop->time))
       timeout = (tm_remains(t) TO_MS) + 1;
@@ -488,6 +586,9 @@ birdloop_main(void *arg)
 
     if (loop->poll_changed)
       sockets_prepare(loop);
+    else
+      if ((timeout < 0) || (timeout > 5000))
+	flush_local_pages();
 
     btime duration = current_time() - loop_begin;
     if (duration > config->watchdog_warning)
@@ -501,7 +602,7 @@ birdloop_main(void *arg)
     {
       if (errno == EINTR || errno == EAGAIN)
 	goto try;
-      die("poll: %m");
+      bug("poll: %m");
     }
 
     birdloop_enter(loop);
@@ -514,7 +615,7 @@ birdloop_main(void *arg)
 
     loop_begin = current_time();
 
-    if (rv)
+    if (rv && !loop->poll_changed)
       sockets_fire(loop);
 
     atomic_exchange_explicit(&loop->ping_sent, 0, memory_order_acq_rel);
@@ -523,13 +624,23 @@ birdloop_main(void *arg)
   /* Flush remaining events */
   ASSERT_DIE(!ev_run_list(&loop->event_list));
 
-  /* No timers allowed */
-  ASSERT_DIE(timers_count(&loop->time) == 0);
+  /* Drop timers */
+  while (t = timers_first(&loop->time))
+    tm_stop(t);
+
+  /* No sockets allowed */
   ASSERT_DIE(EMPTY_LIST(loop->sock_list));
   ASSERT_DIE(loop->sock_num == 0);
 
   birdloop_leave(loop);
   loop->stopped(loop->stop_data);
-}
 
+  flush_local_pages();
+  return NULL;
+}
 
+void
+birdloop_yield(void)
+{
+  usleep(100);
+}