6 files changed, 266 insertions, 200 deletions

diff --git a/sysdep/unix/coroutine.c b/sysdep/unix/coroutine.c
index 2eba142c..4758c056 100644
--- a/sysdep/unix/coroutine.c
+++ b/sysdep/unix/coroutine.c
@@ -21,10 +21,9 @@
 #include "lib/resource.h"
 #include "lib/timer.h"
 
-/* Using a rather big stack for coroutines to allow for stack-local allocations.
- * In real world, the kernel doesn't alloc this memory until it is used.
- * */
-#define CORO_STACK_SIZE	1048576
+#include "conf/conf.h"
+
+#define CORO_STACK_SIZE	65536
 
 /*
  *	Implementation of coroutines based on POSIX threads
@@ -79,6 +78,11 @@ domain_free(struct domain_generic *dg)
   xfree(dg);
 }
 
+uint dg_order(struct domain_generic *dg)
+{
+  return dg->order;
+}
+
 void do_lock(struct domain_generic *dg, struct domain_generic **lsp)
 {
   if ((char *) lsp - (char *) &locking_stack != dg->order)
@@ -89,7 +93,11 @@ void do_lock(struct domain_generic *dg, struct domain_generic **lsp)
   if (*lsp)
     bug("Inconsistent locking stack state on lock");
 
+  btime lock_begin = current_time();
   pthread_mutex_lock(&dg->mutex);
+  btime duration = current_time() - lock_begin;
+  if (config && (duration > config->watchdog_warning))
+    log(L_WARN "Locking of %s took %d ms", dg->name, (int) (duration TO_MS));
 
   if (dg->prev || dg->locked_by)
     bug("Previous unlock not finished correctly");
@@ -140,11 +148,16 @@ static struct resclass coro_class = {
   .free = coro_free,
 };
 
+_Thread_local struct coroutine *this_coro = NULL;
+
 static void *coro_entry(void *p)
 {
   struct coroutine *c = p;
+
   ASSERT_DIE(c->entry);
 
+  this_coro = c;
+
   c->entry(c->data);
   ASSERT_DIE(coro_cleaned_up);
 
diff --git a/sysdep/unix/io-loop.c b/sysdep/unix/io-loop.c
index a15d866a..275d38a0 100644
--- a/sysdep/unix/io-loop.c
+++ b/sysdep/unix/io-loop.c
@@ -15,50 +15,47 @@
 #include <sys/time.h>
 
 #include "nest/bird.h"
-#include "sysdep/unix/io-loop.h"
 
 #include "lib/buffer.h"
+#include "lib/coro.h"
 #include "lib/lists.h"
 #include "lib/resource.h"
 #include "lib/event.h"
 #include "lib/timer.h"
 #include "lib/socket.h"
 
-
-struct birdloop
-{
-  pool *pool;
-  pthread_t thread;
-  pthread_mutex_t mutex;
-
-  u8 stop_called;
-  u8 poll_active;
-  u8 wakeup_masked;
-  int wakeup_fds[2];
-
-  struct timeloop time;
-  list event_list;
-  list sock_list;
-  uint sock_num;
-
-  BUFFER(sock *) poll_sk;
-  BUFFER(struct pollfd) poll_fd;
-  u8 poll_changed;
-  u8 close_scheduled;
-};
-
+#include "lib/io-loop.h"
+#include "sysdep/unix/io-loop.h"
+#include "conf/conf.h"
 
 /*
  *	Current thread context
  */
 
-static _Thread_local struct birdloop *birdloop_current;
+_Thread_local struct birdloop *birdloop_current;
+static _Thread_local struct birdloop *birdloop_wakeup_masked;
+static _Thread_local uint birdloop_wakeup_masked_count;
 
