1 files changed, 210 insertions, 105 deletions

diff --git a/nest/proto.c b/nest/proto.c
index 061205c1..319b35dd 100644
--- a/nest/proto.c
+++ b/nest/proto.c
@@ -55,9 +55,31 @@ static void channel_update_limit(struct channel *c, struct limit *l, int dir, st
 static void channel_reset_limit(struct channel *c, struct limit *l, int dir);
 static void channel_feed_end(struct channel *c);
 static void channel_export_stopped(struct rt_export_request *req);
+static void channel_check_stopped(struct channel *c);
 
 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); }
@@ -147,7 +169,7 @@ proto_cf_find_channel(struct proto_config *pc, uint net_type)
  * Returns pointer to channel or NULL
  */
 struct channel *
-proto_find_channel_by_table(struct proto *p, struct rtable *t)
+proto_find_channel_by_table(struct proto *p, rtable *t)
 {
   struct channel *c;
 
@@ -291,10 +313,18 @@ proto_remove_channels(struct proto *p)
     proto_remove_channel(p, c);
 }
 
+struct roa_subscription {
+  node roa_node;
+  struct settle settle;
+  struct channel *c;
+  struct rt_export_request req;
+};
+
 static void
-channel_roa_in_changed(struct rt_subscription *s)
+channel_roa_in_changed(struct settle *se)
 {
-  struct channel *c = s->data;
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, settle, se);
+  struct channel *c = s->c;
   int active = !!c->reload_req.hook;
 
   CD(c, "Reload triggered by RPKI change%s", active ? " - already active" : "");
@@ -306,9 +336,11 @@ channel_roa_in_changed(struct rt_subscription *s)
 }
 
 static void
-channel_roa_out_changed(struct rt_subscription *s)
+channel_roa_out_changed(struct settle *se)
 {
-  struct channel *c = s->data;
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, settle, se);
+  struct channel *c = s->c;
+
   CD(c, "Feeding triggered by RPKI change");
 
   c->refeed_pending = 1;
@@ -317,29 +349,47 @@ channel_roa_out_changed(struct rt_subscription *s)
     rt_stop_export(&c->out_req, channel_export_stopped);
 }
 
-/* Temporary code, subscriptions should be changed to resources */
-struct roa_subscription {
-  struct rt_subscription s;
-  node roa_node;
-};
+static void
+channel_export_one_roa(struct rt_export_request *req, const net_addr *net UNUSED, struct rt_pending_export *first)
+{
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, req, req);
+
+  /* TODO: use the information about what roa has changed */
+  settle_kick(&s->settle, &main_birdloop);
+
+  rpe_mark_seen_all(req->hook, first, NULL);
+}
+
+static void
+channel_dump_roa_req(struct rt_export_request *req)
+{
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, req, req);
+  struct channel *c = s->c;
+  struct rtable_private *tab = SKIP_BACK(struct rtable_private, exporter.e, req->hook->table);
+
+  debug("  Channel %s.%s ROA %s change notifier from table %s request %p\n",
+      c->proto->name, c->name,
+      (s->settle.hook == channel_roa_in_changed) ? "import" : "export",
+      tab->name, req);
+}
 
 static int
 channel_roa_is_subscribed(struct channel *c, rtable *tab, int dir)
 {
-  void (*hook)(struct rt_subscription *) =
+  void (*hook)(struct settle *) =
     dir ? channel_roa_in_changed : channel_roa_out_changed;
 
   struct roa_subscription *s;
   node *n;
 
   WALK_LIST2(s, n, c->roa_subscriptions, roa_node)
-    if ((s->s.tab == tab) && (s->s.hook == hook))
+    if ((tab == SKIP_BACK(rtable, priv.exporter.e, s->req.hook->table))
+	  && (s->settle.hook == hook))
       return 1;
 
   return 0;
 }
 
