Converting the former BFD loop to a universal IO loop and protocol loop.

There is a simple universal IO loop, taking care of events, timers and
sockets. Primarily, one instance of a protocol should use exactly one IO
loop to do all its work, as is now done in BFD.

Contrary to previous versions, the loop is now launched and cleaned by
the nest/proto.c code, allowing for a protocol to just request its own
loop by setting the loop's lock order in config higher than the_bird.

It is not supported nor checked if any protocol changed the requested
lock order in reconfigure. No protocol should do it at all.

7 files changed, 143 insertions, 66 deletions

diff --git a/nest/a-path_test.c b/nest/a-path_test.c
index 9ed0a786..2e6e4956 100644
--- a/nest/a-path_test.c
+++ b/nest/a-path_test.c
@@ -204,10 +204,13 @@ t_as_path_converting(void)
 }
 #endif
 
+void resource_sys_init(void);
+
 int
 main(int argc, char *argv[])
 {
   bt_init(argc, argv);
+  resource_sys_init();
 
   bt_test_suite(t_as_path_match, "Testing AS path matching and some a-path utilities.");
   bt_test_suite(t_path_format, "Testing formating as path into byte buffer");
diff --git a/nest/a-set_test.c b/nest/a-set_test.c
index 96b6a727..efd1b67d 100644
--- a/nest/a-set_test.c
+++ b/nest/a-set_test.c
@@ -240,10 +240,14 @@ t_set_ec_delete(void)
   return 1;
 }
 
+
+void resource_sys_init(void);
+
 int
 main(int argc, char *argv[])
 {
   bt_init(argc, argv);
+  resource_sys_init();
 
   bt_test_suite(t_set_int_contains, "Testing sets of integers: contains, get_data");
   bt_test_suite(t_set_int_format,   "Testing sets of integers: format");
diff --git a/nest/config.Y b/nest/config.Y
index 92f1aad2..6e7689ed 100644
--- a/nest/config.Y
+++ b/nest/config.Y
@@ -434,6 +434,7 @@ proto: dev_proto '}' ;
 dev_proto_start: proto_start DIRECT {
      this_proto = proto_config_new(&proto_device, $1);
      init_list(&DIRECT_CFG->iface_list);
+     this_proto->late_if_feed = 1;
    }
  ;
 
diff --git a/nest/proto.c b/nest/proto.c
index fae0647a..ac0fb232 100644
--- a/nest/proto.c
+++ b/nest/proto.c
@@ -15,6 +15,7 @@
 #include "lib/event.h"
 #include "lib/timer.h"
 #include "lib/string.h"
+#include "lib/coro.h"
 #include "conf/conf.h"
 #include "nest/route.h"
 #include "nest/iface.h"
@@ -58,7 +59,28 @@ static void channel_feed_end(struct channel *c);
 static void channel_export_stopped(struct rt_export_request *req);
 
 static inline int proto_is_done(struct proto *p)
-{ return (p->proto_state == PS_DOWN) && (p->active_channels == 0); }
+{ return (p->proto_state == PS_DOWN) && proto_is_inactive(p); }
+
+static inline event_list *proto_event_list(struct proto *p)
+{ return p->loop == &main_birdloop ? &global_event_list : birdloop_event_list(p->loop); }
+
+static inline event_list *proto_work_list(struct proto *p)
+{ return p->loop == &main_birdloop ? &global_work_list : birdloop_event_list(p->loop); }
+
+static inline void proto_send_event(struct proto *p)
+{ ev_send(proto_event_list(p), p->event); }
+
+#define PROTO_ENTER_FROM_MAIN(p)    ({ \
+    ASSERT_DIE(birdloop_inside(&main_birdloop)); \
+    struct birdloop *_loop = (p)->loop; \
+    if (_loop != &main_birdloop) birdloop_enter(_loop); \
+    _loop; \
+    })
+
+#define PROTO_LEAVE_FROM_MAIN(loop) ({ if (loop != &main_birdloop) birdloop_leave(loop); })
+
+#define PROTO_LOCKED_FROM_MAIN(p)	for (struct birdloop *_proto_loop = PROTO_ENTER_FROM_MAIN(p); _proto_loop; PROTO_LEAVE_FROM_MAIN(_proto_loop), (_proto_loop = NULL))
+
 
 static inline int channel_is_active(struct channel *c)
 { return (c->channel_state != CS_DOWN); }
