ROA subscriptions are also converted to export requests.

By this, the requesting channels do the timers in their own loops,
avoiding unnecessary synchronization when the central timer went off.

This is of course less effective for now, yet it allows to easily
implement selective reloads in future.

5 files changed, 112 insertions, 127 deletions

diff --git a/nest/config.Y b/nest/config.Y
index 3ad6530d..ea7846e2 100644
--- a/nest/config.Y
+++ b/nest/config.Y
@@ -227,8 +227,6 @@ table_opt:
        cf_error("Trie option not supported for %s table", net_label[this_table->addr_type]);
      this_table->trie_used = $2;
    }
- | MIN SETTLE TIME expr_us { this_table->min_settle_time = $4; }
- | MAX SETTLE TIME expr_us { this_table->max_settle_time = $4; }
  | GC THRESHOLD expr { this_table->gc_threshold = $3; }
  | GC PERIOD expr_us { this_table->gc_period = (uint) $3; if ($3 > 3600 S_) cf_error("GC period must be at most 3600 s"); }
  | CORK THRESHOLD expr expr {
@@ -323,6 +321,8 @@ channel_item_:
  | RECEIVE LIMIT limit_spec { this_channel->rx_limit = $3; }
  | IMPORT LIMIT limit_spec { this_channel->in_limit = $3; }
  | EXPORT LIMIT limit_spec { this_channel->out_limit = $3; }
+ | MIN SETTLE TIME expr_us { this_channel->min_settle_time = $4; }
+ | MAX SETTLE TIME expr_us { this_channel->max_settle_time = $4; }
  | PREFERENCE expr { this_channel->preference = $2; check_u16($2); }
  | IMPORT KEEP FILTERED bool {
      if ($4)
diff --git a/nest/proto.c b/nest/proto.c
index d2374792..f4615c2f 100644
--- a/nest/proto.c
+++ b/nest/proto.c
@@ -55,6 +55,7 @@ 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) && proto_is_inactive(p); }
@@ -312,10 +313,19 @@ proto_remove_channels(struct proto *p)
     proto_remove_channel(p, c);
 }
 
+struct roa_subscription {
+  node roa_node;
+  timer t;
+  btime base_settle_time;		/* Start of settling interval */
+  struct channel *c;
+  struct rt_export_request req;
+};
+
 static void
-channel_roa_in_changed(void *_data)
+channel_roa_in_changed(struct timer *t)
 {
-  struct channel *c = _data;
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, t, t);
+  struct channel *c = s->c;
   int active = !!c->reload_req.hook;
 
   CD(c, "Reload triggered by RPKI change%s", active ? " - already active" : "");
@@ -327,9 +337,11 @@ channel_roa_in_changed(void *_data)
 }
 
 static void
-channel_roa_out_changed(void *_data)
+channel_roa_out_changed(struct timer *t)
 {
-  struct channel *c = _data;
+  struct roa_subscription *s = SKIP_BACK(struct roa_subscription, t, t);
+  struct channel *c = s->c;
+
   CD(c, "Feeding triggered by RPKI change");
 
   c->refeed_pending = 1;
@@ -338,29 +350,56 @@ channel_roa_out_changed(void *_data)
     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 */
+
+  if (!tm_active(&s->t))
+  {
+    s->base_settle_time = current_time();
+    tm_start(&s->t, s->base_settle_time + s->c->min_settle_time);
+  }
+  else
+    tm_set(&s->t,
+	MIN(s->base_settle_time + s->c->max_settle_time,
+	    current_time() + s->c->min_settle_time));
+
+
+  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;
+  rtable *tab = SKIP_BACK(rtable, exporter, req->hook->table);
+
+  debug("  Channel %s.%s ROA %s change notifier from table %s request %p\n",
+      c->proto->name, c->name,
+      (s->t.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)(void *) =
+  void (*hook)(struct timer *) =
     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.event->hook == hook))
+    if ((s->req.hook->table == &tab->exporter) && (s->t.hook == hook))
       return 1;
 
   return 0;
 }
 
-
 static void
 channel_roa_subscribe(struct channel *c, rtable *tab, int dir)
 {
@@ -368,21 +407,40 @@ channel_roa_subscribe(struct channel *c, rtable *tab, int dir)
     return;
 
   struct roa_subscription *s = mb_allocz(c->proto->pool, sizeof(struct roa_subscription));
-  s->s.event = ev_new_init(c->proto->pool, dir ? channel_roa_in_changed : channel_roa_out_changed, c);
-  s->s.list = proto_work_list(c->proto);
 
-  rt_subscribe(tab, &s->s);
+  *s = (struct roa_subscription) {
+    .t = { .hook = dir ? channel_roa_in_changed : channel_roa_out_changed, },
+    .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->exporter, &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);
-  rfree(s->s.event);
   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
@@ -525,7 +583,7 @@ 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);
@@ -533,7 +591,7 @@ channel_check_stopped(struct channel *c)
 
       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);
@@ -876,6 +934,9 @@ 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->min_settle_time = 1 S;
+  cf->max_settle_time = 20 S;
+
   add_tail(&proto->channels, &cf->n);
 
   return cf;