-static inline void
-birdloop_set_current(struct birdloop *loop)
+event_list *
+birdloop_event_list(struct birdloop *loop)
 {
-  birdloop_current = loop;
-  local_timeloop = loop ? &loop->time : &main_timeloop;
+  return &loop->event_list;
+}
+
+struct timeloop *
+birdloop_time_loop(struct birdloop *loop)
+{
+  return &loop->time;
+}
+
+_Bool
+birdloop_inside(struct birdloop *loop)
+{
+  for (struct birdloop *c = birdloop_current; c; c = c->prev_loop)
+    if (loop == c)
+      return 1;
+
+  return 0;
 }
 
 /*
@@ -135,57 +132,17 @@ wakeup_do_kick(struct birdloop *loop)
   pipe_kick(loop->wakeup_fds[1]);
 }
 
-static inline void
-wakeup_kick(struct birdloop *loop)
-{
-  if (!loop->wakeup_masked)
-    wakeup_do_kick(loop);
-  else
-    loop->wakeup_masked = 2;
-}
-
-/* For notifications from outside */
 void
-wakeup_kick_current(void)
+birdloop_ping(struct birdloop *loop)
 {
-  if (birdloop_current && birdloop_current->poll_active)
-    wakeup_kick(birdloop_current);
-}
-
-
-/*
- *	Events
- */
+  u32 ping_sent = atomic_fetch_add_explicit(&loop->ping_sent, 1, memory_order_acq_rel);
+  if (ping_sent)
+    return;
 
-static inline uint
-events_waiting(struct birdloop *loop)
-{
-  return !EMPTY_LIST(loop->event_list);
-}
-
-static inline void
-events_init(struct birdloop *loop)
-{
-  init_list(&loop->event_list);
-}
-
-static void
-events_fire(struct birdloop *loop)
-{
-  times_update();
-  ev_run_list(&loop->event_list);
-}
-
-void
-ev2_schedule(event *e)
-{
-  if (birdloop_current->poll_active && EMPTY_LIST(birdloop_current->event_list))
-    wakeup_kick(birdloop_current);
-
-  if (e->n.next)
-    rem_node(&e->n);
-
-  add_tail(&birdloop_current->event_list, &e->n);
+  if (loop == birdloop_wakeup_masked)
+    birdloop_wakeup_masked_count++;
+  else
+    wakeup_do_kick(loop);
 }
 
 
@@ -213,13 +170,13 @@ sockets_add(struct birdloop *loop, sock *s)
   s->index = -1;
   loop->poll_changed = 1;
 
-  if (loop->poll_active)
-    wakeup_kick(loop);
+  birdloop_ping(loop);
 }
 
 void
 sk_start(sock *s)
 {
+  ASSERT_DIE(birdloop_current != &main_birdloop);
   sockets_add(birdloop_current, s);
 }
 
@@ -230,28 +187,21 @@ sockets_remove(struct birdloop *loop, sock *s)
   loop->sock_num--;
 
   if (s->index >= 0)
+  {
     loop->poll_sk.data[s->index] = NULL;
-
-  s->index = -1;
-  loop->poll_changed = 1;
-
-  /* Wakeup moved to sk_stop() */
+    s->index = -1;
+    loop->poll_changed = 1;
+    loop->close_scheduled = 1;
+    birdloop_ping(loop);
+  }
+  else
+    close(s->fd);
 }
 
 void
 sk_stop(sock *s)
 {
   sockets_remove(birdloop_current, s);
-
-  if (birdloop_current->poll_active)
-  {
-    birdloop_current->close_scheduled = 1;
-    wakeup_kick(birdloop_current);
-  }
-  else
-    close(s->fd);
-
-  s->fd = -1;
 }
 
 static inline uint sk_want_events(sock *s)
@@ -351,12 +301,15 @@ sockets_fire(struct birdloop *loop)
 
     if (pfd->revents & POLLIN)
       while (e && *psk && (*psk)->rx_hook)
-	e = sk_read(*psk, 0);
+	e = sk_read(*psk, pfd->revents);
 
     e = 1;
     if (pfd->revents & POLLOUT)
+    {
+      loop->poll_changed = 1;
       while (e && *psk)
 	e = sk_write(*psk);
+    }
   }
 }
 
@@ -365,104 +318,168 @@ sockets_fire(struct birdloop *loop)
  *	Birdloop
  */
 
-static void *birdloop_main(void *arg);
+struct birdloop main_birdloop;
+
+static void birdloop_enter_locked(struct birdloop *loop);
+
+void
+birdloop_init(void)
+{
+  wakeup_init(&main_birdloop);
+
+  main_birdloop.time.domain = the_bird_domain.the_bird;
+  main_birdloop.time.loop = &main_birdloop;
+
+  times_update();
+  timers_init(&main_birdloop.time, &root_pool);
+
+  birdloop_enter_locked(&main_birdloop);
+}
+
+static void birdloop_main(void *arg);
 
 struct birdloop *