@@ -473,6 +495,7 @@ channel_start_export(struct channel *c)
 
   c->out_req = (struct rt_export_request) {
     .name = rn,
+    .list = proto_work_list(c->proto),
     .trace_routes = c->debug | c->proto->debug,
     .dump_req = channel_dump_export_req,
     .log_state_change = channel_export_log_state_change,
@@ -517,7 +540,7 @@ channel_check_stopped(struct channel *c)
 	return;
 
       channel_set_state(c, CS_DOWN);
-      ev_schedule(c->proto->event);
+      proto_send_event(c->proto);
 
       break;
     case CS_PAUSE:
@@ -853,6 +876,7 @@ channel_setup_in_table(struct channel *c, int best)
   };
   c->in_table->get = (struct rt_export_request) {
     .name = cat->name,
+    .list = proto_work_list(c->proto),
     .trace_routes = c->debug | c->proto->debug,
     .dump_req = channel_in_get_dump_req,
     .log_state_change = channel_get_log_state_change,
@@ -895,6 +919,7 @@ channel_setup_out_table(struct channel *c)
   };
   c->out_table->get = (struct rt_export_request) {
     .name = cat->name,
+    .list = proto_work_list(c->proto),
     .trace_routes = c->debug | c->proto->debug,
     .dump_req = channel_out_get_dump_req,
     .log_state_change = channel_get_log_state_change,
@@ -997,7 +1022,7 @@ channel_do_down(struct channel *c)
 
   /* Schedule protocol shutddown */
   if (proto_is_done(c->proto))
-    ev_schedule(c->proto->event);
+    proto_send_event(c->proto);
 }
 
 void