@@ -956,6 +1017,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->min_settle_time != cf->min_settle_time)
+       || (c->max_settle_time != cf->max_settle_time))
+  {
+    c->min_settle_time = cf->min_settle_time;
+    c->max_settle_time = cf->max_settle_time;
+
+    struct roa_subscription *s;
+    node *n;
+
+    WALK_LIST2(s, n, c->roa_subscriptions, roa_node)
+      if (tm_active(&s->t))
+	tm_set(&s->t,
+	    MIN(s->base_settle_time + c->max_settle_time,
+	      current_time() + c->min_settle_time));
+  }
+
   /* Execute channel-specific reconfigure hook */
   if (c->channel->reconfigure && !c->channel->reconfigure(c, cf, &import_changed, &export_changed))
     return 0;
diff --git a/nest/protocol.h b/nest/protocol.h
index 709d6415..a3eeb198 100644
--- a/nest/protocol.h
+++ b/nest/protocol.h
@@ -463,6 +463,9 @@ struct channel_config {
   struct channel_limit in_limit;	/* Limit for importing routes from protocol */
   struct channel_limit out_limit;	/* Limit for exporting routes to protocol */
 
+  btime min_settle_time;		/* Minimum settle time for ROA-induced reload */
+  btime max_settle_time;		/* Maximum settle time for ROA-induced reload */
+
   u8 net_type;				/* Routing table network type (NET_*), 0 for undefined */
   u8 ra_mode;				/* Mode of received route advertisements (RA_*) */
   u16 preference;			/* Default route preference */
@@ -490,6 +493,9 @@ struct channel {
   struct limit in_limit;		/* Input limit */
   struct limit out_limit;		/* Output limit */
 
+  btime min_settle_time;		/* Minimum settle time for ROA-induced reload */
+  btime max_settle_time;		/* Maximum settle time for ROA-induced reload */
+
   u8 limit_actions[PLD_MAX];		/* Limit actions enum */
   u8 limit_active;			/* Flags for active limits */
 
@@ -543,7 +549,7 @@ struct channel {
 
   struct rt_exporter *out_table;	/* Internal table for exported routes */
 
-  list roa_subscriptions;		/* List of active ROA table subscriptions based on filters roa_check() */
+  list roa_subscriptions;		/* List of active ROA table subscriptions based on filters' roa_check() calls */
 };
 
 #define RIK_REJECTED	1			/* Routes rejected in import filter are kept */
diff --git a/nest/rt-table.c b/nest/rt-table.c
index 72f85370..2fa736e2 100644
--- a/nest/rt-table.c
+++ b/nest/rt-table.c
@@ -82,9 +82,8 @@
  * will be re-validated later in this round anyway.
  *
  * The third mechanism is used for RPKI re-validation of IP routes and it is the
- * simplest. It is just a list of subscribers in src table, who are notified
- * when any change happened, but only after a settle time. Also, in RPKI case
- * the dst is not a table, but a channel, who refeeds routes through a filter.
+ * simplest. It is also an auxiliary export request belonging to the
+ * appropriate channel, triggering its reload/refeed timer after a settle time.
  */
 
 #undef LOCAL_DEBUG
@@ -134,7 +133,6 @@ static void rt_next_hop_update(rtable *tab);
 static inline void rt_next_hop_resolve_rte(rte *r);
 static inline void rt_flowspec_resolve_rte(rte *r, struct channel *c);
 static inline void rt_prune_table(rtable *tab);
-static inline void rt_schedule_notify(rtable *tab);
 static void rt_kick_prune_timer(rtable *tab);
 static void rt_feed_by_fib(void *);
 static void rt_feed_by_trie(void *);
@@ -1270,15 +1268,10 @@ rte_announce(rtable *tab, net *net, struct rte_storage *new, struct rte_storage
   if ((new == old) && (new_best == old_best))
     return;
 
-  if (new_best_valid || old_best_valid)
-  {
-    if (new_best_valid)
-      new_best->rte.sender->stats.pref++;
-    if (old_best_valid)
-      old_best->rte.sender->stats.pref--;
-  }
-
-  rt_schedule_notify(tab);
+  if (new_best_valid)
+    new_best->rte.sender->stats.pref++;
+  if (old_best_valid)
+    old_best->rte.sender->stats.pref--;
 
   if (EMPTY_LIST(tab->exporter.hooks) && EMPTY_LIST(tab->exporter.pending))
   {
@@ -2371,78 +2364,6 @@ rt_kick_prune_timer(rtable *tab)
 }
 
 
-static inline btime
-rt_settled_time(rtable *tab)
-{
-  ASSUME(tab->base_settle_time != 0);
-
-  return MIN(tab->last_rt_change + tab->config->min_settle_time,
-	     tab->base_settle_time + tab->config->max_settle_time);
-}
-
-static void
-rt_settle_timer(timer *t)
-{
-  rtable *tab = t->data;
-
-  if (!tab->base_settle_time)
-    return;
-
-  btime settled_time = rt_settled_time(tab);
-  if (current_time() < settled_time)
-  {
-    tm_set(tab->settle_timer, settled_time);
-    return;
-  }
-
-  /* Settled */
-  tab->base_settle_time = 0;
-
-  struct rt_subscription *s;
-  WALK_LIST(s, tab->subscribers)
-    ev_send(s->list, s->event);
-}
-
-static void
-rt_kick_settle_timer(rtable *tab)
-{
-  tab->base_settle_time = current_time();
-
-  if (!tab->settle_timer)
-    tab->settle_timer = tm_new_init(tab->rp, rt_settle_timer, tab, 0, 0);
-
-  if (!tm_active(tab->settle_timer))
-    tm_set(tab->settle_timer, rt_settled_time(tab));
-}
-
-static inline void
-rt_schedule_notify(rtable *tab)
-{
-  if (EMPTY_LIST(tab->subscribers))
-    return;
-
-  if (tab->base_settle_time)
-    return;
-
-  rt_kick_settle_timer(tab);
-}
-
-void
-rt_subscribe(rtable *tab, struct rt_subscription *s)
-{
-  s->tab = tab;
-  rt_lock_table(tab);
-  DBG("rt_subscribe(%s)\n", tab->name);
-  add_tail(&tab->subscribers, &s->n);
-}
-
-void
-rt_unsubscribe(struct rt_subscription *s)
-{
-  rem_node(&s->n);
-  rt_unlock_table(s->tab);
-}
-
 static void
 rt_flowspec_export_one(struct rt_export_request *req, const net_addr *net, struct rt_pending_export *first)
 {
@@ -2580,7 +2501,6 @@ rt_free(resource *_r)
   fib_free(&r->fib);
   hmap_free(&r->id_map);
   rfree(r->rt_event);
-  rfree(r->settle_timer);
   mb_free(r);
   */
 }
@@ -2642,8 +2562,6 @@ rt_setup(pool *pp, struct rtable_config *cf)
   hmap_init(&t->id_map, p, 1024);
   hmap_set(&t->id_map, 0);
 
-  init_list(&t->subscribers);
-
   t->rt_event = ev_new_init(p, rt_event, t);
   t->uncork_event = ev_new_init(p, rt_uncork_event, t);
   t->prune_timer = tm_new_init(p, rt_prune_timer, t, 0, 0);
@@ -3629,8 +3547,6 @@ rt_new_table(struct symbol *s, uint addr_type)
   c->addr_type = addr_type;
   c->gc_threshold = 1000;
   c->gc_period = (uint) -1;	/* set in rt_postconfig() */
-  c->min_settle_time = 1 S;
-  c->max_settle_time = 20 S;
   c->cork_threshold.low = 128;
   c->cork_threshold.high = 512;
   c->debug = new_config->table_debug;
diff --git a/nest/rt.h b/nest/rt.h
index 01e444f6..c5948512 100644
--- a/nest/rt.h
+++ b/nest/rt.h
@@ -61,8 +61,6 @@ struct rtable_config {
   byte sorted;				/* Routes of network are sorted according to rte_better() */
   byte trie_used;			/* Rtable has attached trie */
   byte debug;				/* Whether to log */
-  btime min_settle_time;		/* Minimum settle time for notifications */
-  btime max_settle_time;		/* Maximum settle time for notifications */
   btime export_settle_time;		/* Delay before exports are announced */
   struct rt_cork_threshold cork_threshold;	/* Cork threshold values */
 };
@@ -112,7 +110,6 @@ typedef struct rtable {
   struct event *uncork_event;		/* Called when uncork happens */
   struct timer *prune_timer;		/* Timer for periodic pruning / GC */
   btime last_rt_change;			/* Last time when route changed */
-  btime base_settle_time;		/* Start time of rtable settling interval */
   btime gc_time;			/* Time of last GC */
   uint gc_counter;			/* Number of operations since last GC */
   byte prune_state;			/* Table prune state, 1 -> scheduled, 2-> running */
@@ -130,18 +127,9 @@ typedef struct rtable {
   u32 trie_old_lock_count;		/* Old prefix trie locked by walks */
   struct tbf rl_pipe;			/* Rate limiting token buffer for pipe collisions */
 
-  list subscribers;			/* Subscribers for notifications */
-  struct timer *settle_timer;		/* Settle time for notifications */
   struct f_trie *flowspec_trie;		/* Trie for evaluation of flowspec notifications */
 } rtable;
 
-struct rt_subscription {
-  node n;
-  rtable *tab;
-  event *event;
-  event_list *list;
-};
-
 extern struct rt_cork {
   _Atomic uint active;
   event_list queue;
@@ -461,8 +449,6 @@ void rt_lock_table(rtable *);
 void rt_unlock_table(rtable *);
 struct f_trie * rt_lock_trie(rtable *tab);
 void rt_unlock_trie(rtable *tab, struct f_trie *trie);
-void rt_subscribe(rtable *tab, struct rt_subscription *s);
-void rt_unsubscribe(struct rt_subscription *s);
 void rt_flowspec_link(rtable *src, rtable *dst);
 void rt_flowspec_unlink(rtable *src, rtable *dst);
 rtable *rt_setup(pool *, struct rtable_config *);