-birdloop_new(void)
+birdloop_new(pool *pp, uint order, const char *name)
 {
-  pool *p = rp_new(NULL, "Birdloop root");
+  struct domain_generic *dg = domain_new(name, order);
+
+  pool *p = rp_new(pp, name);
   struct birdloop *loop = mb_allocz(p, sizeof(struct birdloop));
   loop->pool = p;
-  pthread_mutex_init(&loop->mutex, NULL);
 
-  wakeup_init(loop);
+  loop->time.domain = dg;
+  loop->time.loop = loop;
+
+  birdloop_enter(loop);
 
-  events_init(loop);
+  wakeup_init(loop);
+  ev_init_list(&loop->event_list, loop, name);
   timers_init(&loop->time, p);
   sockets_init(loop);
 
+  loop->time.coro = coro_run(p, birdloop_main, loop);
+
+  birdloop_leave(loop);
+
   return loop;
 }
 
+static void
+birdloop_do_stop(struct birdloop *loop, void (*stopped)(void *data), void *data)
+{
+  loop->stopped = stopped;
+  loop->stop_data = data;
+  wakeup_do_kick(loop);
+}
+
 void
-birdloop_start(struct birdloop *loop)
+birdloop_stop(struct birdloop *loop, void (*stopped)(void *data), void *data)
 {
-  int rv = pthread_create(&loop->thread, NULL, birdloop_main, loop);
-  if (rv)
-    die("pthread_create(): %M", rv);
+  DG_LOCK(loop->time.domain);
+  birdloop_do_stop(loop, stopped, data);
+  DG_UNLOCK(loop->time.domain);
 }
 
 void
-birdloop_stop(struct birdloop *loop)
+birdloop_stop_self(struct birdloop *loop, void (*stopped)(void *data), void *data)
 {
-  pthread_mutex_lock(&loop->mutex);
-  loop->stop_called = 1;
-  wakeup_do_kick(loop);
-  pthread_mutex_unlock(&loop->mutex);
+  ASSERT_DIE(loop == birdloop_current);
+  ASSERT_DIE(DG_IS_LOCKED(loop->time.domain));
 
-  int rv = pthread_join(loop->thread, NULL);
-  if (rv)
-    die("pthread_join(): %M", rv);
+  birdloop_do_stop(loop, stopped, data);
 }
 
 void
 birdloop_free(struct birdloop *loop)
 {
+  ASSERT_DIE(loop->links == 0);
+  domain_free(loop->time.domain);
   rfree(loop->pool);
 }
 
+static void
+birdloop_enter_locked(struct birdloop *loop)
+{
+  ASSERT_DIE(DG_IS_LOCKED(loop->time.domain));
+  ASSERT_DIE(!birdloop_inside(loop));
+
+  /* Store the old context */
+  loop->prev_loop = birdloop_current;
+
+  /* Put the new context */
+  birdloop_current = loop;
+}
 
 void
 birdloop_enter(struct birdloop *loop)
 {
-  /* TODO: these functions could save and restore old context */
-  pthread_mutex_lock(&loop->mutex);
-  birdloop_set_current(loop);
+  DG_LOCK(loop->time.domain);
+  return birdloop_enter_locked(loop);
+}
+
+static void
+birdloop_leave_locked(struct birdloop *loop)
+{
+  /* Check the current context */
+  ASSERT_DIE(birdloop_current == loop);
+
+  /* Restore the old context */
+  birdloop_current = loop->prev_loop;
 }
 
 void
 birdloop_leave(struct birdloop *loop)
 {
-  /* TODO: these functions could save and restore old context */
-  birdloop_set_current(NULL);
-  pthread_mutex_unlock(&loop->mutex);
+  birdloop_leave_locked(loop);
+  DG_UNLOCK(loop->time.domain);
 }
 
 void
 birdloop_mask_wakeups(struct birdloop *loop)
 {
-  pthread_mutex_lock(&loop->mutex);
-  loop->wakeup_masked = 1;
-  pthread_mutex_unlock(&loop->mutex);
+  ASSERT_DIE(birdloop_wakeup_masked == NULL);
+  birdloop_wakeup_masked = loop;
 }
 
 void
 birdloop_unmask_wakeups(struct birdloop *loop)
 {
-  pthread_mutex_lock(&loop->mutex);
-  if (loop->wakeup_masked == 2)
+  ASSERT_DIE(birdloop_wakeup_masked == loop);
+  birdloop_wakeup_masked = NULL;
+  if (birdloop_wakeup_masked_count)
     wakeup_do_kick(loop);
-  loop->wakeup_masked = 0;
-  pthread_mutex_unlock(&loop->mutex);
+
+  birdloop_wakeup_masked_count = 0;
 }
 