@@ -1085,9 +1110,12 @@ channel_request_table_feeding(struct channel *c)
 void
 channel_request_feeding(struct channel *c)
 {
+  if (c->gr_wait || !c->proto->rt_notify)
+    return;
+
   CD(c, "Refeed requested");
 
-  ASSERT(c->out_req.hook);
+  ASSERT_DIE(c->out_req.hook);
 
   if (c->out_table)
     channel_aux_request_refeed(c->out_table);
@@ -1331,17 +1359,35 @@ proto_configure_channel(struct proto *p, struct channel **pc, struct channel_con
   return 1;
 }
 
+static void
+proto_cleanup(struct proto *p)
+{
+  rfree(p->pool);
+  p->pool = NULL;
+
+  p->active = 0;
+  proto_log_state_change(p);
+  proto_rethink_goal(p);
+}
 
 static void
-proto_event(void *ptr)
+proto_loop_stopped(void *ptr)
 {
   struct proto *p = ptr;
 
-  if (p->do_start)
-  {
-    if_feed_baby(p);
-    p->do_start = 0;
-  }
+  birdloop_enter(&main_birdloop);
+
+  p->loop = &main_birdloop;
+  p->event->list = NULL;
+  proto_cleanup(p);
+
+  birdloop_leave(&main_birdloop);
+}
+
+static void
+proto_event(void *ptr)
+{
+  struct proto *p = ptr;
 
   if (p->do_stop)
   {
@@ -1351,14 +1397,10 @@ proto_event(void *ptr)
   }
 
   if (proto_is_done(p))
-  {
-    rfree(p->pool);
-    p->pool = NULL;
-
-    p->active = 0;
-    proto_log_state_change(p);
-    proto_rethink_goal(p);
-  }
+    if (p->loop != &main_birdloop)
+      birdloop_stop_self(p->loop, proto_loop_stopped, p);
+    else
+      proto_cleanup(p);
 }
 
 
@@ -1399,6 +1441,7 @@ proto_init(struct proto_config *c, node *n)
   struct protocol *pr = c->protocol;
   struct proto *p = pr->init(c);
 
+  p->loop = &main_birdloop;
   p->proto_state = PS_DOWN;
   p->last_state_change = current_time();
   p->vrf = c->vrf;
@@ -1415,11 +1458,27 @@ proto_init(struct proto_config *c, node *n)
 static void
 proto_start(struct proto *p)
 {
-  /* Here we cannot use p->cf->name since it won't survive reconfiguration */
-  p->pool = rp_new(proto_pool, p->proto->name);
+  DBG("Kicking %s up\n", p->name);
+  PD(p, "Starting");
+
+  int ns = strlen("Protocol ") + strlen(p->cf->name) + 1;
+  void *nb = mb_alloc(proto_pool, ns);
+  ASSERT_DIE(ns - 1 == bsnprintf(nb, ns, "Protocol %s", p->cf->name));
+
+  p->pool = rp_new(proto_pool, nb);
 
   if (graceful_restart_state == GRS_INIT)
     p->gr_recovery = 1;
+
+  if (p->cf->loop_order != DOMAIN_ORDER(the_bird))
+    p->loop = birdloop_new(p->pool, p->cf->loop_order, nb);
+
+  p->event->list = proto_event_list(p);
+
+  mb_move(nb, p->pool);
+
+  PROTO_LOCKED_FROM_MAIN(p)
+    proto_notify_state(p, (p->proto->start ? p->proto->start(p) : PS_UP));
 }
 
 
@@ -1455,6 +1514,7 @@ proto_config_new(struct protocol *pr, int class)
   cf->class = class;
   cf->debug = new_config->proto_default_debug;
   cf->mrtdump = new_config->proto_default_mrtdump;
+  cf->loop_order = DOMAIN_ORDER(the_bird);
 
   init_list(&cf->channels);
 
@@ -1744,11 +1804,20 @@ protos_commit(struct config *new, struct config *old, int force_reconfig, int ty
 }
 
 static void
-proto_rethink_goal(struct proto *p)
+proto_shutdown(struct proto *p)
 {
-  struct protocol *q;
-  byte goal;
+  if (p->proto_state == PS_START || p->proto_state == PS_UP)
+  {
+    /* Going down */
+    DBG("Kicking %s down\n", p->name);
+    PD(p, "Shutting down");
+    proto_notify_state(p, (p->proto->shutdown ? p->proto->shutdown(p) : PS_DOWN));
+  }
+}
 
+static void
+proto_rethink_goal(struct proto *p)
+{
   if (p->reconfiguring && !p->active)
   {
     struct proto_config *nc = p->cf_new;
@@ -1768,32 +1837,12 @@ proto_rethink_goal(struct proto *p)
 
   /* Determine what state we want to reach */
   if (p->disabled || p->reconfiguring)
-    goal = PS_DOWN;
-  else
-    goal = PS_UP;
-
-  q = p->proto;
-  if (goal == PS_UP)
-  {
-    if (!p->active)
-    {
-      /* Going up */
-      DBG("Kicking %s up\n", p->name);
-      PD(p, "Starting");
-      proto_start(p);
-      proto_notify_state(p, (q->start ? q->start(p) : PS_UP));
-    }
-  }
-  else
   {
-    if (p->proto_state == PS_START || p->proto_state == PS_UP)
-    {
-      /* Going down */
-      DBG("Kicking %s down\n", p->name);
-      PD(p, "Shutting down");
-      proto_notify_state(p, (q->shutdown ? q->shutdown(p) : PS_DOWN));
-    }
+    PROTO_LOCKED_FROM_MAIN(p)
+      proto_shutdown(p);
   }
+  else if (!p->active)
+    proto_start(p);
 }
 
 struct proto *
@@ -1998,7 +2047,7 @@ protos_dump_all(void)
 #define DPF(x)	(p->x ? " " #x : "")
     debug("  protocol %s (%p) state %s with %d active channels flags: %s%s%s%s\n",
 	p->name, p, p_states[p->proto_state], p->active_channels,
-	DPF(disabled), DPF(active), DPF(do_start), DPF(do_stop), DPF(reconfiguring));
+	DPF(disabled), DPF(active), DPF(do_stop), DPF(reconfiguring));
 #undef DPF
 
     struct channel *c;
@@ -2286,8 +2335,8 @@ static inline void
 proto_do_start(struct proto *p)
 {
   p->active = 1;
-  p->do_start = 1;
-  ev_schedule(p->event);
+  if (!p->cf->late_if_feed)
+    if_feed_baby(p);
 }
 
 static void
@@ -2300,6 +2349,9 @@ proto_do_up(struct proto *p)
   }
 
   proto_start_channels(p);
