rt_setup to do!

nest/route.h

@@ -433,19 +433,20 @@ static inline int rt_cork_check(event *e)
 
 struct rt_pending_export {
   struct rt_export_item it;
-  struct rt_pending_export *_Atomic next;
+  struct rt_pending_export *_Atomic next;	/* Next export for the same net */
+  struct rt_pending_export *the_other;		/* Export interlink for table cleanup */
 };
 
 struct rt_net_pending_export {
-  struct rt_pending_export *first, *last;
+  struct rt_pending_export * _Atomic first, * _Atomic last;
 };
 
 typedef struct network {
   struct rte_storage * _Atomic routes;		/* Available routes for this network */
 
   /* Uncleaned pending exports */
-  struct rt_net_pending_export _Atomic all;
-  struct rt_net_pending_export _Atomic best;
+  struct rt_net_pending_export all;
+  struct rt_net_pending_export best;
 } net;
 
 struct rte_storage {
@@ -671,7 +672,7 @@ void rt_flowspec_link(rtable *src, rtable *dst);
 void rt_flowspec_unlink(rtable *src, rtable *dst);
 rtable *rt_setup(pool *, struct rtable_config *);
 
-struct rt_export_feed *rt_net_feed(rtable *t, const net_addr *a, const struct rt_export_item *first);
+struct rt_export_feed *rt_net_feed(rtable *t, const net_addr *a, const struct rt_pending_export *first);
 rte rt_net_best(rtable *t, const net_addr *a);
 int rt_examine(rtable *t, net_addr *a, struct channel *c, const struct filter *filter);
 rte *rt_export_merged(struct channel *c, const struct rt_export_feed *feed, linpool *pool, int silent);

nest/rt-table.c

@@ -124,6 +124,7 @@ list routing_tables;
 list deleted_routing_tables;
 
 netindex_hash *rt_global_netindex_hash;
+#define RT_INITIAL_ROUTES_BLOCK_SIZE   128
 
 struct rt_cork rt_cork;
 
@@ -909,7 +910,7 @@ channel_notify_accepted(void *_channel)
 
       case RT_EXPORT_UPDATE:
 	{
-	  struct rt_export_feed *f = rt_net_feed(c->table, u->update->new ? u->update->new->net : u->update->old->net, u->update);
+	  struct rt_export_feed *f = rt_net_feed(c->table, u->update->new ? u->update->new->net : u->update->old->net, SKIP_BACK(struct rt_pending_export, it, u->update));
 	  rt_notify_accepted(c, f);
 	  for (uint i=0; i<f->count_exports; i++)
 	    rt_export_processed(&c->out_req, f->exports[i]);
@@ -1039,7 +1040,7 @@ channel_notify_merged(void *_channel)
 
       case RT_EXPORT_UPDATE:
 	{
-	  struct rt_export_feed *f = rt_net_feed(c->table, u->update->new ? u->update->new->net : u->update->old->net, u->update);
+	  struct rt_export_feed *f = rt_net_feed(c->table, u->update->new ? u->update->new->net : u->update->old->net, SKIP_BACK(struct rt_pending_export, it, u->update));
 	  rt_notify_merged(c, f);
 	  for (uint i=0; i<f->count_exports; i++)
 	    rt_export_processed(&c->out_req, f->exports[i]);
@@ -1125,11 +1126,11 @@ channel_notify_basic(void *_channel)
   }
 }
 
-static void
-rte_announce_to(struct rt_exporter *e, _Atomic struct rt_net_pending_export *npe, const rte *new, const rte *old)
+static struct rt_pending_export *
+rte_announce_to(struct rt_exporter *e, struct rt_net_pending_export *npe, const rte *new, const rte *old)
 {
   if (new == old)
-    return;
+    return NULL;
 
   struct rt_pending_export rpe = {
     .it = {
@@ -1139,16 +1140,16 @@ rte_announce_to(struct rt_exporter *e, _Atomic struct rt_net_pending_export *npe
   };
 
   SKIP_BACK_DECLARE(struct rt_pending_export, pushed, it, rt_exporter_push(e, &rpe.it));
-  struct rt_net_pending_export nloc = atomic_load_explicit(npe, memory_order_relaxed);
 
-  if (nloc.last)
-    ASSERT_DIE(atomic_exchange_explicit(&nloc.last->next, pushed, memory_order_acq_rel) == NULL);
-  
-  nloc.last = pushed;
-  if (!nloc.first)
-    nloc.first = pushed;
+  struct rt_pending_export *last = atomic_load_explicit(&npe->last, memory_order_relaxed);
+  if (last)
+    ASSERT_DIE(atomic_exchange_explicit(&last->next, pushed, memory_order_acq_rel) == NULL);
+ 
+  atomic_store_explicit(&npe->last, pushed, memory_order_release);
+  if (!atomic_load_explicit(&npe->first, memory_order_relaxed))
+    atomic_store_explicit(&npe->first, pushed, memory_order_release);
 
-  atomic_store_explicit(npe, nloc, memory_order_release);
+  return pushed;
 }
 
 static void
@@ -1180,54 +1181,89 @@ rte_announce(struct rtable_private *tab, const struct netindex *i UNUSED, net *n
     return;
   }
 
-  rte_announce_to(&tab->export_all, &net->all, new, old);
-  rte_announce_to(&tab->export_best, &net->best, new_best, old_best);
+  struct rt_pending_export *best_rpe, *all_rpe;
+  best_rpe = rte_announce_to(&tab->export_best, &net->best, new_best, old_best);
+  all_rpe = rte_announce_to(&tab->export_all, &net->all, new, old);
+
+  if (best_rpe && all_rpe)
+  {
+    best_rpe->the_other = all_rpe;
+    all_rpe->the_other = best_rpe;
+  }
 
   rt_check_cork_high(tab);
 }
 
-static void
-rt_cleanup_export(struct lfjour *j, struct lfjour_item *i)
+static _Bool
+rt_cleanup_export(struct rtable_private *tab, struct rt_net_pending_export *npe, struct rt_pending_export *rpe)
 {
-  SKIP_BACK_DECLARE(struct rtable_private, tab, journal, j);
-  SKIP_BACK_DECLARE(struct rt_pending_export, rpe, li, i);
-
-  /* Unlink this export from struct network */
-  ASSERT_DIE(rpe->new || rpe->old);
-  const net_addr *n = rpe->new ?
-    rpe->new->net :
-    rpe->old->net;
-  struct netindex *ni = NET_TO_INDEX(n);
-  ASSERT_DIE(ni->index < atomic_load_explicit(&tab->routes_block_size, memory_order_relaxed));
-  net *routes = atomic_load_explicit(&tab->routes, memory_order_relaxed);
-  net *net = &routes[ni->index];
-
-  ASSERT_DIE(rpe == atomic_load_explicit(&net->first, memory_order_relaxed));
+  struct rt_pending_export *first = atomic_load_explicit(&npe->first, memory_order_relaxed);
+  struct rt_pending_export *last = atomic_load_explicit(&npe->last, memory_order_relaxed);
+  ASSERT_DIE(rpe == first);
 
   /* Update the first and last pointers */
-  struct rt_pending_export *last = rpe,
-			   *next = atomic_load_explicit(&rpe->next, memory_order_relaxed);
-  if (atomic_compare_exchange_strong_explicit(
-	&net->last, &last, NULL,
-	memory_order_acq_rel, memory_order_acquire))
-    ASSERT_DIE(next == NULL);
+  atomic_store_explicit(
+      &npe->first,
+      atomic_load_explicit(&rpe->next, memory_order_relaxed),
+      memory_order_release
+      );
 
-  ASSERT_DIE(atomic_compare_exchange_strong_explicit(
-	&net->first, &rpe, next,
-	memory_order_acq_rel, memory_order_relaxed));
+  if (rpe == last)
+    atomic_store_explicit(&npe->last, NULL, memory_order_release);
 
   /* Wait for very slow table readers */
   synchronize_rcu();
 
-  if (rpe->old)
+  /* Cleanups may need to be finalized later */
+  if (rpe->the_other)
   {
-    ASSERT_DIE(rpe->old->flags & REF_OBSOLETE);
-    hmap_clear(&tab->id_map, rpe->old->id);
-    rte_free(SKIP_BACK(struct rte_storage, rte, rpe->old), tab);
+    rpe->the_other->the_other = NULL;
+    rpe->the_other = NULL;
+  }
+  else if (rpe->it.old && (rpe->it.old->flags & REF_OBSOLETE))
+  {
+    /* When cleaning up the best journal, the old route need not be obsolete */
+    hmap_clear(&tab->id_map, rpe->it.old->id);
+    rte_free(SKIP_BACK(struct rte_storage, rte, rpe->it.old), tab);
   }
 
-  if (!routes && !next)
-    tab->gc_counter++;
+  return (rpe == last);
+}
+
+static struct network *
+rt_cleanup_export_get_net(struct rtable_private *tab, struct rt_pending_export *rpe)
+{
+  /* Unlink this export from struct network */
+  ASSERT_DIE(rpe->it.new || rpe->it.old);
+  const net_addr *n = rpe->it.new ?
+    rpe->it.new->net :
+    rpe->it.old->net;
+  struct netindex *ni = NET_TO_INDEX(n);
+  ASSERT_DIE(ni->index < atomic_load_explicit(&tab->routes_block_size, memory_order_relaxed));
+  net *routes = atomic_load_explicit(&tab->routes, memory_order_relaxed);
+  return &routes[ni->index];
+}
+
+static void
+rt_cleanup_export_best(struct lfjour *j, struct lfjour_item *i)
+{
+  SKIP_BACK_DECLARE(struct rt_pending_export, rpe, it.li, i);
+  SKIP_BACK_DECLARE(struct rtable_private, t, export_best.journal, j);
+  net *net = rt_cleanup_export_get_net(t, rpe);
+  if (rt_cleanup_export(t, &net->best, rpe) &&
+      !atomic_load_explicit(&net->routes, memory_order_relaxed))
+    t->gc_counter++;
+}
+
+static void
+rt_cleanup_export_all(struct lfjour *j, struct lfjour_item *i)
+{
+  SKIP_BACK_DECLARE(struct rt_pending_export, rpe, it.li, i);
+  SKIP_BACK_DECLARE(struct rtable_private, t, export_all.journal, j);
+  net *net = rt_cleanup_export_get_net(t, rpe);
+  if (rt_cleanup_export(t, &net->all, rpe) &&
+      !atomic_load_explicit(&net->routes, memory_order_relaxed))
+    t->gc_counter++;
 }
 
 static void
@@ -1257,9 +1293,8 @@ rt_import_cleared(void *_ih)
 }
 
 static void
-rt_cleanup_done(struct lfjour *j, u64 begin_seq, u64 end_seq)
+rt_cleanup_done(struct rtable_private *tab, u64 begin_seq, u64 end_seq)
 {
-  SKIP_BACK_DECLARE(struct rtable_private, tab, journal, j);
   ASSERT_DIE(DG_IS_LOCKED(tab->lock.rtable));
 
   if (~end_seq)
@@ -1292,49 +1327,21 @@ rt_cleanup_done(struct lfjour *j, u64 begin_seq, u64 end_seq)
     rt_kick_prune_timer(tab);
 }
 
-#define RT_EXPORT_BULK	1024
-
 static void
-rt_export_hook(void *_data)
+rt_cleanup_done_best(struct lfjour *j, u64 begin_seq, u64 end_seq)
 {
-  struct rt_export_hook *c = _data;
-  struct lfjour_recipient *r = &c->recipient;
-
-  ASSERT_DIE(atomic_load_explicit(&c->export_state, memory_order_relaxed) == TES_READY);
-
-  /* Process the export */
-  for (uint i=0; i<RT_EXPORT_BULK; i++)
-  {
-    /* Get the next export if exists */
-    struct lfjour_item *li = lfjour_get(r);
-
-    /* Stop exporting if no export is available */
-    if (!li)
-      return;
-
-    /* Process sequence number reset event */
-    if (lfjour_reset_seqno(r))
-      bmap_reset(&c->seq_map, 16);
-
-    /* Process the export */
-    rte_export(c, SKIP_BACK(struct rt_pending_export, li, li));
-
-    /* And release the export */
-    lfjour_release(r);
-  }
-
-  /*
-   * is this actually needed?
-  if (used)
-    RT_LOCKED(tab, t)
-      if (no_next || t->cork_active)
-	rt_export_used(c->table, c->req->name, no_next ? "finished export bulk" : "cork active");
-	*/
-
-  /* Request continuation */
-  rt_send_export_event(c);
+  SKIP_BACK_DECLARE(struct rtable_private, t, export_best.journal, j);
+  rt_cleanup_done(t, begin_seq, end_seq);
 }
 
+static void
+rt_cleanup_done_all(struct lfjour *j, u64 begin_seq, u64 end_seq)
+{
+  SKIP_BACK_DECLARE(struct rtable_private, t, export_all.journal, j);
+  rt_cleanup_done(t, begin_seq, end_seq);
+}
+
+#define RT_EXPORT_BULK	1024
 
 static inline int
 rte_validate(struct channel *ch, rte *e)
@@ -1875,7 +1882,7 @@ rte_import(struct rt_import_request *req, const net_addr *n, rte *new, struct rt
 }
 
 struct rt_export_feed *
-rt_net_feed(rtable *t, net_addr *a)
+rt_net_feed(rtable *t, const net_addr *a, const struct rt_pending_export *first)
 {
   RT_READ(t, tr);
 
@@ -1885,18 +1892,37 @@ rt_net_feed(rtable *t, net_addr *a)
     return 0;
 
   /* Get the feed itself. It may change under our hands tho. */
-  struct rt_pending_export *first = atomic_load_explicit(&n->first, memory_order_acquire);
-  struct rt_pending_export *last = atomic_load_explicit(&n->last, memory_order_acquire);
+  struct rt_pending_export *first_in_net, *last_in_net;
+  first_in_net = atomic_load_explicit(&n->all.first, memory_order_acquire);
+  last_in_net = atomic_load_explicit(&n->all.last, memory_order_acquire);
+
+  /* Export item validity check: we must find it between first_in_net and last_in_net */
+  if (first)
+  {
+    struct rt_pending_export *rpe = first_in_net;
+    while (rpe)
+      if (rpe == first)
+	break;
+      else if (rpe == last_in_net)
+	bug("Called rt_net_feed with an invalid first journal item");
+      else
+	rpe = atomic_load_explicit(&rpe->next, memory_order_acquire);
+
+    if (rpe != first)
+      bug("Called rt_net_feed with an invalid first journal item");
+  }
+  else
+    first = first_in_net;
 
   /* Count the elements */
   uint rcnt = rte_feed_count(tr, n);
   uint ecnt = 0;
   uint ocnt = 0;
-  for (struct rt_pending_export *rpe = first; rpe;
+  for (const struct rt_pending_export *rpe = first; rpe;
       rpe = atomic_load_explicit(&rpe->next, memory_order_acquire))
   {
     ecnt++;
-    if (rpe->old)
+    if (rpe->it.old)
       ocnt++;
   }
 
@@ -1926,20 +1952,20 @@ rt_net_feed(rtable *t, net_addr *a)
     {
       uint e = 0;
       uint rpos = rcnt;
-      for (struct rt_pending_export *rpe = first; rpe;
+      for (const struct rt_pending_export *rpe = first; rpe;
 	  rpe = atomic_load_explicit(&rpe->next, memory_order_acquire))
 	if (e >= ecnt)
 	  RT_READ_RETRY(tr);
 	else
 	{
-	  feed->exports[e++] = rpe->seq;
+	  feed->exports[e++] = rpe->it.seq;
 
 	  /* Copy also obsolete routes */
-	  if (rpe->old)
+	  if (rpe->it.old)
 	  {
 	    ASSERT_DIE(rpos < rcnt + ocnt);
-	    feed->block[rpos++] = *rpe->old;
-	    ea_free_later(ea_ref(rpe->old->attrs));
+	    feed->block[rpos++] = *rpe->it.old;
+	    ea_free_later(ea_ref(rpe->it.old->attrs));
 	  }
 	}
 
@@ -1948,15 +1974,16 @@ rt_net_feed(rtable *t, net_addr *a)
   }
 
   /* Check that it indeed didn't change and the last export is still the same. */
-  if (last != atomic_load_explicit(&n->last, memory_order_acquire) ||
-      first != atomic_load_explicit(&n->first, memory_order_acquire))
+  if (
+      (first_in_net != atomic_load_explicit(&n->all.first, memory_order_acquire))
+   || (last_in_net != atomic_load_explicit(&n->all.last, memory_order_acquire)))
     RT_READ_RETRY(tr);
 
   return feed;
 }
 
 rte
-rt_net_best(rtable *t, net_addr *a)
+rt_net_best(rtable *t, const net_addr *a)
 {
   rte rt = {};
 
@@ -1988,34 +2015,6 @@ rt_examine(rtable *t, net_addr *a, struct channel *c, const struct filter *filte
   return v > 0;
 }
 
-static void
-rt_table_export_done(void *hh)
-{
-  struct rt_export_hook *hook = hh;
-  struct rt_export_request *req = hook->req;
-  void (*stopped)(struct rt_export_request *) = hook->stopped;
-  rtable *t = hook->tab;
-
-  /* Drop the hook */
-  RT_LOCKED(t, tab)
-  {
-    /* Unlink from the table */
-    if (lfjour_of_recipient(&hook->recipient))
-      lfjour_unregister(&hook->recipient);
-
-    DBG("Export hook %p in table %s finished uc=%u\n", hook, tab->name, tab->use_count);
-
-    /* Free the hook */
-    rp_free(hook->pool);
-  }
-
-  /* Inform the stopper */
-  CALL(stopped, req);
-
-  /* Unlock the table */
-  rt_unlock_table(t);
-}
-
 static inline void
 rt_set_import_state(struct rt_import_hook *hook, u8 state)
 {
@@ -2025,24 +2024,6 @@ rt_set_import_state(struct rt_import_hook *hook, u8 state)
   CALL(hook->req->log_state_change, hook->req, state);
 }
 
-u8
-rt_set_export_state(struct rt_export_hook *hook, u32 expected_mask, u8 state)
-{
-  hook->last_state_change = current_time();
-  u8 old = atomic_exchange_explicit(&hook->export_state, state, memory_order_release);
-  if (!((1 << old) & expected_mask))
-    bug("Unexpected export state change from %s to %s, expected mask %02x",
-      rt_export_state_name(old),
-      rt_export_state_name(state),
-      expected_mask
-      );
-
-  if (old != state)
-    CALL(hook->req->log_state_change, hook->req, state);
-
-  return old;
-}
-
 void
 rt_request_import(rtable *t, struct rt_import_request *req)
 {
@@ -2087,164 +2068,6 @@ rt_stop_import(struct rt_import_request *req, void (*stopped)(struct rt_import_r
   }
 }
 
-static void rt_table_export_start_feed(struct rtable_private *tab, struct rt_export_hook *hook);
-static void
-rt_table_export_uncork(void *_hook)
-{
-  ASSERT_DIE(birdloop_inside(&main_birdloop));
-
-  struct rt_export_hook *hook = _hook;
-  struct birdloop *loop = hook->req->list->loop;
-
-  if (loop != &main_birdloop)
-    birdloop_enter(loop);
-
-  u8 state;
-  RT_LOCKED(hook->tab, tab)
-    switch (state = atomic_load_explicit(&hook->export_state, memory_order_relaxed))
-    {
-      case TES_HUNGRY:
-	rt_table_export_start_feed(tab, hook);
-	break;
-      case TES_STOP:
-	hook->event->hook = rt_table_export_done;
-	rt_send_export_event(hook);
-	break;
-      default:
-	bug("Uncorking a table export in a strange state: %u", state);
-    }
-
-  if (loop != &main_birdloop)
-    birdloop_leave(loop);
-}
-
-static void
-rt_table_export_start_locked(struct rtable_private *tab, struct rt_export_request *req)
-{
-  rt_lock_table(tab);
-
-  pool *p = rp_new(req->pool, req->pool->domain, "Export hook");
-  struct rt_export_hook *hook = req->hook = mb_allocz(p, sizeof(struct rt_export_hook));
-  hook->req = req;
-  hook->tab = RT_PUB(tab);
-  hook->pool = p;
-  atomic_store_explicit(&hook->export_state, TES_DOWN, memory_order_release);
-  hook->event = ev_new_init(p, rt_table_export_uncork, hook);
-
-  if (rt_cork_check(hook->event))
-    rt_set_export_state(hook, BIT32_ALL(TES_DOWN), TES_HUNGRY);
-  else
-    rt_table_export_start_feed(tab, hook);
-}
-
-static void
-rt_table_export_start_feed(struct rtable_private *tab, struct rt_export_hook *hook)
-{
-  struct rt_export_request *req = hook->req;
-
-  /* stats zeroed by mb_allocz */
-  switch (req->prefilter.mode)
-  {
-    case TE_ADDR_IN:
-    case TE_ADDR_NONE:
-    case TE_ADDR_TRIE:
-    case TE_ADDR_HOOK:
-      hook->feed_index = 0;
-      hook->event->hook = rt_feed_by_fib;
-      break;
-
-    case TE_ADDR_EQUAL:
-      hook->event->hook = rt_feed_equal;
-      break;
-
-    case TE_ADDR_FOR:
-      hook->event->hook = rt_feed_for;
-      break;
-
-    default:
-      bug("Requested an unknown export address mode");
-  }
-
-  DBG("New export hook %p req %p in table %s uc=%u\n", hook, req, tab->name, tab->use_count);
-
-  hook->recipient = (struct lfjour_recipient) {
-    .event = hook->event,
-    .target = req->list,
-  };
-  lfjour_register(&tab->journal, &hook->recipient);
-
-  SKIP_BACK_DECLARE(struct rt_pending_export, rpe, li, atomic_load_explicit(&hook->recipient.last, memory_order_relaxed));
-  req_trace(req, D_STATES, "Export initialized, last export %p (%lu)", rpe, rpe ? rpe->seq : 0);
-
-  bmap_init(&hook->seq_map, hook->pool, 16);
-
-  /* Regular export */
-  rt_set_export_state(hook, BIT32_ALL(TES_DOWN, TES_HUNGRY), TES_FEEDING);
-  rt_send_export_event(hook);
-}
-
-#if 0
-static void
-rt_table_export_start(struct rt_exporter *re, struct rt_export_request *req)
-{
-  RT_LOCKED(SKIP_BACK(rtable, priv.exporter, re), tab)
-    rt_table_export_start_locked(tab, req);
-}
-#endif
-
-void rt_request_export(rtable *t, struct rt_export_request *req)
-{
-  RT_LOCKED(t, tab)
-    rt_table_export_start_locked(tab, req);  /* Is locked inside */
-}
-
-static void
-rt_stop_export_locked(struct rtable_private *tab, struct rt_export_hook *hook)
-{
-  struct rt_export_request *req = hook->req;
-
-  /* Update export state, get old */
-  switch (rt_set_export_state(hook, BIT32_ALL(TES_HUNGRY, TES_FEEDING, TES_READY), TES_STOP))
-  {
-    case TES_STOP:
-      rt_trace(tab, D_EVENTS, "Stopping export hook %s already requested", req->name);
-      return;
-
-    case TES_HUNGRY:
-      rt_trace(tab, D_EVENTS, "Stopping export hook %s must wait for uncorking", req->name);
-      return;
-
-    case TES_FEEDING:
-      break;
-  }
-
-  rt_trace(tab, D_EVENTS, "Stopping export hook %s right now", req->name);
-
-  /* Reset the event as the stopped event */
-  ASSERT_DIE(birdloop_inside(req->list->loop));
-  hook->event->hook = rt_table_export_done;
-
-  /* Run the stopped event */
-  rt_send_export_event(hook);
-}
-
-void
-rt_stop_export(struct rt_export_request *req, void (*stopped)(struct rt_export_request *))
-{
-  ASSERT_DIE(birdloop_inside(req->list->loop));
-  struct rt_export_hook *hook = req->hook;
-  ASSERT_DIE(hook);
-
-  RT_LOCKED(hook->tab, t)
-  {
-    /* Set the stopped callback */
-    hook->stopped = stopped;
-
-    /* Do the rest */
-    rt_stop_export_locked(t, hook);
-  }
-}
-
 
 /**
  * rt_refresh_begin - start a refresh cycle
@@ -2421,6 +2244,8 @@ rt_dump_hooks(rtable *tp)
        ih->last_state_change, ih->import_state, ih->stopped);
   }
 
+#if 0
+  /* FIXME: I'm very lazy to write this now */
   WALK_TLIST(lfjour_recipient, r, &tab->journal.recipients)
   {
     SKIP_BACK_DECLARE(struct rt_export_hook, eh, recipient, r);
@@ -2430,6 +2255,7 @@ rt_dump_hooks(rtable *tp)
        eh, eh->req, eh->refeed_pending, eh->last_state_change,
        atomic_load_explicit(&eh->export_state, memory_order_relaxed));
   }
+#endif
   debug("\n");
 
   }
@@ -2513,43 +2339,72 @@ struct rt_flowspec_link {
   rtable *dst;
   u32 uc;
   struct rt_export_request req;
+  event event;
 };
 
 #include "lib/tlists.h"
 
 
 static void
-rt_flowspec_export_one(struct rt_export_request *req, const net_addr *net, struct rt_pending_export *first)
+rt_flowspec_export(void *_link)
 {
-  SKIP_BACK_DECLARE(struct rt_flowspec_link, ln, req, req);
+  struct rt_flowspec_link *ln = _link;
   rtable *dst_pub = ln->dst;
   ASSUME(rt_is_flow(dst_pub));
 
-  RT_LOCKED(dst_pub, dst)
+  RT_EXPORT_WALK(&ln->req, u)
   {
+    const net_addr *n = NULL;
+    switch (u->kind)
+    {
+      case RT_EXPORT_STOP:
+	bug("Main table export stopped");
 
-  /* No need to inspect it further if recalculation is already scheduled */
-  if ((dst->nhu_state == NHU_SCHEDULED) || (dst->nhu_state == NHU_DIRTY)
-      || !trie_match_net(dst->flowspec_trie, net))
-  {
-    rpe_mark_seen_all(req->hook, first, NULL, NULL);
-    return;
-  }
+      case RT_EXPORT_FEED:
+	if (u->feed->count_routes)
+	  n = u->feed->block[0].net;
+	break;
 
-  /* This net may affect some flowspecs, check the actual change */
-  const rte *o = RTE_VALID_OR_NULL(first->old_best);
-  const rte *new_best = first->new_best;
+      case RT_EXPORT_UPDATE:
+	n = (u->update->new ?: u->update->old)->net;
+	break;
+    }
 
-  RPE_WALK(first, rpe, NULL)
-  {
-    rpe_mark_seen(req->hook, rpe);
-    new_best = rpe->new_best;
-  }
+    if (!n)
+      continue;
 
-  /* Yes, something has actually changed. Schedule the update. */
-  if (o != RTE_VALID_OR_NULL(new_best))
-    rt_schedule_nhu(dst);
+    RT_LOCKED(dst_pub, dst)
+    {
+      /* No need to inspect it further if recalculation is already scheduled */
+      if ((dst->nhu_state == NHU_SCHEDULED) || (dst->nhu_state == NHU_DIRTY))
+	break;
 
+      /* Irrelevant prefix */
+      if (!trie_match_net(dst->flowspec_trie, n))
+	break;
+
+      /* Conflate following updates */
+      const rte *old = RTE_VALID_OR_NULL(u->update->old);
+      const rte *new = u->update->new;
+      for (
+	  SKIP_BACK_DECLARE(struct rt_pending_export, rpe, it, u->update);
+	  rpe = atomic_load_explicit(&rpe->next, memory_order_acquire) ;)
+      {
+	ASSERT_DIE(new == rpe->it.old);
+	new = rpe->it.new;
+	rt_export_processed(&ln->req, rpe->it.seq);
+      }
+
+      /* Ignore idempotent */
+      if ((old == new) || old && new && rte_same(old, new))
+	break;
+
+      /* Actually, schedule NHU */
+      rt_schedule_nhu(dst);
+    }
+
+    if (!task_still_in_limit())
+      return ev_send_loop(dst_pub->loop, &ln->event);
   }
 }
 
@@ -2560,25 +2415,19 @@ rt_flowspec_dump_req(struct rt_export_request *req)
   debug("  Flowspec link for table %s (%p)\n", ln->dst->name, req);
 }
 
-static void
-rt_flowspec_log_state_change(struct rt_export_request *req, u8 state)
-{
-  SKIP_BACK_DECLARE(struct rt_flowspec_link, ln, req, req);
-  rt_trace(ln->dst, D_STATES, "Flowspec link from %s export state changed to %s",
-      ln->src->name, rt_export_state_name(state));
-}
-
 static struct rt_flowspec_link *
 rt_flowspec_find_link(struct rtable_private *src, rtable *dst)
 {
   WALK_TLIST(rt_flowspec_link, ln, &src->flowspec_links)
-    if (ln->dst == dst && ln->req.hook)
-      switch (atomic_load_explicit(&ln->req.hook->export_state, memory_order_acquire))
+    if (ln->dst == dst)
+      switch (rt_export_get_state(&ln->req))
       {
-	case TES_HUNGRY:
 	case TES_FEEDING:
 	case TES_READY:
 	  return ln;
+
+	default:
+	  bug("Unexpected flowspec link state");
       }
 
   return NULL;
@@ -2606,16 +2455,21 @@ rt_flowspec_link(rtable *src_pub, rtable *dst_pub)
       ln->dst = dst_pub;
       ln->req = (struct rt_export_request) {
 	.name = mb_sprintf(p, "%s.flowspec.notifier", dst_pub->name),
-	.list = birdloop_event_list(dst_pub->loop),
+	.r = {
+	  .event = &ln->event,
+	  .target = birdloop_event_list(dst_pub->loop),
+	},
 	.pool = p,
 	.trace_routes = src->config->debug,
-	.dump_req = rt_flowspec_dump_req,
-	.log_state_change = rt_flowspec_log_state_change,
-	.export_one = rt_flowspec_export_one,
+	.dump = rt_flowspec_dump_req,
+      };
+      ln->event = (event) {
+	.hook = rt_flowspec_export,
+	.data = ln,
       };
       rt_flowspec_link_add_tail(&src->flowspec_links, ln);
 
-      rt_table_export_start_locked(src, &ln->req);
+      rt_export_subscribe(&src->export_best, &ln->req);
 
       lock_dst = 1;
     }
@@ -2629,21 +2483,13 @@ rt_flowspec_link(rtable *src_pub, rtable *dst_pub)
   birdloop_leave(dst_pub->loop);
 }
 
-static void
-rt_flowspec_link_stopped(struct rt_export_request *req)
-{
-  SKIP_BACK_DECLARE(struct rt_flowspec_link, ln, req, req);
-  rtable *dst = ln->dst;
-
-  mb_free(ln);
-  rt_unlock_table(dst);
-}
-
 void
 rt_flowspec_unlink(rtable *src, rtable *dst)
 {
   birdloop_enter(dst->loop);
 
+  _Bool unlock_dst = 0;
+
   struct rt_flowspec_link *ln;
   RT_LOCKED(src, t)
   {
@@ -2654,11 +2500,16 @@ rt_flowspec_unlink(rtable *src, rtable *dst)
     if (!--ln->uc)
     {
       rt_flowspec_link_rem_node(&t->flowspec_links, ln);
-      ln->req.hook->stopped = rt_flowspec_link_stopped;
-      rt_stop_export_locked(t, ln->req.hook);
+      rt_export_unsubscribe(&ln->req);
+      ev_postpone(&ln->event);
+      mb_free(ln);
+      unlock_dst = 1;
     }
   }
 
+  if (unlock_dst)
+    rt_unlock_table(dst);
+
   birdloop_leave(dst->loop);
 }
 
@@ -2696,19 +2547,22 @@ rt_free(resource *_r)
 }
 
 static void
-rt_res_dump(resource *_r, unsigned indent)
+rt_res_dump(resource *_r, unsigned indent UNUSED)
 {
   SKIP_BACK_DECLARE(struct rtable_private, r, r, _r);
 
   debug("name \"%s\", addr_type=%s, rt_count=%u, use_count=%d\n",
       r->name, net_label[r->addr_type], r->rt_count, r->use_count);
 
+#if 0
+  /* TODO: rethink this completely */
   /* TODO: move this to lfjour */
   char x[32];
   bsprintf(x, "%%%dspending export %%p\n", indent + 2);
 
   WALK_TLIST(lfjour_block, n, &r->journal.pending)
     debug(x, "", n);
+#endif
 }
 
 static struct resclass rt_class = {