-static void *
+void
+birdloop_link(struct birdloop *loop)
+{
+  ASSERT_DIE(birdloop_inside(loop));
+  loop->links++;
+}
+
+void
+birdloop_unlink(struct birdloop *loop)
+{
+  ASSERT_DIE(birdloop_inside(loop));
+  loop->links--;
+}
+
+static void
 birdloop_main(void *arg)
 {
   struct birdloop *loop = arg;
   timer *t;
   int rv, timeout;
 
-  birdloop_set_current(loop);
+  btime loop_begin = current_time();
 
-  pthread_mutex_lock(&loop->mutex);
+  birdloop_enter(loop);
   while (1)
   {
-    events_fire(loop);
-    timers_fire(&loop->time);
-
-    times_update();
-    if (events_waiting(loop))
+    timers_fire(&loop->time, 0);
+    if (ev_run_list(&loop->event_list))
       timeout = 0;
     else if (t = timers_first(&loop->time))
       timeout = (tm_remains(t) TO_MS) + 1;
@@ -472,8 +489,11 @@ birdloop_main(void *arg)
     if (loop->poll_changed)
       sockets_prepare(loop);
 
-    loop->poll_active = 1;
-    pthread_mutex_unlock(&loop->mutex);
+    btime duration = current_time() - loop_begin;
+    if (duration > config->watchdog_warning)
+      log(L_WARN "I/O loop cycle took %d ms", (int) (duration TO_MS));
+
+    birdloop_leave(loop);
 
   try:
     rv = poll(loop->poll_fd.data, loop->poll_fd.used, timeout);
@@ -484,25 +504,32 @@ birdloop_main(void *arg)
       die("poll: %m");
     }
 
-    pthread_mutex_lock(&loop->mutex);
-    loop->poll_active = 0;
+    birdloop_enter(loop);
 
     if (loop->close_scheduled)
       sockets_close_fds(loop);
 
-    if (loop->stop_called)
+    if (loop->stopped)
       break;
 
+    loop_begin = current_time();
+
     if (rv)
       sockets_fire(loop);
 
-    timers_fire(&loop->time);
+    atomic_exchange_explicit(&loop->ping_sent, 0, memory_order_acq_rel);
   }
 
-  loop->stop_called = 0;
-  pthread_mutex_unlock(&loop->mutex);
+  /* Flush remaining events */
+  ASSERT_DIE(!ev_run_list(&loop->event_list));
+
+  /* No timers allowed */
+  ASSERT_DIE(timers_count(&loop->time) == 0);
+  ASSERT_DIE(EMPTY_LIST(loop->sock_list));
+  ASSERT_DIE(loop->sock_num == 0);
 
-  return NULL;
+  birdloop_leave(loop);
+  loop->stopped(loop->stop_data);
 }
 
 
diff --git a/sysdep/unix/io-loop.h b/sysdep/unix/io-loop.h
index d858b04e..4024b6c5 100644
--- a/sysdep/unix/io-loop.h
+++ b/sysdep/unix/io-loop.h
@@ -4,31 +4,32 @@
  *	Can be freely distributed and used under the terms of the GNU GPL.
  */
 
-#ifndef _BIRD_IO_LOOP_H_
-#define _BIRD_IO_LOOP_H_
+#ifndef _BIRD_SYSDEP_UNIX_IO_LOOP_H_
+#define _BIRD_SYSDEP_UNIX_IO_LOOP_H_
 
