Refactoring BGP export to use a common rt_exporter structure

Introducing a new omnipotent internal API to just pass route updates from whatever point wherever we want. In the next commit, we aim to convert the main routing tables to this API, while also implementing the rest of it (like feeding). For now, it's just refactoring of this code and structures out of BGP.
2024-05-11 16:54:54 +00:00 · 2024-05-02 11:39:34 +02:00
parent 0fff47aeea
commit 41fe5e659e
7 changed files with 464 additions and 188 deletions
--- a/nest/Makefile
+++ b/nest/Makefile
@@ -1,4 +1,4 @@
-src := cli.c cmds.c iface.c locks.c mpls.c neighbor.c password.c proto.c proto-build.c rt-attr.c rt-dev.c rt-fib.c rt-show.c rt-table.c
+src := cli.c cmds.c iface.c locks.c mpls.c neighbor.c password.c proto.c proto-build.c rt-attr.c rt-dev.c rt-export.c rt-fib.c rt-show.c rt-table.c
 obj := $(src-o-files)
 $(all-daemon)
 $(cf-local)
--- a/nest/route.h
+++ b/nest/route.h
@@ -74,6 +74,145 @@ struct rtable_config {
 					   route refresh is running */
 };

+/*
+ *	Route export journal
+ *
+ *	The journal itself is held in struct rt_exporter.
+ *	Workflow:
+ *	  (1) Initialize by rt_exporter_init()
+ *	  (2) Push data by rt_exporter_push() (the export item is copied)
+ *	  (3) Shutdown by rt_exporter_shutdown(), event is called after cleanup
+ *
+ *	Subscribers:
+ *	  (1) Initialize by rt_export_subscribe()
+ *	  (2a) Get data by rt_export_get();
+ *	  (2b) Release data after processing by rt_export_release()
+ *	  (3) Request refeed by rt_export_refeed()
+ *	  (4) Unsubscribe by rt_export_unsubscribe()
+ */
+
+struct rt_exporter {
+  struct lfjour journal;			/* Journal for update keeping */
+  void (*stopped)(struct rt_exporter *);	/* Callback when exporter can stop */
+  void (*cleanup_done)(struct rt_exporter *);	/* Callback when cleanup has been done */
+};
+
+struct rt_export_receiver {
+  /* Formal name */
+  char *name;
+
+  /* Memory and event list */
+  pool *pool;
+
+  /* State information */
+  enum rt_export_state {
+#define RT_EXPORT_STATES \
+    DOWN, \
+    HUNGRY, \
+    FEEDING, \
+    PARTIAL, \
+    READY, \
+    STOP, \
+
+#define RT_EXPORT_STATES_ENUM_HELPER(p)        TES_##p,
+    MACRO_FOREACH(RT_EXPORT_STATES_ENUM_HELPER, RT_EXPORT_STATES)
+    TES_MAX
+#undef RT_EXPORT_STATES_ENUM_HELPER
+  } _Atomic export_state;
+  btime last_state_change;
+
+  /* Prefiltering */
+  struct rt_prefilter {
+    /* Network prefilter mode (TE_ADDR_*) */
+    enum {
+      TE_ADDR_NONE = 0,                /* No address matching */
+      TE_ADDR_EQUAL,           /* Exact query - show route <addr> */
+      TE_ADDR_FOR,             /* Longest prefix match - show route for <addr> */
+      TE_ADDR_IN,              /* Interval query - show route in <addr> */
+      TE_ADDR_TRIE,            /* Query defined by trie */
+      TE_ADDR_HOOK,            /* Query processed by supplied custom hook */
+    } mode;
+
+    union {
+      const struct f_trie *trie;
+      const net_addr *addr;
+      int (*hook)(const struct rt_prefilter *, const net_addr *);
+    };
+  } prefilter;
+
+  /* Feeding */
+  u32 feed_index;
+  struct bmap feed_map;                /* Which _nets_ were already fed */
+  struct rt_feeding_request {
+    struct rt_feeding_request *next;           /* Next in request chain */
+    void (*done)(struct rt_feeding_request *); /* Called when this refeed finishes */
+    struct rt_prefilter prefilter;             /* Reload only matching nets */
+    PACKED enum {
+      RFRS_INACTIVE = 0,                       /* Inactive request */
+      RFRS_PENDING,                            /* Request enqueued, do not touch */
+      RFRS_RUNNING,                            /* Request active, do not touch */
+    } state;
+  } *feeding, *feed_pending;
+
+  /* Regular updates */
+  struct bmap seq_map;         /* Which lfjour items are already processed */
+  struct lfjour_recipient r;
+
+  /* Statistics */
+  struct rt_export_stats {
+    u32 updates_received;              /* Number of route updates received */
+    u32 withdraws_received;            /* Number of route withdraws received */
+  } stats;
+
+  /* Tracing */
+  u8 trace_routes;
+  void (*dump)(struct rt_export_receiver *req);
+};
+
+struct rt_export_union {
+  enum rt_export_kind {
+    RT_EXPORT_NONE = 0,
+    RT_EXPORT_FEED,
+    RT_EXPORT_UPDATE,
+    RT_EXPORT_STOP,
+  } kind;
+  union {
+    const struct rt_export_item {
+      LFJOUR_ITEM_INHERIT(li);		/* Member of lockfree journal */
+      char data[0];			/* Memcpy helper */
+      const rte *new, *old;		/* Route update */
+    } *update;
+    const struct rt_export_feed {
+      uint count_routes, count_exports;
+      const struct netindex *ni;
+      rte *block;
+      u64 *exports;
+      char data[0];
+    } *feed;
+  };
+};
+
+void rt_exporter_init(struct rt_exporter *, struct settle_config *);
+const struct rt_export_item *rt_exporter_push(struct rt_exporter *, const struct rt_export_item *);
+void rt_exporter_shutdown(struct rt_exporter *, void (*stopped)(struct rt_exporter *));
+
+void rt_export_subscribe(struct rt_exporter *, struct rt_export_receiver *);
+void rt_export_unsubscribe(struct rt_exporter *, struct rt_export_receiver *);
+
+const struct rt_export_union *rt_export_get(struct rt_export_receiver *);
+void rt_export_release(struct rt_export_receiver *, const struct rt_export_union *);
+void rt_export_processed(struct rt_export_receiver *, const struct rt_export_item *);
+void rt_export_refeed(struct rt_export_receiver *, struct rt_feeding_request *);
+static inline enum rt_export_state rt_export_get_state_r(struct rt_export_receiver *r)
+{ return atomic_load_explicit(&r->export_state, memory_order_acquire); }
+
+/*
+ *	The original rtable
+ *
+ *	To be kept as is for now until we refactor the new structures out of BGP Attrs.
+ */
+
+
 struct rt_export_hook;
 struct rt_export_request;
 struct rt_exporter;