-
 static void
 channel_roa_subscribe(struct channel *c, rtable *tab, int dir)
 {
@@ -348,26 +398,46 @@ channel_roa_subscribe(struct channel *c, rtable *tab, int dir)
 
   struct roa_subscription *s = mb_allocz(c->proto->pool, sizeof(struct roa_subscription));
 
-  s->s.hook = dir ? channel_roa_in_changed : channel_roa_out_changed;
-  s->s.data = c;
-  rt_subscribe(tab, &s->s);
+  *s = (struct roa_subscription) {
+    .settle = SETTLE_INIT(&c->roa_settle, dir ? channel_roa_in_changed : channel_roa_out_changed, NULL),
+    .c = c,
+    .req = {
+      .name = mb_sprintf(c->proto->pool, "%s.%s.roa-%s.%s",
+	  c->proto->name, c->name, dir ? "in" : "out", tab->name),
+      .list = proto_work_list(c->proto),
+      .trace_routes = c->debug | c->proto->debug,
+      .dump_req = channel_dump_roa_req,
+      .export_one = channel_export_one_roa,
+    },
+  };
 
   add_tail(&c->roa_subscriptions, &s->roa_node);
+  rt_request_export(tab, &s->req);
 }
 
 static void
-channel_roa_unsubscribe(struct roa_subscription *s)
+channel_roa_unsubscribed(struct rt_export_request *req)
 {
-  rt_unsubscribe(&s->s);
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, req, req);
+  struct channel *c = s->c;
+
   rem_node(&s->roa_node);
   mb_free(s);
+  
+  channel_check_stopped(c);
+}
+
+static void
+channel_roa_unsubscribe(struct roa_subscription *s)
+{
+  rt_stop_export(&s->req, channel_roa_unsubscribed);
 }
 
 static void
 channel_roa_subscribe_filter(struct channel *c, int dir)
 {
   const struct filter *f = dir ? c->in_filter : c->out_filter;
-  struct rtable *tab;
+  rtable *tab;
   int valid = 1, found = 0;
 
   if ((f == FILTER_ACCEPT) || (f == FILTER_REJECT))
@@ -428,17 +498,13 @@ channel_start_import(struct channel *c)
     return;
   }
 
-  int nlen = strlen(c->name) + strlen(c->proto->name) + 2;
-  char *rn = mb_allocz(c->proto->pool, nlen);
-  bsprintf(rn, "%s.%s", c->proto->name, c->name);
-
   c->in_req = (struct rt_import_request) {
-    .name = rn,
+    .name = mb_sprintf(c->proto->pool, "%s.%s", c->proto->name, c->name),
     .trace_routes = c->debug | c->proto->debug,
+    .list = proto_work_list(c->proto),
     .dump_req = channel_dump_import_req,
     .log_state_change = channel_import_log_state_change,
     .preimport = channel_preimport,
-    .rte_modify = c->proto->rte_modify,
   };
 
   ASSERT(c->channel_state == CS_UP);
@@ -462,12 +528,10 @@ channel_start_export(struct channel *c)
   }
 
   ASSERT(c->channel_state == CS_UP);
-  int nlen = strlen(c->name) + strlen(c->proto->name) + 2;
-  char *rn = mb_allocz(c->proto->pool, nlen);
-  bsprintf(rn, "%s.%s", c->proto->name, c->name);
 
   c->out_req = (struct rt_export_request) {
-    .name = rn,
+    .name = mb_sprintf(c->proto->pool, "%s.%s", c->proto->name, c->name),
+    .list = proto_work_list(c->proto),
     .addr = c->out_subprefix,
     .addr_mode = c->out_subprefix ? TE_ADDR_IN : TE_ADDR_NONE,
     .trace_routes = c->debug | c->proto->debug,
@@ -501,7 +565,7 @@ channel_start_export(struct channel *c)
   }
 
   DBG("%s.%s: Channel start export req=%p\n", c->proto->name, c->name, &c->out_req);
-  rt_request_export(&c->table->exporter, &c->out_req);
+  rt_request_export(c->table, &c->out_req);
 }
 
 static void