-#include "nest/bird.h"
-#include "lib/lists.h"
-#include "lib/resource.h"
-#include "lib/event.h"
-#include "lib/timer.h"
-#include "lib/socket.h"
+struct birdloop
+{
+  pool *pool;
 
+  struct timeloop time;
+  event_list event_list;
+  list sock_list;
+  uint sock_num;
 
-void ev2_schedule(event *e);
+  BUFFER(sock *) poll_sk;
+  BUFFER(struct pollfd) poll_fd;
+  u8 poll_changed;
+  u8 close_scheduled;
 
-void sk_start(sock *s);
-void sk_stop(sock *s);
+  _Atomic u32 ping_sent;
+  int wakeup_fds[2];
 
-struct birdloop *birdloop_new(void);
-void birdloop_start(struct birdloop *loop);
-void birdloop_stop(struct birdloop *loop);
-void birdloop_free(struct birdloop *loop);
+  uint links;
 
-void birdloop_enter(struct birdloop *loop);
-void birdloop_leave(struct birdloop *loop);
-void birdloop_mask_wakeups(struct birdloop *loop);
-void birdloop_unmask_wakeups(struct birdloop *loop);
+  void (*stopped)(void *data);
+  void *stop_data;
 
+  struct birdloop *prev_loop;
+};
 
-#endif /* _BIRD_IO_LOOP_H_ */
+#endif
diff --git a/sysdep/unix/io.c b/sysdep/unix/io.c
index 90bb5d64..c91f2856 100644
--- a/sysdep/unix/io.c
+++ b/sysdep/unix/io.c
@@ -43,6 +43,7 @@
 #include "conf/conf.h"
 
 #include "sysdep/unix/unix.h"
+#include "sysdep/unix/io-loop.h"
 #include CONFIG_INCLUDE_SYSIO_H
 
 /* Maximum number of calls of tx handler for one socket in one
@@ -800,18 +801,16 @@ sk_free(resource *r)
     sk_ssh_free(s);
 #endif
 
-  if (s->fd < 0)
+  if ((s->fd < 0) || (s->flags & SKF_THREAD))
     return;
 
-  /* FIXME: we should call sk_stop() for SKF_THREAD sockets */
-  if (!(s->flags & SKF_THREAD))
-  {
-    if (s == current_sock)
-      current_sock = sk_next(s);
-    if (s == stored_sock)
-      stored_sock = sk_next(s);
+  if (s == current_sock)
+    current_sock = sk_next(s);
+  if (s == stored_sock)
+    stored_sock = sk_next(s);
+
+  if (enlisted(&s->n))
     rem_node(&s->n);
-  }
 
   if (s->type != SK_SSH && s->type != SK_SSH_ACTIVE)
     close(s->fd);
@@ -1104,7 +1103,11 @@ sk_passive_connected(sock *s, int type)
     return 1;
   }
 
-  sk_insert(t);
+  if (s->flags & SKF_PASSIVE_THREAD)
+    t->flags |= SKF_THREAD;
+  else
+    sk_insert(t);
+
   sk_alloc_bufs(t);
   s->rx_hook(t, 0);
   return 1;
@@ -1508,6 +1511,36 @@ sk_open_unix(sock *s, char *name)
   return 0;
 }
 
+static void
+sk_reloop_hook(void *_vs)
+{
+  sock *s = _vs;
+  if (birdloop_inside(&main_birdloop))
+  {
+    s->flags &= ~SKF_THREAD;
+    sk_insert(s);
+  }
+  else
+  {
+    s->flags |= SKF_THREAD;
+    sk_start(s);
+  }
+}
+
+void
+sk_reloop(sock *s, struct birdloop *loop)
+{
+  if (enlisted(&s->n))
+    rem_node(&s->n);
+
+  s->reloop = (event) {
+    .hook = sk_reloop_hook,
+    .data = s,
+  };
+
+  ev_send_loop(loop, &s->reloop);
+}
+
 
 #define CMSG_RX_SPACE MAX(CMSG4_SPACE_PKTINFO+CMSG4_SPACE_TTL, \
 			  CMSG6_SPACE_PKTINFO+CMSG6_SPACE_TTL)
