BMP: Use generic channel feed instead of direct walk over rtable

Now we use rt_notify() and channels for both feed and notifications,
in both import tables (pre-policy) and regular tables (post-policy).

Remove direct walk in bmp_route_monitor_snapshot().

nest/proto.c

@@ -451,10 +451,7 @@ channel_start_export(struct channel *c)
   ASSERT(c->channel_state == CS_UP);
   ASSERT(c->export_state == ES_DOWN);
 
-  if (!c->bmp_hack)
-    channel_schedule_feed(c, 1);	/* Sets ES_FEEDING */
-  else
-    c->export_state = ES_READY;
+  channel_schedule_feed(c, 1);	/* Sets ES_FEEDING */
 }
 
 static void

nest/protocol.h

@@ -500,7 +500,7 @@ struct channel_config {
   u8 merge_limit;			/* Maximal number of nexthops for RA_MERGED */
   u8 in_keep_filtered;			/* Routes rejected in import filter are kept */
   u8 rpki_reload;			/* RPKI changes trigger channel reload */
-  u8 bmp_hack;				/* No feed, no flush */
+  u8 bmp_hack;				/* No flush */
 };
 
 struct channel {
@@ -553,7 +553,7 @@ struct channel {
   u8 reload_pending;			/* Reloading and another reload is scheduled */
   u8 refeed_pending;			/* Refeeding and another refeed is scheduled */
   u8 rpki_reload;			/* RPKI changes trigger channel reload */
-  u8 bmp_hack;				/* No feed, no flush */
+  u8 bmp_hack;				/* No flush */
 
   struct rtable *out_table;		/* Internal table for exported routes */
 

proto/bmp/bmp.c

@@ -223,6 +223,8 @@ bmp_send_peer_up_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
   const byte *tx_data, const size_t tx_data_size,
   const byte *rx_data, const size_t rx_data_size);
 
+static void bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs);
+
 // Stores necessary any data in list
 struct bmp_data_node {
   node n;
@@ -237,9 +239,6 @@ struct bmp_data_node {
   bool policy;
 };
 
-static void
-bmp_route_monitor_snapshot(struct bmp_proto *p, struct bmp_stream *bs);
-
 static void
 bmp_common_hdr_serialize(buffer *stream, const enum bmp_message_type type, const u32 data_size)
 {
@@ -571,6 +570,7 @@ bmp_add_stream(struct bmp_proto *p, struct bmp_peer *bp, u32 afi, bool policy, s
   bmp_lock_table(p, bs->table);
 
   bs->sender = sender;
+  bs->sync = false;
 
   return bs;
 }
@@ -634,7 +634,7 @@ bmp_remove_peer(struct bmp_proto *p, struct bmp_peer *bp)
 }
 
 static void
-bmp_peer_up_(struct bmp_proto *p, struct bgp_proto *bgp,
+bmp_peer_up_(struct bmp_proto *p, struct bgp_proto *bgp, bool sync,
 	    const byte *tx_open_msg, uint tx_open_length,
 	    const byte *rx_open_msg, uint rx_open_length)
 {
@@ -651,9 +651,20 @@ bmp_peer_up_(struct bmp_proto *p, struct bgp_proto *bgp,
 
   bmp_send_peer_up_notif_msg(p, bgp, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
 
-  struct bmp_stream *bs;
-  WALK_LIST(bs, bp->streams)
-    bmp_route_monitor_snapshot(p, bs);
+  /*
+   * We asssume peer_up() notifications are received before any route
+   * notifications from that peer. Therefore, peers established after BMP
+   * session coould be considered synced with empty RIB.
+   */
+  if (sync)
+  {
+    struct bmp_stream *bs;
+    WALK_LIST(bs, bp->streams)
+    {
+      bmp_route_monitor_end_of_rib(p, bs);
+      bs->sync = true;
+    }
+  }
 }
 
 void
@@ -663,7 +674,7 @@ bmp_peer_up(struct bgp_proto *bgp,
 {
   struct bmp_proto *p; node *n;
   WALK_LIST2(p, n, bmp_proto_list, bmp_node)
-      bmp_peer_up_(p, bgp, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
+    bmp_peer_up_(p, bgp, true, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
 }
 
 static void
@@ -675,7 +686,7 @@ bmp_peer_init(struct bmp_proto *p, struct bgp_proto *bgp)
       !conn->local_open_msg || !conn->remote_open_msg)
     return;
 
-  bmp_peer_up_(p, bgp, conn->local_open_msg, conn->local_open_length,
+  bmp_peer_up_(p, bgp, false, conn->local_open_msg, conn->local_open_length,
 	       conn->remote_open_msg, conn->remote_open_length);
 }
 
@@ -856,25 +867,6 @@ bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs)
   bmp_route_monitor_put_update(p, bs, rx_end_payload, pos - rx_end_payload, current_real_time());
 }
 
-static void
-bmp_route_monitor_snapshot(struct bmp_proto *p, struct bmp_stream *bs)
-{
-  struct rtable *tab = bs->table->table;
-
-  struct fib_iterator fit = {};
-  FIB_ITERATE_INIT(&fit, &tab->fib);
-  FIB_ITERATE_START(&tab->fib, &fit, net, n)
-  {
-    rte *e;
-    for (e = n->routes; e; e = e->next)
-      if (e->sender == &bs->sender->c)
-	bmp_route_monitor_notify(p, bs, n->n.addr, e, e->src);
-  }
-  FIB_ITERATE_END;
-
-  bmp_route_monitor_end_of_rib(p, bs);
-}
-
 static void
 bmp_send_peer_down_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
   const byte *data, const size_t data_size)
@@ -1003,6 +995,34 @@ bmp_rt_notify(struct proto *P, struct channel *c, struct network *net,
   bmp_route_monitor_notify(p, bs, net->n.addr, new, (new ?: old)->src);
 }
 
+static void
+bmp_feed_end(struct channel *c)
+{
+  struct bmp_proto *p = (void *) c->proto;
+
+  struct bmp_table *bt = bmp_find_table(p, c->table);
+  if (!bt)
+    return;
+
+  /*
+   * Unsynced streams are added in one moment during BMP session establishment,
+   * therefore we can assume that all unsynced streams (for given channel)
+   * already received full feed now and are synced.
+   *
+   * TODO: Use more efficent way to find bmp_stream from bmp_table
+   */
+
+  HASH_WALK(p->stream_map, next, bs)
+  {
+    if ((bs->table == bt) && !bs->sync)
+    {
+      bmp_route_monitor_end_of_rib(p, bs);
+      bs->sync = true;
+    }
+  }
+  HASH_WALK_END;
+}
+
 
 /**
  * bmp_startup - enter established state
@@ -1181,6 +1201,7 @@ bmp_init(struct proto_config *CF)
 
   P->rt_notify = bmp_rt_notify;
   P->preexport = bmp_preexport;
+  P->feed_end = bmp_feed_end;
 
   p->cf = cf;
   p->local_addr = cf->local_addr;

proto/bmp/bmp.h

@@ -87,6 +87,7 @@ struct bmp_stream {
   node n;
   struct bgp_proto *bgp;
   u32 key;
+  bool sync;
   struct bmp_stream *next;
   struct bmp_table *table;
   struct bgp_channel *sender;