@@ -510,15 +574,15 @@ channel_check_stopped(struct channel *c)
   switch (c->channel_state)
   {
     case CS_STOP:
-      if (c->out_req.hook || c->in_req.hook)
+      if (!EMPTY_LIST(c->roa_subscriptions) || c->out_req.hook || c->in_req.hook)
 	return;
 
       channel_set_state(c, CS_DOWN);
-      ev_schedule(c->proto->event);
+      proto_send_event(c->proto);
 
       break;
     case CS_PAUSE:
-      if (c->out_req.hook)
+      if (!EMPTY_LIST(c->roa_subscriptions) || c->out_req.hook)
 	return;
 
       channel_set_state(c, CS_START);
@@ -535,8 +599,6 @@ channel_import_stopped(struct rt_import_request *req)
 {
   struct channel *c = SKIP_BACK(struct channel, in_req, req);
 
-  req->hook = NULL;
-
   mb_free(c->in_req.name);
   c->in_req.name = NULL;
 
@@ -555,7 +617,7 @@ channel_export_stopped(struct rt_export_request *req)
   {
     c->refeeding = 1;
     c->refeed_pending = 0;
-    rt_request_export(&c->table->exporter, req);
+    rt_request_export(c->table, req);
     return;
   }
 
@@ -599,7 +661,7 @@ channel_schedule_reload(struct channel *c)
 {
   ASSERT(c->in_req.hook);
 
-  rt_request_export(&c->table->exporter, &c->reload_req);
+  rt_request_export(c->table, &c->reload_req);
 }
 
 static void
@@ -634,6 +696,7 @@ channel_setup_in_table(struct channel *c)
 {
   c->reload_req = (struct rt_export_request) {
     .name = mb_sprintf(c->proto->pool, "%s.%s.import", c->proto->name, c->name),
+    .list = proto_work_list(c->proto),
     .trace_routes = c->debug | c->proto->debug,
     .export_bulk = channel_reload_export_bulk,
     .dump_req = channel_reload_dump_req,
@@ -698,8 +761,6 @@ channel_do_stop(struct channel *c)
 
   CALL(c->channel->shutdown, c);
 
-  /* This have to be done in here, as channel pool is freed before channel_do_down() */
-  c->out_table = NULL;
 }
 
 static void
@@ -721,7 +782,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
@@ -843,7 +904,7 @@ channel_config_new(const struct channel_class *cc, const char *name, uint net_ty
     if (proto->net_type && (net_type != proto->net_type))
       cf_error("Different channel type");
 
-    tab = new_config->def_tables[net_type];
+    tab = rt_get_default_table(new_config, net_type);
   }
 
   if (!cc)
@@ -862,6 +923,11 @@ channel_config_new(const struct channel_class *cc, const char *name, uint net_ty
   cf->debug = new_config->channel_default_debug;
   cf->rpki_reload = 1;
 
+  cf->roa_settle = (struct settle_config) {
+    .min = 1 S,
+    .max = 20 S,
+  };
+
   add_tail(&proto->channels, &cf->n);
 
   return cf;
@@ -942,6 +1008,22 @@ channel_reconfigure(struct channel *c, struct channel_config *cf)
   c->in_req.trace_routes = c->out_req.trace_routes = c->debug | c->proto->debug;
   c->rpki_reload = cf->rpki_reload;
 
+  if (	  (c->roa_settle.min != cf->roa_settle.min)
+       || (c->roa_settle.max != cf->roa_settle.max))
+  {
+    c->roa_settle = cf->roa_settle;
+
+    struct roa_subscription *s;
+    node *n;
+
+    WALK_LIST2(s, n, c->roa_subscriptions, roa_node)
+    {
+      s->settle.cf = cf->roa_settle;
+      if (settle_active(&s->settle))
+	settle_kick(&s->settle, &main_birdloop);
+    }
+  }
+
   /* Execute channel-specific reconfigure hook */
   if (c->channel->reconfigure && !c->channel->reconfigure(c, cf, &import_changed, &export_changed))
     return 0;
@@ -1032,34 +1114,48 @@ 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;
+  proto_cleanup(p);
+
+  birdloop_leave(&main_birdloop);
+}
+
+static void
+proto_event(void *ptr)
+{
+  struct proto *p = ptr;
 
   if (p->do_stop)
   {
     if (p->proto == &proto_unix_iface)
       if_flush_ifaces(p);
+
     p->do_stop = 0;
   }
 
   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);
 }
 
 
@@ -1100,10 +1196,10 @@ 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;
-  p->vrf_set = c->vrf_set;
   insert_node(&p->n, n);
 
   p->event = ev_new_init(proto_pool, proto_event, p);
@@ -1116,11 +1212,19 @@ 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");
+
+  p->pool = rp_newf(proto_pool, "Protocol %s", p->cf->name);
 
   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, p->pool->name);
+
+  PROTO_LOCKED_FROM_MAIN(p)
+    proto_notify_state(p, (p->proto->start ? p->proto->start(p) : PS_UP));
 }
 
 