@@ -234,23 +373,6 @@ struct rte_storage {

 /* Table-channel connections */

-struct rt_prefilter {
-  union {
-    const struct f_trie *trie;
-    const net_addr *addr;	/* Network prefilter address */
-    int (*hook)(const struct rt_prefilter *, const net_addr *);
-  };
-				/* Network prefilter mode (TE_ADDR_*) */
-  enum {
-    TE_ADDR_NONE = 0,		/* No address matching */
-    TE_ADDR_EQUAL,		/* Exact query - show route <addr> */
-    TE_ADDR_FOR,		/* Longest prefix match - show route for <addr> */
-    TE_ADDR_IN,			/* Interval query - show route in <addr> */
-    TE_ADDR_TRIE,		/* Query defined by trie */
-    TE_ADDR_HOOK,		/* Query processed by supplied custom hook */
-  } mode;
-} PACKED;
-
 struct rt_import_request {
  struct rt_import_hook *hook;		/* The table part of importer */
  char *name;
@@ -298,14 +420,6 @@ struct rt_pending_export {
  const rte *new, *new_best, *old, *old_best;
 };

-struct rt_export_feed {
-  uint count_routes, count_exports;
-  struct netindex *ni;
-  rte *block;
-  u64 *exports;
-  char data[0];
-};
-
 struct rt_export_request {
  struct rt_export_hook *hook;		/* Table part of the export */
  char *name;
@@ -356,11 +470,7 @@ struct rt_export_hook {

  struct rt_export_request *req;	/* The requestor */

-  struct rt_export_stats {
-    /* Export - from core to protocol */
-    u32 updates_received;		/* Number of route updates received */
-    u32 withdraws_received;		/* Number of route withdraws received */
-  } stats;
+  struct rt_export_stats stats;

  btime last_state_change;		/* Time of last state transition */

@@ -389,17 +499,6 @@ struct rt_export_hook {
 #define TIS_CLEARED	5
 #define TIS_MAX		6

-#define TES_DOWN	0
-#define TES_HUNGRY	1
-#define TES_FEEDING	2
-#define TES_READY	3
-#define TES_STOP	4
-#define TES_MAX		5
-
-
-#define TFT_FIB		1
-#define TFT_TRIE	2
-#define TFT_HASH	3

 void rt_request_import(rtable *tab, struct rt_import_request *req);
 void rt_request_export(rtable *tab, struct rt_export_request *req);
@@ -449,7 +548,6 @@ void rt_init_export(struct rt_exporter *re, struct rt_export_hook *hook);
 struct rt_export_hook *rt_alloc_export(struct rt_exporter *re, pool *pool, uint size);
 void rt_stop_export_common(struct rt_export_hook *hook);
 void rt_export_stopped(struct rt_export_hook *hook);
-void rt_exporter_init(struct rt_exporter *re);

 /*
 * Channel export hooks. To be refactored out.
--- a/nest/rt-export.c
+++ b/nest/rt-export.c
@@ -0,0 +1,219 @@
+/*
+ *	BIRD -- Route Export Mechanisms
+ *
+ *	(c) 2024       Maria Matejka <mq@jmq.cz>
+ *
+ *	Can be freely distributed and used under the terms of the GNU GPL.
+ */
+
+#include "nest/bird.h"
+#include "nest/route.h"
+
+_Thread_local static struct rt_export_union reu;
+
+static inline enum rt_export_state
+rt_export_change_state(struct rt_export_receiver *r, u32 expected_mask, enum rt_export_state state)
+{
+  r->last_state_change = current_time();
+  enum rt_export_state old = atomic_exchange_explicit(&r->export_state, state, memory_order_acq_rel);
+  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
+      );
+
+  return old;
+}
+
+const struct rt_export_union *
+rt_export_get(struct rt_export_receiver *r)
+{
+  switch (rt_export_get_state_r(r))
+  {
+    case TES_DOWN:
+      bug("Tried to get an export from a stopped receiver");
+
+    case TES_STOP:
+      reu.kind = RT_EXPORT_STOP;
+      return &reu;
+
+    case TES_PARTIAL:
+    case TES_FEEDING:
+      /* TODO: implement feeds */
+      bug("not implemented yet");
+
+    case TES_READY:
+      break;
+
+    case TES_HUNGRY:
+    case TES_MAX:
+      bug("invalid export state");
+  }
+
+  /* Process sequence number reset event */
+  if (lfjour_reset_seqno(&r->r))
+    bmap_reset(&r->seq_map, 4);
+
+  reu.update = SKIP_BACK(struct rt_export_item, li, lfjour_get(&r->r));
+
+  if (!reu.update)
+    return NULL;
+
+  reu.kind = RT_EXPORT_UPDATE;
+
+  if (bmap_test(&r->seq_map, reu.update->seq))
+  {
+//    log(L_DEBUG "seen, ignoring %p", reu.update);
+    rt_export_release(r, &reu);
+    return rt_export_get(r);
+  }
+  else
+  {
+//    log(L_DEBUG "getting %p", reu.update);
+    return &reu;
+  }
+}
+
+void
+rt_export_release(struct rt_export_receiver *r, const struct rt_export_union *u)
+{
+  ASSERT_DIE(u == &reu);
+
+  switch (reu.kind)
+  {
+    case RT_EXPORT_UPDATE:
+//      log(L_DEBUG "releaseing %p", u.update);
+      lfjour_release(&r->r);
+      reu.kind = RT_EXPORT_NONE;
+      return;
+
+    case RT_EXPORT_FEED:
+      bug("not implemented yet");
+
+    case RT_EXPORT_STOP:
+      /* Checking that we have indeed stopped the exporter */
+      ASSERT_DIE(rt_export_get_state_r(r) == TES_DOWN);
+      break;
+
+    default:
+      bug("strange export kind");
+  }
+}
+
+void
+rt_export_processed(struct rt_export_receiver *r, const struct rt_export_item *u)
+{
+//  log(L_DEBUG "processed %p", u);
+
+  /* Check sequence number reset event */
+  if (lfjour_reset_seqno(&r->r))
+    bmap_reset(&r->seq_map, 4);
+
+  ASSERT_DIE(!bmap_test(&r->seq_map, u->seq));
+  bmap_set(&r->seq_map, u->seq);
+}
+
+void
+rt_export_subscribe(struct rt_exporter *e, struct rt_export_receiver *r)
+{
+  rt_export_change_state(r, BIT32_ALL(TES_DOWN), TES_FEEDING);
+
+  lfjour_register(&e->journal, &r->r);
+
+  r->stats = (struct rt_export_stats) {};
+  r->last_state_change = current_time();
+  bmap_init(&r->seq_map, r->pool, 4);
+
+  /* We should init feeding but this is not implemented yet */
+  rt_export_change_state(r, BIT32_ALL(TES_FEEDING), TES_READY);
+}
+
+void
+rt_export_unsubscribe(struct rt_exporter *e UNUSED, struct rt_export_receiver *r)
+{
+  switch (rt_export_change_state(r, BIT32_ALL(TES_FEEDING, TES_PARTIAL, TES_READY, TES_STOP), TES_DOWN))
+  {
+    case TES_FEEDING:
+    case TES_PARTIAL:
+    case TES_READY:
+    case TES_STOP:
+      lfjour_unregister(&r->r);
+      break;
+    default:
+      bug("not implemented");
+  }
+}
+
+static void
+rt_exporter_cleanup_done(struct lfjour *j, u64 begin_seq UNUSED, u64 end_seq UNUSED)
+{
+  SKIP_BACK_DECLARE(struct rt_exporter, e, journal, j);
+
+  /* TODO: log the begin_seq / end_seq values */
+
+  CALL(e->cleanup_done, e);
+  if (e->stopped && (lfjour_count_recipients(j) == 0))
+  {
+    settle_cancel(&j->announce_timer);
+    ev_postpone(&j->cleanup_event);
+    e->stopped(e);
+  }
+}
+
+void
+rt_exporter_init(struct rt_exporter *e, struct settle_config *scf)
+{
+  e->journal.cleanup_done = rt_exporter_cleanup_done;
+  lfjour_init(&e->journal, scf);
+}
+
+const struct rt_export_item *
+rt_exporter_push(struct rt_exporter *e, const struct rt_export_item *uit)
+{
+  /* Get the object */
+  SKIP_BACK_DECLARE(struct rt_export_item, it, li, lfjour_push_prepare(&e->journal));
+
+  /* Copy the data, keeping the header */
+  memcpy(&it->data, &uit->data, e->journal.item_size - OFFSETOF(struct rt_export_item, data));
+
+  /* Commit the update */
+/*  log(L_DEBUG "pushing %p %016lx %016lx %016lx %016lx", it,
+      ((u64 *)it->data)[0],
+      ((u64 *)it->data)[1],
+      ((u64 *)it->data)[2],
+      ((u64 *)it->data)[3]);*/
+
+  lfjour_push_commit(&e->journal);
+
+  /* Return the update pointer */
+  return it;
+}
+
+void
+rt_exporter_shutdown(struct rt_exporter *e, void (*stopped)(struct rt_exporter *))
+{
+  /* We have to tell every receiver to stop */
+  _Bool done = 1;
+  WALK_TLIST(lfjour_recipient, r, &e->journal.recipients)
+  {
+    done = 0;
+    rt_export_change_state(
+	SKIP_BACK(struct rt_export_receiver, r, r),
+	BIT32_ALL(TES_FEEDING, TES_PARTIAL, TES_READY, TES_STOP),
+	TES_STOP);
+  }
+
+  /* The rest is done via the cleanup routine */
+  lfjour_do_cleanup_now(&e->journal);
+
+  if (done)
+  {
+    ev_postpone(&e->journal.cleanup_event);
+    settle_cancel(&e->journal.announce_timer);
+    CALL(stopped, e);
+  }
+  else
+//  e->stopped = stopped;
+    bug("not implemented yet");
+}
--- a/proto/bgp/attrs.c
+++ b/proto/bgp/attrs.c
@@ -1575,7 +1575,7 @@ bgp_finish_attrs(struct bgp_parse_state *s, ea_list **to)
 */

 /* bgp_bucket hash */
-#define RBH_KEY(b)		(b)->new ? (b)->new->rte.attrs : NULL
+#define RBH_KEY(b)		(b)->it.new ? (b)->it.new->attrs : NULL
 #define RBH_NEXT(b)		b->next_hash
 #define RBH_EQ(a1,a2)		((a1) == (a2))
 #define RBH_FN(a)		((a) ? ea_get_storage(a)->hash_key : 0)
@@ -1604,15 +1604,16 @@ bgp_announce_tx(struct bgp_channel *c, struct bgp_route *new, struct bgp_route *
  struct bgp_bucket *bo = HASH_FIND(c->bucket_hash, RBH, new->rte.attrs);

  /* Create the actual bucket */
-  SKIP_BACK_DECLARE(struct bgp_bucket, bn, li, lfjour_push_prepare(&c->tx_journal));
-  ASSERT_DIE(bn);
-
-  *bn = (struct bgp_bucket) {
-    .li = bn->li,  /* Keep the item's internal state */
-    .new = new,
-    .old = old,
+  struct bgp_bucket bnloc = {
+    .it = {
+      .new = new ? &new->rte : NULL,
+      .old = old ? &old->rte : NULL,
+    },
  };

+  /* Push the bucket */
+  SKIP_BACK_DECLARE(struct bgp_bucket, bn, it, rt_exporter_push(&c->tx_journal, &bnloc.it));
+
  if (bo)
  {
    /* Unlink the old bucket */
@@ -1631,14 +1632,11 @@ bgp_announce_tx(struct bgp_channel *c, struct bgp_route *new, struct bgp_route *

  BTX_TRACE(D_ROUTES, "announcing %N (%u/%u), bucket seq %lu, old %lu",
      new->rte.net, new->rte.src->global_id, old ? old->rte.src->global_id : 0,
-      bn->seq, bo ? bo->seq : 0ULL);
+      bn->it.seq, bo ? bo->it.seq : 0ULL);
  BTX_DEBUG("announcing %N (%u/%u): new %p/%p, old %p/%p, bucket %p (seq %lu), old %p (seq %lu)",
      new->rte.net, new->rte.src->global_id, old ? old->rte.src->global_id : 0,
      new, new->rte.attrs, old, old ? old->rte.attrs : NULL,
-      bn, bn->seq, bo, bo ? bo->seq : 0ULL);
-
-  /* Announce the TX journal item */
-  lfjour_push_commit(&c->tx_journal);
+      bn, bn->it.seq, bo, bo ? bo->it.seq : 0ULL);
 }

 /* Store new route */
@@ -1726,17 +1724,17 @@ bgp_new_export(struct bgp_channel *c, rte *r)
 }

 void
-bgp_withdraw_bucket(struct bgp_channel *c, struct bgp_bucket *b)
+bgp_withdraw_bucket(struct bgp_channel *c, const struct bgp_bucket *b)
 {
  /* Convert all routes with the given bucket to withdraws and mark them too long. */
  while (b)
  {
-    BTX_TRACE(D_ROUTES, "attrs too long for %N (src %u)", b->new->rte.net, b->new->rte.src);
-    if (!(b->new->rte.flags & REF_OBSOLETE))
+    BTX_TRACE(D_ROUTES, "attrs too long for %N (src %u)", b->it.new->net, b->it.new->src);
+    if (!(b->it.new->flags & REF_OBSOLETE))
    {
      rte r = {
-	.net = b->new->rte.net,
-	.src = b->new->rte.src,
+	.net = b->it.new->net,
+	.src = b->it.new->src,
 	.pflags = BGP_REF_TOO_LONG,
      };
      bgp_new_export(c, &r);
@@ -1783,31 +1781,30 @@ bgp_free_network(struct bgp_channel *c, struct bgp_network *n)
 }

 void
-bgp_bucket_sent(struct bgp_channel *c, struct bgp_bucket *b)
+bgp_bucket_sent(struct bgp_channel *c, const struct bgp_bucket *b)
 {
  /* If withdraw, move the pointer forward */
  if (b == c->withdraw_bucket)
    c->withdraw_bucket = b->next_attrs;

-  /* Mark in the bitmap */
-  ASSERT_DIE(!bmap_test(&c->tx_sent, b->seq));
-  bmap_set(&c->tx_sent, b->seq);
-
  /* If last, remove from the hash */
  if (!b->next_attrs)
    HASH_REMOVE2(c->bucket_hash, RBH, c->pool, b);
+
+  /* Mark done */
+  rt_export_processed(&c->tx_recipient, &b->it);
 }

 static void
 bgp_bucket_done_common(struct bgp_channel *c, struct bgp_bucket *b, struct bgp_network *n)
 {
-  ASSERT_DIE(b->new);
+  ASSERT_DIE(b->it.new);
  ASSERT_DIE(b != c->withdraw_bucket);

-  if (b->old)
+  if (b->it.old)
  {
-    ASSERT_DIE(b->old->rte.flags & REF_OBSOLETE);
-    bgp_free_route(c, n, b->old);
+    ASSERT_DIE(b->it.old->flags & REF_OBSOLETE);
+    bgp_free_route(c, n, SKIP_BACK(struct bgp_route, rte, b->it.old));
  }
 }

@@ -1816,35 +1813,35 @@ bgp_bucket_drop_new(struct bgp_channel *c, struct bgp_bucket *b, struct bgp_netw
 {
  /* No drop of an already obsolete route,
   * will be dropped from the old pointer in the future */
-  if (b->new->rte.flags & REF_OBSOLETE)
+  if (b->it.new->flags & REF_OBSOLETE)
    return;

  struct bgp_route **rp = &n->routes;
-  while ((*rp) && (*rp != b->new))
+  while ((*rp) && (&(*rp)->rte != b->it.new))
    rp = &((*rp)->next_net);

  ASSERT_DIE(*rp);
  *rp = (*rp)->next_net;
-  bgp_free_route(c, n, b->new);
+  bgp_free_route(c, n, SKIP_BACK(struct bgp_route, rte, b->it.new));
 }

 /* Bucket done with solid export table */
 static void
 bgp_bucket_done_keep(struct lfjour *j, struct lfjour_item *i)
 {
-  SKIP_BACK_DECLARE(struct bgp_channel, c, tx_journal, j);
-  SKIP_BACK_DECLARE(struct bgp_bucket, b, li, i);
-  struct netindex *ni = NET_TO_INDEX(b->new->rte.net);
+  SKIP_BACK_DECLARE(struct bgp_channel, c, tx_journal.journal, j);
+  SKIP_BACK_DECLARE(struct bgp_bucket, b, it.li, i);
+  struct netindex *ni = NET_TO_INDEX(b->it.new->net);
  struct bgp_network *n = HASH_FIND(c->network_hash, PXH, ni);

  /* TX resend -> no need to free a route */
-  if (b->new && b->old && (b->new == b->old))
+  if (b->it.new && b->it.old && (b->it.new == b->it.old))
    return;

  bgp_bucket_done_common(c, b, n);

  /* Dropping withdrawal route placeholder */
-  if (!b->new->rte.attrs)
+  if (!b->it.new->attrs)
    bgp_bucket_drop_new(c, b, n);
 }

@@ -1852,9 +1849,9 @@ bgp_bucket_done_keep(struct lfjour *j, struct lfjour_item *i)
 static void
 bgp_bucket_done_drop(struct lfjour *j, struct lfjour_item *i)
 {
-  SKIP_BACK_DECLARE(struct bgp_channel, c, tx_journal, j);
-  SKIP_BACK_DECLARE(struct bgp_bucket, b, li, i);
-  struct netindex *ni = NET_TO_INDEX(b->new->rte.net);
+  SKIP_BACK_DECLARE(struct bgp_channel, c, tx_journal.journal, j);
+  SKIP_BACK_DECLARE(struct bgp_bucket, b, it.li, i);
+  struct netindex *ni = NET_TO_INDEX(b->it.new->net);
  struct bgp_network *n = HASH_FIND(c->network_hash, PXH, ni);

  bgp_bucket_done_common(c, b, n);
@@ -1883,14 +1880,13 @@ bgp_channel_init_tx(struct bgp_channel *c)
  /* Initialize data structures */
  HASH_INIT(c->network_hash, c->pool, 8);
  HASH_INIT(c->bucket_hash, c->pool, 8);
-  bmap_init(&c->tx_sent, c->pool, 4);

  /* Setup journal */
-  c->tx_journal.loop = c->c.proto->loop;
-  c->tx_journal.domain = NULL;
-  c->tx_journal.item_size = sizeof(struct bgp_bucket);
-  c->tx_journal.item_done = (c->keep_export = c->cf->export_table) ? bgp_bucket_done_keep : bgp_bucket_done_drop;
-  lfjour_init(&c->tx_journal, &c->cf->tx_settle);
+  c->tx_journal.journal.loop = c->c.proto->loop;
+  c->tx_journal.journal.domain = NULL;
+  c->tx_journal.journal.item_size = sizeof(struct bgp_bucket);
+  c->tx_journal.journal.item_done = (c->keep_export = c->cf->export_table) ? bgp_bucket_done_keep : bgp_bucket_done_drop;
+  rt_exporter_init(&c->tx_journal, &c->cf->tx_settle);
  BTX_TRACE(D_EVENTS, "inited");

  /* Setup internal recipient */
@@ -1898,33 +1894,31 @@ bgp_channel_init_tx(struct bgp_channel *c)
    .hook = bgp_export_kick,
    .data = c,
  };
-  c->tx_recipient = (struct lfjour_recipient) {
-    .event = &c->tx_kick_event,
-    .target = birdloop_event_list(bgp->p.loop),
+  c->tx_recipient = (struct rt_export_receiver) {
+    .pool = c->pool,
+    .r = {
+      .event = &c->tx_kick_event,
+      .target = birdloop_event_list(bgp->p.loop),
+    },
  };
-  lfjour_register(&c->tx_journal, &c->tx_recipient);
+  rt_export_subscribe(&c->tx_journal, &c->tx_recipient);
 }

 static void
 bgp_channel_flush_tx_journal(struct bgp_channel *c)
 {
-  for (struct lfjour_item *it;
-      it = lfjour_get(&c->tx_recipient);
-      lfjour_release(&c->tx_recipient))
+  for (const struct rt_export_union *u;
+      u = rt_export_get(&c->tx_recipient);
+      rt_export_release(&c->tx_recipient, u))
  {
-    /* Process sequence number reset event */
-    if (lfjour_reset_seqno(&c->tx_recipient))
-    {
-      BTX_DEBUG("seqno reset");
-      bmap_reset(&c->tx_sent, 4);
-    }
+    ASSERT_DIE(u->kind == RT_EXPORT_UPDATE);

    /* Mark done */
-    BTX_DEBUG("flushing bucket seqno %lu", it->seq);
-    bgp_bucket_sent(c, SKIP_BACK(struct bgp_bucket, li, it));
+    BTX_DEBUG("flushing bucket seqno %lu", u->update->seq);
+    bgp_bucket_sent(c, SKIP_BACK(struct bgp_bucket, it, u->update));
  }

-  lfjour_do_cleanup_now(&c->tx_journal);
+  lfjour_do_cleanup_now(&c->tx_journal.journal);
 }

 void
@@ -1937,7 +1931,7 @@ bgp_channel_flush_tx(struct bgp_channel *c)
  ASSERT_DIE(!c->withdraw_bucket);

  /* How many readers do we have now? */
-  uint rcnt = lfjour_count_recipients(&c->tx_journal);
+  uint rcnt = lfjour_count_recipients(&c->tx_journal.journal);
  ASSERT_DIE(rcnt >= 1);

  if (rcnt > 1)
@@ -1992,7 +1986,7 @@ bgp_channel_flush_tx(struct bgp_channel *c)
    c->tx_walk_active = 0;
    HASH_MAY_RESIZE_DOWN(c->network_hash, PXH, c->pool);

-    ASSERT_DIE(!lfjour_pending_items(&c->tx_journal));
+    ASSERT_DIE(!lfjour_pending_items(&c->tx_journal.journal));
    ASSERT_DIE(!c->withdraw_bucket);
    HASH_WALK(c->bucket_hash, next_hash, b)
      bug("Found a stale bucket %p", b);
@@ -2012,17 +2006,9 @@ bgp_channel_resend_tx(struct bgp_channel *c)
 }

 static void
-bgp_check_export_table_stopped(struct lfjour *j, u64 begin UNUSED, u64 end)
+bgp_channel_tx_stopped(struct rt_exporter *e)
 {
-  if (!EMPTY_TLIST(lfjour_recipient, &j->recipients))
-    return;
-
-  ASSERT_DIE(!~end);
-
-  settle_cancel(&j->announce_timer);
-  ev_postpone(&j->cleanup_event);
-
-  SKIP_BACK_DECLARE(struct bgp_channel, c, tx_journal, j);
+  SKIP_BACK_DECLARE(struct bgp_channel, c, tx_journal, e);

  c->tx_walk_active = 1;
  HASH_WALK_DELSAFE(c->network_hash, next_hash, n)
@@ -2041,9 +2027,10 @@ bgp_channel_stop_tx(struct bgp_channel *c)
   * bucket hash can stay, it will be eventually just freed */
  c->withdraw_bucket = NULL;

+  /* Stop the whole exporter */
  channel_add_obstacle(&c->c);
-  c->tx_journal.cleanup_done = bgp_check_export_table_stopped;
-  lfjour_unregister(&c->tx_recipient);
+  rt_export_unsubscribe(&c->tx_journal, &c->tx_recipient);
+  rt_exporter_shutdown(&c->tx_journal, bgp_channel_tx_stopped);
 }


--- a/proto/bgp/bgp.c
+++ b/proto/bgp/bgp.c
@@ -2504,8 +2504,6 @@ bgp_channel_reconfigure(struct channel *C, struct channel_config *CC, int *impor
      (new->aigp_originate != old->aigp_originate))
    *export_changed = 1;

-  c->tx_settle.cf = new->tx_settle;
-
  c->cf = new;
  return 1;
 }
--- a/proto/bgp/bgp.h
+++ b/proto/bgp/bgp.h
@@ -405,15 +405,12 @@ struct bgp_channel {
  HASH(struct bgp_network) network_hash;/* Network hash for export table */
  HASH(struct bgp_bucket) bucket_hash;	/* Hash table of last route buckets */

-  struct bgp_bucket *withdraw_bucket;	/* First withdraw bucket for direct access */
+  const struct bgp_bucket *withdraw_bucket;	/* First withdraw bucket for direct access */

-  struct lfjour tx_journal;		/* Changes waiting to be sent */
-  struct settle tx_settle;		/* TX settle timer to not send everything immediately */
-  struct lfjour_recipient tx_recipient;	/* Internal recipient for tx_journal */
+  struct rt_exporter tx_journal;	/* Changes waiting to be sent */
+  struct rt_export_receiver tx_recipient;/* Internal recipient for tx_journal */
  event tx_kick_event;			/* Update propagator */

-  struct bmap tx_sent;			/* Sequence numbers of sent TX changes */
-
  struct bgp_tx_stats {
    u64 routes_prepared;		/* Number of routes ever prepared for sending */
    u64 routes_skipped;			/* Number of routes ever skipped because of newer version exists */
@@ -466,11 +463,9 @@ struct bgp_route {
 };

 struct bgp_bucket {
-  LFJOUR_ITEM_INHERIT(li);
-  struct bgp_route *new;
-  struct bgp_route *old;
-  struct bgp_bucket *next_hash;
-  struct bgp_bucket *next_attrs;
+  struct rt_export_item it;		/* Common export inheritance */
+  struct bgp_bucket *next_hash;		/* Next in bgp_bucket hash chain */
+  const struct bgp_bucket *next_attrs;	/* Next update with the same attributes */
 };

 #define BTX_TRACE_CONDITION(category)		((c->c.debug | c->c.proto->debug) & (category))
@@ -671,8 +666,8 @@ void bgp_channel_stop_tx(struct bgp_channel *c);
 void bgp_channel_flush_tx(struct bgp_channel *c);
 void bgp_channel_resend_tx(struct bgp_channel *c);

-void bgp_bucket_sent(struct bgp_channel *c, struct bgp_bucket *b);
-void bgp_withdraw_bucket(struct bgp_channel *c, struct bgp_bucket *b);
+void bgp_bucket_sent(struct bgp_channel *c, const struct bgp_bucket *b);
+void bgp_withdraw_bucket(struct bgp_channel *c, const struct bgp_bucket *b);

 int bgp_rte_better(const rte *, const rte *);
 int bgp_rte_mergable(const rte *pri, const rte *sec);
--- a/proto/bgp/packets.c
+++ b/proto/bgp/packets.c
@@ -2258,13 +2258,13 @@ bgp_get_af_desc(u32 afi)
 }

 static inline uint
-bgp_encode_nlri(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, byte *end)
+bgp_encode_nlri(struct bgp_write_state *s, const struct bgp_bucket *buck, byte *buf, byte *end)
 {
  ASSERT_DIE(buck);

  uint (*encode)(struct bgp_write_state *s, struct bgp_route *rte, byte *buf, uint size) = s->channel->desc->encode_nlri;
  struct bgp_channel *c = s->channel;
-  ea_list *attrs = buck->new->rte.attrs;
+  ea_list *attrs = buck->it.new->attrs;

 #define EIMAXLEN 32
  u64 ei[EIMAXLEN];
@@ -2291,14 +2291,13 @@ bgp_encode_nlri(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, b
  while (buck)
  {
    /* Check consistency */
-    struct bgp_route *rte = buck->new;
+    SKIP_BACK_DECLARE(struct bgp_route, rte, rte, buck->it.new);
    ASSERT_DIE(rte->rte.attrs == attrs);
-    ASSERT_DIE(!bmap_test(&s->channel->tx_sent, buck->seq));

    /* Check obsolescence */
    if (!(rte->rte.flags & REF_OBSOLETE))
    {
-      BTX_DEBUG("encoding NLRI %N of bucket %p seqno %lu", rte->rte.net, buck, buck->seq);
+      BTX_DEBUG("encoding NLRI %N of bucket %p seqno %lu", rte->rte.net, buck, buck->it.seq);

      /* Actually do the encoding */
      uint sz = encode(s, rte, pos, end - pos);
@@ -2307,12 +2306,12 @@ bgp_encode_nlri(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, b
      if (!sz)
 	goto done;

-      ENL_PUSH(buck->seq, 0);
+      ENL_PUSH(buck->it.seq, 0);
    }
    else
    {
-      BTX_DEBUG("not encoding obsolete NLRI %N of bucket %p seqno %lu", rte->rte.net, buck, buck->seq);
-      ENL_PUSH(buck->seq | ENL_IGNORED, 0);
+      BTX_DEBUG("not encoding obsolete NLRI %N of bucket %p seqno %lu", rte->rte.net, buck, buck->it.seq);
+      ENL_PUSH(buck->it.seq | ENL_IGNORED, 0);
    }

    /* Mark bucket done */
@@ -2342,7 +2341,7 @@ bgp_update_next_hop(struct bgp_export_state *s, eattr *a, ea_list **to)
 #define MAX_ATTRS_LENGTH (end-buf+BGP_HEADER_LENGTH - 1024)

 static byte *
-bgp_create_ip_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, byte *end)
+bgp_create_ip_reach(struct bgp_write_state *s, const struct bgp_bucket *buck, byte *buf, byte *end)
 {
  /*
   *	2 B	Withdrawn Routes Length (zero)
@@ -2352,11 +2351,11 @@ bgp_create_ip_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *bu
   *	var	IPv4 Network Layer Reachability Information
   */

-  ASSERT_DIE(buck->new->rte.attrs);
+  ASSERT_DIE(buck->it.new->attrs);

  int lr, la;

-  la = bgp_encode_attrs(s, buck->new->rte.attrs, buf+4, buf + MAX_ATTRS_LENGTH);
+  la = bgp_encode_attrs(s, buck->it.new->attrs, buf+4, buf + MAX_ATTRS_LENGTH);
  if (la < 0)
  {
    /* Attribute list too long */
@@ -2376,9 +2375,9 @@ bgp_create_ip_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *bu
 }

 static byte *
-bgp_create_mp_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, byte *end)
+bgp_create_mp_reach(struct bgp_write_state *s, const struct bgp_bucket *buck, byte *buf, byte *end)
 {
-  ASSERT_DIE(buck->new->rte.attrs);
+  ASSERT_DIE(buck->it.new->attrs);

  /*
   *	2 B	IPv4 Withdrawn Routes Length (zero)
@@ -2408,7 +2407,7 @@ bgp_create_mp_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *bu

  /* Encode attributes to temporary buffer */
  byte *abuf = alloca(MAX_ATTRS_LENGTH);
-  la = bgp_encode_attrs(s, buck->new->rte.attrs, abuf, abuf + MAX_ATTRS_LENGTH);
+  la = bgp_encode_attrs(s, buck->it.new->attrs, abuf, abuf + MAX_ATTRS_LENGTH);
  if (la < 0)
  {
    /* Attribute list too long */
@@ -2448,7 +2447,7 @@ bgp_create_mp_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *bu
 #undef MAX_ATTRS_LENGTH

 static byte *
-bgp_create_ip_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, byte *end)
+bgp_create_ip_unreach(struct bgp_write_state *s, const struct bgp_bucket *buck, byte *buf, byte *end)
 {
  /*
   *	2 B	Withdrawn Routes Length
@@ -2458,7 +2457,7 @@ bgp_create_ip_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *
   *	---	IPv4 Network Layer Reachability Information (unused)
   */

-  ASSERT_DIE(!buck->new->rte.attrs);
+  ASSERT_DIE(!buck->it.new->attrs);
  uint len = bgp_encode_nlri(s, buck, buf+2, end);
  if (!len)
    return NULL;
@@ -2470,7 +2469,7 @@ bgp_create_ip_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *
 }

 static byte *
-bgp_create_mp_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, byte *end)
+bgp_create_mp_unreach(struct bgp_write_state *s, const struct bgp_bucket *buck, byte *buf, byte *end)
 {
  /*
   *	2 B	Withdrawn Routes Length (zero)
@@ -2485,7 +2484,7 @@ bgp_create_mp_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *
   *	---	IPv4 Network Layer Reachability Information (unused)
   */

-  ASSERT_DIE(!buck->new->rte.attrs);
+  ASSERT_DIE(!buck->it.new->attrs);
  uint len = bgp_encode_nlri(s, buck, buf+11, end);
  if (!len)
    return NULL;
@@ -2506,7 +2505,7 @@ bgp_create_mp_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *
 #ifdef CONFIG_BMP

 static byte *
-bgp_create_update_bmp(struct bgp_channel *c, byte *buf, struct bgp_bucket *buck, bool update)
+bgp_create_update_bmp(struct bgp_channel *c, byte *buf, const struct bgp_bucket *buck, bool update)
 {
  struct bgp_proto *p = (void *) c->c.proto;
  byte *end = buf + (BGP_MAX_EXT_MSG_LENGTH - BGP_HEADER_LENGTH);
@@ -2597,7 +2596,7 @@ static byte *
 bgp_create_update(struct bgp_channel *c, byte *buf)
 {
  struct bgp_proto *p = (void *) c->c.proto;
-  struct bgp_bucket *buck;
+  const struct bgp_bucket *buck;
  byte *end = buf + (bgp_max_packet_length(p->conn) - BGP_HEADER_LENGTH);
  byte *res = NULL;

@@ -2615,13 +2614,6 @@ bgp_create_update(struct bgp_channel *c, byte *buf)
  /* Try unreachable bucket */
  if (buck = c->withdraw_bucket)
  {
-    /* Process sequence number reset event */
-    if (lfjour_reset_seqno(&c->tx_recipient))
-    {
-      BTX_DEBUG("seqno reset");
-      bmap_reset(&c->tx_sent, 4);
-    }
-
    /* Encode the packet */
    BTX_DEBUG("found a withdraw bucket");
    res = (c->afi == BGP_AF_IPV4) && !c->ext_next_hop ?
@@ -2633,47 +2625,34 @@ bgp_create_update(struct bgp_channel *c, byte *buf)
  while (!res)
  {
    /* Anything to encode? */
-    struct lfjour_item *it = lfjour_get(&c->tx_recipient);
-    if (!it)
+    const struct rt_export_union *u = rt_export_get(&c->tx_recipient);
+    if (!u)
    {
      BTX_TRACE(D_STATES, "everything sent");
      return NULL;
    }

-    /* Process sequence number reset event */
-    if (lfjour_reset_seqno(&c->tx_recipient))
+    SKIP_BACK_DECLARE(const struct bgp_bucket, buck, it, u->update);
+    if (buck->it.new->stale_cycle != c->tx_stale_cycle)
    {
-      BTX_DEBUG("seqno reset");
-      bmap_reset(&c->tx_sent, 4);
+      BTX_DEBUG("ignoring obsolete bucket %p (seq %lu, stale %u vs. %u), attrs %p",
+	  buck, buck->it.seq, buck->it.new->stale_cycle, c->tx_stale_cycle, buck->it.new->attrs);
    }
-
-    /* Not encoded yet? */
-    if (bmap_test(&c->tx_sent, it->seq))
-      BTX_DEBUG("not encoding bucket %p (seq %lu), already done", it, it->seq);
    else
-    {
-      SKIP_BACK_DECLARE(struct bgp_bucket, buck, li, it);
-      if (buck->new->rte.stale_cycle != c->tx_stale_cycle)
+      TMP_SAVED
      {
-	BTX_DEBUG("ignoring obsolete bucket %p (seq %lu, stale %u vs. %u), attrs %p",
-	    buck, it->seq, buck->new->rte.stale_cycle, c->tx_stale_cycle, buck->new->rte.attrs);
+	BTX_DEBUG("encoding bucket %p (seq %lu), attrs %p", buck, buck->it.seq, buck->it.new->attrs);
+
+	/* Try to encode the bucket */
+	res = !s.mp_reach ?
+	  bgp_create_ip_reach(&s, buck, buf, end):
+	  bgp_create_mp_reach(&s, buck, buf, end);
+
+	BTX_DEBUG("encoded to %p", res);
      }
-      else
-	TMP_SAVED
-	{
-	  BTX_DEBUG("encoding bucket %p (seq %lu), attrs %p", buck, it->seq, buck->new->rte.attrs);
-
-	  /* Try to encode the bucket */
-	  res = !s.mp_reach ?
-	    bgp_create_ip_reach(&s, buck, buf, end):
-	    bgp_create_mp_reach(&s, buck, buf, end);
-
-	  BTX_DEBUG("encoded to %p", res);
-	}
-    }

    /* Release the item */
-    lfjour_release(&c->tx_recipient);
+    rt_export_release(&c->tx_recipient, u);
  }

  /* Done! */