+
+  if (p->cf->late_if_feed)
+    if_feed_baby(p);
 }
 
 static inline void
@@ -2314,9 +2366,6 @@ proto_do_stop(struct proto *p)
   p->down_sched = 0;
   p->gr_recovery = 0;
 
-  p->do_stop = 1;
-  ev_schedule(p->event);
-
   if (p->main_source)
   {
     rt_unlock_source(p->main_source);
@@ -2324,6 +2373,9 @@ proto_do_stop(struct proto *p)
   }
 
   proto_stop_channels(p);
+
+  p->do_stop = 1;
+  proto_send_event(p);
 }
 
 static void
@@ -2334,7 +2386,7 @@ proto_do_down(struct proto *p)
 
   /* Shutdown is finished in the protocol event */
   if (proto_is_done(p))
-    ev_schedule(p->event);
+    proto_send_event(p);
 }
 
 
@@ -2573,7 +2625,7 @@ proto_cmd_disable(struct proto *p, uintptr_t arg, int cnt UNUSED)
   p->disabled = 1;
   p->down_code = PDC_CMD_DISABLE;
   proto_set_message(p, (char *) arg, -1);
-  proto_rethink_goal(p);
+  proto_shutdown(p);
   cli_msg(-9, "%s: disabled", p->name);
 }
 
@@ -2606,9 +2658,9 @@ proto_cmd_restart(struct proto *p, uintptr_t arg, int cnt UNUSED)
   p->disabled = 1;
   p->down_code = PDC_CMD_RESTART;
   proto_set_message(p, (char *) arg, -1);
-  proto_rethink_goal(p);
+  proto_shutdown(p);
   p->disabled = 0;
-  proto_rethink_goal(p);
+  /* After the protocol shuts down, proto_rethink_goal() is run from proto_event. */
   cli_msg(-12, "%s: restarted", p->name);
 }
 