@@ -1156,6 +1260,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);
 
@@ -1271,8 +1376,7 @@ proto_reconfigure(struct proto *p, struct proto_config *oc, struct proto_config
   if ((nc->protocol != oc->protocol) ||
       (nc->net_type != oc->net_type) ||
       (nc->disabled != p->disabled) ||
-      (nc->vrf != oc->vrf) ||
-      (nc->vrf_set != oc->vrf_set))
+      (nc->vrf != oc->vrf))
     return 0;
 
   p->name = nc->name;
@@ -1445,11 +1549,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;
@@ -1469,32 +1582,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 *
@@ -1699,7 +1792,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%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;
@@ -1931,24 +2024,32 @@ channel_reset_limit(struct channel *c, struct limit *l, int dir)
   c->limit_active &= ~(1 << dir);
 }
 
+static struct rte_owner_class default_rte_owner_class;
+
 static inline void
 proto_do_start(struct proto *p)
 {
   p->active = 1;
-  p->do_start = 1;
-  ev_schedule(p->event);
+
+  rt_init_sources(&p->sources, p->name, proto_event_list(p));
+  if (!p->sources.class)
+    p->sources.class = &default_rte_owner_class;
+
+  if (!p->cf->late_if_feed)
+    if_feed_baby(p);
 }
 
 static void
 proto_do_up(struct proto *p)
 {
   if (!p->main_source)
-  {
     p->main_source = rt_get_source(p, 0);
-    rt_lock_source(p->main_source);
-  }
+    // Locked automaticaly
 
   proto_start_channels(p);
+
+  if (p->cf->late_if_feed)
+    if_feed_baby(p);
 }
 
 static inline void
@@ -1963,9 +2064,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);
@@ -1973,6 +2071,10 @@ proto_do_stop(struct proto *p)
   }
 
   proto_stop_channels(p);
+  rt_destroy_sources(&p->sources, p->event);
+
+  p->do_stop = 1;
+  proto_send_event(p);
 }
 
 static void
@@ -1983,7 +2085,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);
 }
 
 
@@ -2193,8 +2295,8 @@ proto_cmd_show(struct proto *p, uintptr_t verbose, int cnt)
       cli_msg(-1006, "  Message:        %s", p->message);
     if (p->cf->router_id)
       cli_msg(-1006, "  Router ID:      %R", p->cf->router_id);
-    if (p->vrf_set)
-      cli_msg(-1006, "  VRF:            %s", p->vrf ? p->vrf->name : "default");
+    if (p->vrf)
+      cli_msg(-1006, "  VRF:            %s", p->vrf->name);
 
     if (p->proto->show_proto_info)
       p->proto->show_proto_info(p);
@@ -2222,7 +2324,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);
 }
 
@@ -2255,9 +2357,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);
 }
 
@@ -2332,7 +2434,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
@@ -2347,7 +2451,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");