@@ -2143,8 +2176,9 @@ void
 io_init(void)
 {
   init_list(&sock_list);
-  init_list(&global_event_list);
-  init_list(&global_work_list);
+  ev_init_list(&global_event_list, &main_birdloop, "Global event list");
+  ev_init_list(&global_work_list, &main_birdloop, "Global work list");
+  ev_init_list(&main_birdloop.event_list, &main_birdloop, "Global fast event list");
   krt_io_init();
   // XXX init_times();
   // XXX update_times();
@@ -2158,14 +2192,7 @@ static int short_loops = 0;
 #define SHORT_LOOP_MAX 10
 #define WORK_EVENTS_MAX 10
 
-static int poll_reload_pipe[2];
-
-void
-io_loop_reload(void)
-{
-  char b;
-  write(poll_reload_pipe[1], &b, 1);
-}
+void pipe_drain(int fd);
 
 void
 io_loop(void)
@@ -2178,21 +2205,19 @@ io_loop(void)
   int fdmax = 256;
   struct pollfd *pfd = xmalloc(fdmax * sizeof(struct pollfd));
 
-  if (pipe(poll_reload_pipe) < 0)
-    die("pipe(poll_reload_pipe) failed: %m");
-
   watchdog_start1();
   for(;;)
     {
       times_update();
       events = ev_run_list(&global_event_list);
       events = ev_run_list_limited(&global_work_list, WORK_EVENTS_MAX) || events;
-      timers_fire(&main_timeloop);
+      events = ev_run_list(&main_birdloop.event_list) || events;
+      timers_fire(&main_birdloop.time, 1);
       io_close_event();
 
       // FIXME
       poll_tout = (events ? 0 : 3000); /* Time in milliseconds */
-      if (t = timers_first(&main_timeloop))
+      if (t = timers_first(&main_birdloop.time))
       {
 	times_update();
 	timeout = (tm_remains(t) TO_MS) + 1;
@@ -2200,7 +2225,7 @@ io_loop(void)
       }
 
       /* A hack to reload main io_loop() when something has changed asynchronously. */
-      pfd[0].fd = poll_reload_pipe[0];
+      pfd[0].fd = main_birdloop.wakeup_fds[0];
       pfd[0].events = POLLIN;
 
       nfds = 1;
@@ -2263,9 +2288,9 @@ io_loop(void)
 
       /* And finally enter poll() to find active sockets */
       watchdog_stop();
-      the_bird_unlock();
+      birdloop_leave(&main_birdloop);
       pout = poll(pfd, nfds, poll_tout);
-      the_bird_lock();
+      birdloop_enter(&main_birdloop);
       watchdog_start();
 
       if (pout < 0)
@@ -2279,8 +2304,8 @@ io_loop(void)
 	  if (pfd[0].revents & POLLIN)
 	  {
 	    /* IO loop reload requested */
-	    char b;
-	    read(poll_reload_pipe[0], &b, 1);
+	    pipe_drain(main_birdloop.wakeup_fds[0]);
+	    atomic_exchange_explicit(&main_birdloop.ping_sent, 0, memory_order_acq_rel);
 	    continue;
 	  }
 
diff --git a/sysdep/unix/main.c b/sysdep/unix/main.c
index d35424ff..5da27cb6 100644
--- a/sysdep/unix/main.c
+++ b/sysdep/unix/main.c
@@ -908,10 +908,12 @@ main(int argc, char **argv)
   parse_args(argc, argv);
   log_switch(1, NULL, NULL);
 
+  the_bird_lock();
+
   random_init();
   net_init();
   resource_init();
-  timer_init();
+  birdloop_init();
   olock_init();
   io_init();
   rt_init();
@@ -961,7 +963,6 @@ main(int argc, char **argv)
       dup2(0, 2);
     }
 
-  the_bird_lock();
 
   main_thread_init();
 
diff --git a/sysdep/unix/unix.h b/sysdep/unix/unix.h
index 313c97c3..ad85d1ea 100644
--- a/sysdep/unix/unix.h
+++ b/sysdep/unix/unix.h
@@ -106,7 +106,6 @@ extern volatile sig_atomic_t async_shutdown_flag;
 
 void io_init(void);
 void io_loop(void);
-void io_loop_reload(void);
 void io_log_dump(void);
 int sk_open_unix(struct birdsock *s, char *name);
 struct rfile *rf_open(struct pool *, const char *name, const char *mode);