@@ -2683,7 +2735,9 @@ proto_apply_cmd_symbol(const struct symbol *s, void (* cmd)(struct proto *, uint
 
   if (s->proto->proto)
   {
-    cmd(s->proto->proto, arg, 0);
+    struct proto *p = s->proto->proto;
+    PROTO_LOCKED_FROM_MAIN(p)
+      cmd(p, arg, 0);
     cli_msg(0, "");
   }
   else
@@ -2698,7 +2752,8 @@ proto_apply_cmd_patt(const char *patt, void (* cmd)(struct proto *, uintptr_t, i
 
   WALK_LIST(p, proto_list)
     if (!patt || patmatch(patt, p->name))
-      cmd(p, arg, cnt++);
+      PROTO_LOCKED_FROM_MAIN(p)
+	cmd(p, arg, cnt++);
 
   if (!cnt)
     cli_msg(8003, "No protocols match");
diff --git a/nest/protocol.h b/nest/protocol.h
index f9996b18..2fbd607c 100644
--- a/nest/protocol.h
+++ b/nest/protocol.h
@@ -120,8 +120,10 @@ struct proto_config {
   u8 net_type;				/* Protocol network type (NET_*), 0 for undefined */
   u8 disabled;				/* Protocol enabled/disabled by default */
   u8 vrf_set;				/* Related VRF instance (below) is defined */
+  u8 late_if_feed;			/* Delay interface feed after channels are up */
   u32 debug, mrtdump;			/* Debugging bitfields, both use D_* constants */
   u32 router_id;			/* Protocol specific router ID */
+  uint loop_order;			/* Launch a birdloop on this locking level; use DOMAIN_ORDER(the_bird) for mainloop */
 
   list channels;			/* List of channel configs (struct channel_config) */
   struct iface *vrf;			/* Related VRF instance, NULL if global */
@@ -139,6 +141,7 @@ struct proto {
   struct proto_config *cf_new;		/* Configuration we want to switch to after shutdown (NULL=delete) */
   pool *pool;				/* Pool containing local objects */
   event *event;				/* Protocol event */
+  struct birdloop *loop;		/* BIRDloop running this protocol */
 
   list channels;			/* List of channels to rtables (struct channel) */
   struct channel *main_channel;		/* Primary channel */
@@ -149,12 +152,12 @@ struct proto {
   u32 debug;				/* Debugging flags */
   u32 mrtdump;				/* MRTDump flags */
   uint active_channels;			/* Number of active channels */
+  uint active_coroutines;		/* Number of active coroutines */
   byte net_type;			/* Protocol network type (NET_*), 0 for undefined */
   byte disabled;			/* Manually disabled */
   byte vrf_set;				/* Related VRF instance (above) is defined */
   byte proto_state;			/* Protocol state machine (PS_*, see below) */
   byte active;				/* From PS_START to cleanup after PS_STOP */
-  byte do_start;			/* Start actions are scheduled */
   byte do_stop;				/* Stop actions are scheduled */
   byte reconfiguring;			/* We're shutting down due to reconfiguration */
   byte gr_recovery;			/* Protocol should participate in graceful restart recovery */
@@ -356,6 +359,8 @@ void proto_notify_state(struct proto *p, unsigned state);
  *	as a result of received ROUTE-REFRESH request).
  */
 
+static inline int proto_is_inactive(struct proto *p)
+{ return (p->active_channels == 0) && (p->active_coroutines == 0); }
 
 
 /*
diff --git a/nest/route.h b/nest/route.h
index d6474e09..7c5fd02f 100644
--- a/nest/route.h
+++ b/nest/route.h
@@ -11,6 +11,7 @@
 #define _BIRD_ROUTE_H_
 
 #include "lib/lists.h"
+#include "lib/event.h"
 #include "lib/bitmap.h"
 #include "lib/resource.h"
 #include "lib/net.h"
@@ -329,6 +330,8 @@ struct rt_export_request {
   char *name;
   u8 trace_routes;
 
+  event_list *list;			/* Where to schedule export events */
+
   /* There are two methods of export. You can either request feeding every single change
    * or feeding the whole route feed. In case of regular export, &export_one is preferred.
    * Anyway, when feeding, &export_bulk is preferred, falling back to &export_one.
diff --git a/nest/rt-table.c b/nest/rt-table.c
index c049101a..9c12ef56 100644
--- a/nest/rt-table.c
+++ b/nest/rt-table.c
@@ -1105,6 +1105,12 @@ rt_next_export(struct rt_export_hook *hook, rtable *tab)
     return tab->first_export;
 }
 
+static inline void
+rt_send_export_event(struct rt_export_hook *hook)
+{
+  ev_send(hook->req->list, hook->event);
+}
+
 static void
 rt_announce_exports(timer *tm)
 {
@@ -1116,7 +1122,7 @@ rt_announce_exports(timer *tm)
     if (atomic_load_explicit(&c->export_state, memory_order_acquire) != TES_READY)
       continue;
 
-    ev_schedule_work(c->event);
+    rt_send_export_event(c);
   }
 }
 
@@ -1169,7 +1175,7 @@ rt_export_hook(void *_data)
     rte_update_unlock();
   }
 
-  ev_schedule_work(c->event);
+  rt_send_export_event(c);
 }
 
 
@@ -1732,7 +1738,7 @@ rt_request_export(rtable *tab, struct rt_export_request *req)
   DBG("New export hook %p req %p in table %s uc=%u\n", hook, req, tab->name, tab->use_count);
 
   hook->event = ev_new_init(p, rt_feed_channel, hook);
-  ev_schedule_work(hook->event);
+  rt_send_export_event(hook);
 
   rt_set_export_state(hook, TES_FEEDING);
 }
@@ -1754,7 +1760,7 @@ rt_stop_export(struct rt_export_request *req, void (*stopped)(struct rt_export_r
   hook->event->hook = rt_export_stopped;
   hook->stopped = stopped;
 
-  ev_schedule(hook->event);
+  rt_send_export_event(hook);
 
   rt_set_export_state(hook, TES_STOP);
 }
@@ -2869,7 +2875,7 @@ rt_feed_channel(void *data)
       if (max_feed <= 0)
 	{
 	  FIB_ITERATE_PUT(fit);
-	  ev_schedule_work(c->event);
+	  rt_send_export_event(c);
 	  return;
 	}
 
@@ -2904,7 +2910,7 @@ rt_feed_channel(void *data)
   FIB_ITERATE_END;
 
   c->event->hook = rt_export_hook;
-  ev_schedule_work(c->event);
+  rt_send_export_event(c);
 
   rt_set_export_state(c, TES_READY);
 }