BMP: updated to v3 API

configure.ac

@@ -320,7 +320,7 @@ else
   AC_DEFINE([HAVE_CLOCK_MONOTONIC_COARSE], [1], [Define to 1 if coarse clock is available])
 fi
 
-all_protocols="aggregator bfd babel bgp l3vpn ospf pipe radv rip rpki static mrt"
+all_protocols="aggregator bfd babel bgp l3vpn ospf pipe radv rip rpki static mrt bmp"
 all_protocols=`echo $all_protocols | sed 's/ /,/g'`
 
 if test "$with_protocols" = all ; then

nest/proto.c

@@ -3054,6 +3054,13 @@ proto_states_subscribe(struct lfjour_recipient *r)
     lfjour_register(&proto_state_table_pub.journal, r);
 }
 
+void
+proto_states_unsubscribe(struct lfjour_recipient *r)
+{
+  PST_LOCKED(ts)
+    lfjour_unregister(r);
+}
+
 /* State attribute declarations */
 struct ea_class ea_name = {
   .name = "proto_name",

nest/protocol.h

@@ -448,6 +448,7 @@ void proto_announce_state_later(struct proto *p, ea_list *attr);
 ea_list *channel_get_state(int id);
 ea_list *proto_get_state(int id);
 void proto_states_subscribe(struct lfjour_recipient *r);
+void proto_states_unsubscribe(struct lfjour_recipient *r);
 
 /* Protocol journal attributes */
 extern struct ea_class ea_name, ea_protocol_name, ea_protocol_type,

proto/bgp/attrs.c

@@ -1921,7 +1921,10 @@ bgp_done_prefix(struct bgp_ptx_private *c, struct bgp_prefix *px, struct bgp_buc
 {
   /* BMP hack */
   if (buck->bmp)
+  {
+    rem_node(&px->buck_node);
     return;
+  }
 
   /* Cleanup: We're called from bucket senders. */
   ASSERT_DIE(px->cur == buck);
@@ -1996,7 +1999,7 @@ bgp_out_item_done(struct lfjour *j UNUSED, struct lfjour_item *i UNUSED)
 {}
 
 static struct rt_export_feed *
-bgp_out_feed_net(struct rt_exporter *e, struct rcu_unwinder *u, u32 index, bool (*prefilter)(struct rt_export_feeder *, const net_addr *), struct rt_export_feeder *f, const struct rt_export_item *_first)
+bgp_out_feed_net(struct rt_exporter *e, struct rcu_unwinder *u, u32 index, bool (*prefilter)(struct rt_export_feeder *, const net_addr *), struct rt_export_feeder *f, UNUSED const struct rt_export_item *_first)
 {
   ASSERT_DIE(u == NULL);
   SKIP_BACK_DECLARE(struct bgp_ptx_private, c, exporter, e);

proto/bgp/bgp.h

@@ -460,6 +460,8 @@ struct bgp_ptx_private {
 
   slab *prefix_slab;			/* Slab holding prefix nodes */
   slab *bucket_slab;			/* Slab holding buckets to send */
+
+  char bmp;                            /* This is a fake ptx for BMP encoding */
 };
 
 typedef union bgp_ptx {
@@ -701,7 +703,7 @@ static inline struct bgp_proto *bgp_rte_proto(const rte *rte)
     SKIP_BACK(struct bgp_proto, p.sources, rte->src->owner) : NULL;
 }
 
-byte * bgp_bmp_encode_rte(struct bgp_channel *c, byte *buf, const net_addr *n, const struct rte *new, const struct rte_src *src);
+byte * bgp_bmp_encode_rte(ea_list *c, struct bgp_proto *bgp_p, byte *buf, const struct rte *new);
 
 #define BGP_AIGP_METRIC		1
 #define BGP_AIGP_MAX		U64(0xffffffffffffffff)
@@ -747,6 +749,7 @@ void bgp_log_error(struct bgp_proto *p, u8 class, char *msg, unsigned code, unsi
 
 void bgp_update_next_hop(struct bgp_export_state *s, eattr *a, ea_list **to);
 byte *bgp_create_end_mark_(struct bgp_channel *c, byte *buf);
+byte *bgp_create_end_mark_ea_(ea_list *c, byte *buf);
 
 
 /* Packet types */

proto/bgp/packets.c

@@ -2357,10 +2357,11 @@ bgp_create_ip_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *bu
   return buf+4+la+lr;
 }
 
+
 static byte *
 bgp_create_mp_reach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *buf, byte *end)
 {
-  ASSERT_DIE(s->ptx->withdraw_bucket != buck);
+  ASSERT_DIE((s->ptx->bmp) || (s->ptx->withdraw_bucket != buck));
 
   /*
    *	2 B	IPv4 Withdrawn Routes Length (zero)
@@ -2479,24 +2480,35 @@ 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(ea_list *channel_ea, struct bgp_proto *bgp_p, byte *buf, struct bgp_bucket *buck, bool update)
 {
-  struct bgp_proto *p = (void *) c->c.proto;
+  struct bgp_channel *c;
+  u32 c_id = ea_get_int(channel_ea, &ea_channel_id, 0);
+  BGP_WALK_CHANNELS(bgp_p, c)
+    if (c->c.id == c_id)
+      break;
+
   byte *end = buf + (BGP_MAX_EXT_MSG_LENGTH - BGP_HEADER_LENGTH);
   byte *res = NULL;
   /* FIXME: must be a bit shorter */
 
-  struct bgp_caps *peer = p->conn->remote_caps;
+  struct bgp_caps *peer = bgp_p->conn->remote_caps;
   const struct bgp_af_caps *rem = bgp_find_af_caps(peer, c->afi);
 
+  struct bgp_ptx_private ptx = {
+    .bmp = 1,
+    .c = c,
+  };
+
   struct bgp_write_state s = {
-    .proto = p,
-    .channel = c,
+    .proto = bgp_p,
+    .ptx = &ptx,
     .pool = tmp_linpool,
     .mp_reach = (c->afi != BGP_AF_IPV4) || rem->ext_next_hop,
     .as4_session = 1,
     .add_path = c->add_path_rx,
     .mpls = c->desc->mpls,
+    .ignore_non_bgp_attrs = 1,
   };
 
   if (!update)
@@ -2526,34 +2538,31 @@ bgp_bmp_prepare_bgp_hdr(byte *buf, const u16 msg_size, const u8 msg_type)
 }
 
 byte *
-bgp_bmp_encode_rte(struct bgp_channel *c, byte *buf, const net_addr *n,
-		   const struct rte *new, const struct rte_src *src)
+bgp_bmp_encode_rte(ea_list *c, struct bgp_proto *bgp_p, byte *buf, const struct rte *new)
 {
-//  struct bgp_proto *p = (void *) c->c.proto;
   byte *pkt = buf + BGP_HEADER_LENGTH;
 
-  ea_list *attrs = new ? new->attrs : NULL;
-  uint ea_size = new ? (sizeof(ea_list) + attrs->count * sizeof(eattr)) : 0;
-  uint bucket_size = sizeof(struct bgp_bucket) + ea_size;
-  uint prefix_size = sizeof(struct bgp_prefix) + n->length;
+  uint ea_size = new->attrs ? (sizeof(ea_list) + new->attrs->count * sizeof(eattr)) : 0;
+  uint prefix_size = sizeof(struct bgp_prefix) + new->net->length;
 
   struct lp_state *tmpp = lp_save(tmp_linpool);
 
   /* Temporary bucket */
-  struct bgp_bucket *b = tmp_allocz(bucket_size);
+  struct bgp_bucket *b = tmp_allocz(sizeof(struct bgp_bucket) + ea_size);
   b->bmp = 1;
   init_list(&b->prefixes);
 
-  if (attrs)
-    memcpy(b->eattrs, attrs, ea_size);
+  if (new->attrs)
+    memcpy(b->eattrs, new->attrs, ea_size);
 
   /* Temporary prefix */
   struct bgp_prefix *px = tmp_allocz(prefix_size);
-  px->path_id = (u32) src->private_id;
-  net_copy(px->net, n);
-  add_tail(&b->prefixes, &px->buck_node_xx);
+  px->src = tmp_allocz(sizeof(struct rte_src));
+  memcpy(px->src, new->src, sizeof(struct rte_src));
+  px->ni = NET_TO_INDEX(new->net);
+  add_tail(&b->prefixes, &px->buck_node);
 
-  byte *end = bgp_create_update_bmp(c, pkt, b, !!new);
+  byte *end = bgp_create_update_bmp(c, bgp_p, pkt, b, !!new->attrs);
 
   if (end)
     bgp_bmp_prepare_bgp_hdr(buf, end - buf, PKT_UPDATE);
@@ -2661,6 +2670,31 @@ bgp_create_mp_end_mark(struct bgp_channel *c, byte *buf)
   return buf+10;
 }
 
+static byte *
+bgp_create_mp_end_mark_ea(ea_list *c, byte *buf)
+{
+  put_u16(buf+0, 0);
+  put_u16(buf+2, 6);		/* length 4--9 */
+
+  /* Empty MP_UNREACH_NLRI atribute */
+  buf[4] = BAF_OPTIONAL;
+  buf[5] = BA_MP_UNREACH_NLRI;
+  buf[6] = 3;			/* Length 7--9 */
+  int afi = ea_get_int(c, &ea_bgp_afi, 0);
+  put_af3(buf+7, afi);
+
+  return buf+10;
+}
+
+byte *
+bgp_create_end_mark_ea_(ea_list *c, byte *buf)
+{
+  int afi = ea_get_int(c, &ea_bgp_afi, 0);
+  return (afi == BGP_AF_IPV4) ?
+    bgp_create_ip_end_mark(NULL, buf):
+    bgp_create_mp_end_mark_ea(c, buf);
+}
+
 byte *
 bgp_create_end_mark_(struct bgp_channel *c, byte *buf)
 {

proto/bmp/bmp.c

@@ -48,17 +48,16 @@
 #include "filter/filter.h"
 #include "proto/bgp/bgp.h"
 #include "sysdep/unix/unix.h"
+#include "sysdep/unix/io-loop.h"
 #include "lib/event.h"
 #include "lib/ip.h"
 #include "lib/lists.h"
 #include "lib/resource.h"
 #include "lib/unaligned.h"
+#include "lib/tlists.h"
 #include "nest/iface.h"
 #include "nest/route.h"
 
-// List of BMP instances
-static list STATIC_LIST_INIT(bmp_proto_list);
-
 #define HASH_PEER_KEY(n)		n->bgp
 #define HASH_PEER_NEXT(n)		n->next
 #define HASH_PEER_EQ(b1,b2)		b1 == b2
@@ -213,15 +212,19 @@ enum bmp_term_reason {
     IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(!(p), msg, rv);	\
   } while (0)
 
+static const struct ea_class *bgp_next_hop_ea_class = NULL;
 
 static void bmp_connected(struct birdsock *sk);
 static void bmp_sock_err(sock *sk, int err);
 static void bmp_close_socket(struct bmp_proto *p);
+static void bmp_check_routes(void *bt_);
+static void bmp_feed_end(struct rt_export_request *req);
+static void bmp_process_proto_state_change(struct bmp_proto *p, struct lfjour_item *last_up);
+static void bmp_proto_state_changed(void *_p);
 
 static void
-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);
+bmp_send_peer_up_notif_msg(struct bmp_proto *p, ea_list *bgp,
+    const adata *tx_data, const adata *rx_data, struct bgp_conn_sk_ad *sk);
 
 static void bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs);
 
@@ -279,8 +282,8 @@ bmp_schedule_tx_packet(struct bmp_proto *p, const byte *payload, const size_t si
 {
   ASSERT(p->started);
 
-  struct bmp_data_node *tx_data = mb_alloc(p->tx_mem_pool, sizeof (struct bmp_data_node));
-  tx_data->data = mb_alloc(p->tx_mem_pool, size);
+  struct bmp_data_node *tx_data = mb_allocz(p->tx_mem_pool, sizeof (struct bmp_data_node));
+  tx_data->data = mb_allocz(p->tx_mem_pool, size);
   memcpy(tx_data->data, payload, size);
   tx_data->data_size = size;
   add_tail(&p->tx_queue, &tx_data->n);
@@ -288,7 +291,7 @@ bmp_schedule_tx_packet(struct bmp_proto *p, const byte *payload, const size_t si
   if (sk_tx_buffer_empty(p->sk)
       && !ev_active(p->tx_ev))
   {
-    ev_schedule(p->tx_ev);
+    ev_send_loop(p->p.loop, p->tx_ev);
   }
 }
 
@@ -331,7 +334,7 @@ bmp_fire_tx(void *p_)
     {
       if (!ev_active(p->tx_ev))
       {
-        ev_schedule(p->tx_ev);
+        ev_send_loop(p->p.loop, p->tx_ev);
       }
 
       return;
@@ -441,12 +444,11 @@ static void
 bmp_peer_up_notif_msg_serialize(buffer *stream, const bool is_peer_global,
   const u32 peer_as, const u32 peer_bgp_id, const bool as4_support,
   const ip_addr local_addr, const ip_addr remote_addr, const u16 local_port,
-  const u16 remote_port, const byte *sent_msg, const size_t sent_msg_length,
-  const byte *recv_msg, const size_t recv_msg_length)
+  const u16 remote_port, const adata *sent_msg, const adata *recv_msg)
 {
   const size_t data_size =
     BMP_PER_PEER_HDR_SIZE + BMP_PEER_UP_NOTIF_MSG_FIX_SIZE +
-    BGP_HEADER_LENGTH + sent_msg_length + BGP_HEADER_LENGTH + recv_msg_length;
+    BGP_HEADER_LENGTH + sent_msg->length + BGP_HEADER_LENGTH + recv_msg->length;
 
   bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
   bmp_common_hdr_serialize(stream, BMP_PEER_UP_NOTIF, data_size);
@@ -456,10 +458,10 @@ bmp_peer_up_notif_msg_serialize(buffer *stream, const bool is_peer_global,
   bmp_put_ipa(stream, local_addr);
   bmp_put_u16(stream, local_port);
   bmp_put_u16(stream, remote_port);
-  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + sent_msg_length);
-  bmp_put_data(stream, sent_msg, sent_msg_length);
-  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + recv_msg_length);
-  bmp_put_data(stream, recv_msg, recv_msg_length);
+  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + sent_msg->length);
+  bmp_put_data(stream, sent_msg->data, sent_msg->length);
+  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + recv_msg->length);
+  bmp_put_data(stream, recv_msg->data, recv_msg->length);
 }
 
 static void
@@ -482,42 +484,60 @@ bmp_peer_down_notif_msg_serialize(buffer *stream, const bool is_peer_global,
  */
 
 static struct bmp_table *
-bmp_find_table(struct bmp_proto *p, struct rtable *tab)
+bmp_find_table(struct bmp_proto *p, rtable *tab)
 {
   return HASH_FIND(p->table_map, HASH_TABLE, tab);
 }
 
+const struct channel_class channel_bmp = {
+  .channel_size =	sizeof(struct channel),
+  .config_size =	sizeof(struct channel_config),
+  /*.init =		
+  .start =		
+  .shutdown =		
+  .cleanup =		
+  .reconfigure =	*/
+};
+
 static struct bmp_table *
-bmp_add_table(struct bmp_proto *p, struct rtable *tab)
+bmp_add_table(struct bmp_proto *p, rtable *tab)
 {
   struct bmp_table *bt = mb_allocz(p->p.pool, sizeof(struct bmp_table));
   bt->table = tab;
+  bt->p = p;
   rt_lock_table(bt->table);
 
   HASH_INSERT(p->table_map, HASH_TABLE, bt);
 
-  struct channel_config cc = {
-    .name = "monitor",
-    .channel = &channel_basic,
-    .table = tab->config,
-    .in_filter = FILTER_REJECT,
-    .net_type = tab->addr_type,
-    .ra_mode = RA_ANY,
-    .bmp_hack = 1,
+  bt->event.data = bt;
+
+  bt->event.hook = bmp_check_routes;
+  bt->out_req = (struct rt_export_request) {
+    .name = mb_sprintf(p->p.pool, "%s.export", p->p.name),
+    .r = (struct lfjour_recipient) {
+      .target = proto_event_list(&p->p),
+      .event = &bt->event,
+    },
+    .pool = p->p.pool,
+    .trace_routes = p->p.debug,
+    //.dump = channel_dump_export_req, TODO: this will crash on `dump tables` from CLI
+    .fed = bmp_feed_end,
   };
 
-  bt->channel = proto_add_channel(&p->p, &cc);
-  channel_set_state(bt->channel, CS_UP);
-
+  rt_export_subscribe(tab, all, &bt->out_req);
   return bt;
 }
 
+
 static void
 bmp_remove_table(struct bmp_proto *p, struct bmp_table *bt)
 {
-  channel_set_state(bt->channel, CS_FLUSHING);
+  if (bt->channel)
+  {
+    channel_set_state(bt->channel, CS_STOP);
     channel_set_state(bt->channel, CS_DOWN);
-  proto_remove_channel(&p->p, bt->channel);
+  }
+    rt_export_unsubscribe(all, &bt->out_req);
 
   HASH_REMOVE(p->table_map, HASH_TABLE, bt);
 
@@ -527,14 +547,39 @@ bmp_remove_table(struct bmp_proto *p, struct bmp_table *bt)
   mb_free(bt);
 }
 
-static inline struct bmp_table *bmp_get_table(struct bmp_proto *p, struct rtable *tab)
-{ return bmp_find_table(p, tab) ?: bmp_add_table(p, tab); }
-
 static inline void bmp_lock_table(struct bmp_proto *p UNUSED, struct bmp_table *bt)
 { bt->uc++; }
 
+struct bmp_table *
+bmp_get_table(struct bmp_proto *p, rtable *tab)
+{
+  struct bmp_table *bt = bmp_find_table(p, tab);
+  if (bt)
+  {
+   while (true) {
+      atomic_int i = bt->uc;
+      if (i == 0)
+      {
+        struct bmp_table *new = bmp_add_table(p, tab);
+        bmp_lock_table(p, new);
+        return new;
+      }
+      if (atomic_compare_exchange_strong_explicit(&bt->uc, &i, i+1, memory_order_acq_rel, memory_order_relaxed))
+        return bt;
+    }
+  }
+  struct bmp_table *new = bmp_add_table(p, tab);
+  bmp_lock_table(p, new);
+  return new;
+}
+
 static inline void bmp_unlock_table(struct bmp_proto *p, struct bmp_table *bt)
-{ bt->uc--; if (!bt->uc) bmp_remove_table(p, bt); }
+{ atomic_int i = 1;
+  if (atomic_compare_exchange_strong_explicit(&bt->uc, &i, 0, memory_order_acq_rel, memory_order_relaxed))
+    bmp_remove_table(p, bt);
+  else
+    bt->uc--;
+}
 
 
 /*
@@ -544,20 +589,29 @@ static inline void bmp_unlock_table(struct bmp_proto *p, struct bmp_table *bt)
 static inline u32 bmp_stream_key(u32 afi, bool policy)
 { return afi ^ (policy ? BMP_STREAM_KEY_POLICY : 0); }
 
-static inline u32 bmp_stream_afi(struct bmp_stream *bs)
-{ return bs->key & ~BMP_STREAM_KEY_POLICY; }
-
 static inline bool bmp_stream_policy(struct bmp_stream *bs)
 { return !!(bs->key & BMP_STREAM_KEY_POLICY); }
 
 static struct bmp_stream *
 bmp_find_stream(struct bmp_proto *p, const struct bgp_proto *bgp, u32 afi, bool policy)
 {
-  return HASH_FIND(p->stream_map, HASH_STREAM, bgp, bmp_stream_key(afi, policy));
+  ea_list *bgp_attr = proto_get_state(bgp->p.id);
+  struct bmp_stream *s = HASH_FIND(p->stream_map, HASH_STREAM, bgp_attr, bmp_stream_key(afi, policy));
+
+  while (s == NULL)
+  {
+    struct lfjour_item *li = lfjour_get(&p->proto_state_reader);
+    if (!li)
+      return NULL;
+
+    bmp_process_proto_state_change(p, li);
+    s = HASH_FIND(p->stream_map, HASH_STREAM, bgp_attr, bmp_stream_key(afi, policy));
+  }
+  return s;
 }
 
 static struct bmp_stream *
-bmp_add_stream(struct bmp_proto *p, struct bmp_peer *bp, u32 afi, bool policy, struct rtable *tab, struct bgp_channel *sender)
+bmp_add_stream(struct bmp_proto *p, struct bmp_peer *bp, u32 afi, bool policy, rtable *tab, ea_list *sender, int in_pre_policy)
 {
   struct bmp_stream *bs = mb_allocz(p->p.pool, sizeof(struct bmp_stream));
   bs->bgp = bp->bgp;
@@ -567,10 +621,10 @@ bmp_add_stream(struct bmp_proto *p, struct bmp_peer *bp, u32 afi, bool policy, s
   HASH_INSERT(p->stream_map, HASH_STREAM, bs);
 
   bs->table = bmp_get_table(p, tab);
-  bmp_lock_table(p, bs->table);
 
   bs->sender = sender;
   bs->sync = false;
+  bs->in_pre_policy = in_pre_policy;
 
   return bs;
 }
@@ -593,29 +647,58 @@ bmp_remove_stream(struct bmp_proto *p, struct bmp_stream *bs)
  */
 
 static struct bmp_peer *
-bmp_find_peer(struct bmp_proto *p, const struct bgp_proto *bgp)
+bmp_find_peer(struct bmp_proto *p, ea_list *bgp_attr)
 {
-  return HASH_FIND(p->peer_map, HASH_PEER, bgp);
+  return HASH_FIND(p->peer_map, HASH_PEER, bgp_attr);
 }
 
 static struct bmp_peer *
-bmp_add_peer(struct bmp_proto *p, struct bgp_proto *bgp)
+bmp_add_peer(struct bmp_proto *p, ea_list *bgp_attr)
 {
-  struct bmp_peer *bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
-  bp->bgp = bgp;
+  struct bmp_peer *bp;
+  if (DG_IS_LOCKED(p->p.pool->domain))
+    bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
+  else
+  {
+    DG_LOCK(p->p.pool->domain);
+    bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
+    DG_UNLOCK(p->p.pool->domain);
+  }
+  bp->bgp = bgp_attr;
 
   init_list(&bp->streams);
 
   HASH_INSERT(p->peer_map, HASH_PEER, bp);
 
-  struct bgp_channel *c;
-  BGP_WALK_CHANNELS(bgp, c)
-  {
-    if (p->monitoring_rib.in_pre_policy && c->c.in_table)
-      bmp_add_stream(p, bp, c->afi, false, c->c.in_table, c);
+  const adata *channels_adata = ea_get_adata(bgp_attr, &ea_proto_channel_list);
+  int id_count = channels_adata->length / sizeof(u32);
+  u32 *chann_ids = (u32 *) channels_adata->data;
 
-    if (p->monitoring_rib.in_post_policy && c->c.table)
-      bmp_add_stream(p, bp, c->afi, true, c->c.table, c);
+  for (int i = 0; i < id_count; i++)
+  {
+    ea_list *chan_attr;
+    PST_LOCKED(ts)
+      chan_attr = ts->channels[chann_ids[i]];
+
+    if (chan_attr == NULL)
+      continue;
+
+    rtable *ch_table = (rtable *) ea_get_ptr(chan_attr, &ea_rtable, 0);
+    int in_keep = ea_get_int(chan_attr, &ea_in_keep, 0);
+
+    if (p->monitoring_rib.in_pre_policy && ch_table)
+    {
+      if (in_keep == RIK_PREFILTER)
+        bmp_add_stream(p, bp, ea_get_int(chan_attr, &ea_bgp_afi, 0), false, ch_table, chan_attr, 1);
+      else
+        log(L_WARN "%s: Failed to request pre-policy for %s.%s, import table disabled",
+	    p->p.name,
+	    ea_get_adata(bgp_attr, &ea_name)->data,
+	    ea_get_adata(chan_attr, &ea_name)->data);
+    }
+
+    if (p->monitoring_rib.in_post_policy && ch_table)
+      bmp_add_stream(p, bp, ea_get_int(chan_attr, &ea_bgp_afi, 0), true, ch_table, chan_attr, 0);
   }
 
   return bp;
@@ -634,22 +717,23 @@ bmp_remove_peer(struct bmp_proto *p, struct bmp_peer *bp)
 }
 
 static void
-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)
+bmp_peer_up_(struct bmp_proto *p, ea_list *bgp_attr, bool sync,
+	     const adata *tx_open_msg, const adata *rx_open_msg,
+	     struct bgp_conn_sk_ad *sk)
 {
   if (!p->started)
     return;
 
-  struct bmp_peer *bp = bmp_find_peer(p, bgp);
+  struct bmp_peer *bp = bmp_find_peer(p, bgp_attr);
   if (bp)
     return;
 
-  TRACE(D_STATES, "Peer up for %s", bgp->p.name);
+  const char *name = ea_get_adata(bgp_attr, &ea_name)->data;
+  TRACE(D_STATES, "Peer up for %s", name);
 
-  bp = bmp_add_peer(p, bgp);
+  bp = bmp_add_peer(p, bgp_attr);
 
-  bmp_send_peer_up_notif_msg(p, bgp, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
+  bmp_send_peer_up_notif_msg(p, bgp_attr, tx_open_msg, rx_open_msg, sk);
 
   /*
    * We asssume peer_up() notifications are received before any route
@@ -667,115 +751,68 @@ bmp_peer_up_(struct bmp_proto *p, struct bgp_proto *bgp, bool sync,
   }
 }
 
-void
-bmp_peer_up(struct bgp_proto *bgp,
-	    const byte *tx_open_msg, uint tx_open_length,
-	    const byte *rx_open_msg, uint rx_open_length)
+static bool
+bmp_peer_up_inout(struct bmp_proto *p, ea_list *bgp_attr, bool sync)
 {
-  struct bmp_proto *p; node *n;
-  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
-    bmp_peer_up_(p, bgp, true, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
-}
+  int in_state = ea_get_int(bgp_attr, &ea_bgp_in_conn_state, 0);
+  int out_state = ea_get_int(bgp_attr, &ea_bgp_out_conn_state, 0);
 
-static void
-bmp_peer_init(struct bmp_proto *p, struct bgp_proto *bgp)
-{
-  struct bgp_conn *conn = bgp->conn;
-
-  if (!conn || (conn->state != BS_ESTABLISHED) ||
-      !conn->local_open_msg || !conn->remote_open_msg)
-    return;
-
-  bmp_peer_up_(p, bgp, false, conn->local_open_msg, conn->local_open_length,
-	       conn->remote_open_msg, conn->remote_open_length);
-}
-
-
-
-static const struct birdsock *
-bmp_get_birdsock(const struct bgp_proto *bgp)
-{
-  if (bgp->conn && bgp->conn->sk)
-    return bgp->conn->sk;
-
-  return NULL;
-}
-
-static const struct birdsock *
-bmp_get_birdsock_ext(const struct bgp_proto *bgp)
-{
-  const struct birdsock *sk = bmp_get_birdsock(bgp);
-  if (sk != NULL)
-    return sk;
-
-  if (bgp->incoming_conn.sk)
+  if (in_state == BS_ESTABLISHED)
   {
-    sk = bgp->incoming_conn.sk;
-  }
-  else if (bgp->outgoing_conn.sk)
-  {
-    sk = bgp->outgoing_conn.sk;
+    ASSERT_DIE(out_state != BS_ESTABLISHED);
+
+    const adata *loc_open = ea_get_adata(bgp_attr, &ea_bgp_in_conn_local_open_msg);
+    const adata *rem_open = ea_get_adata(bgp_attr, &ea_bgp_in_conn_remote_open_msg);
+    SKIP_BACK_DECLARE(struct bgp_conn_sk_ad, sk, ad, ea_get_adata(bgp_attr, &ea_bgp_in_conn_sk));
+
+    ASSERT_DIE(loc_open && rem_open);
+    bmp_peer_up_(p, bgp_attr, sync, loc_open, rem_open, sk);
+
+    return true;
   }
 
-  return sk;
-}
-
-static const struct bgp_caps *
-bmp_get_bgp_remote_caps(const struct bgp_proto *bgp)
-{
-  if (bgp->conn && bgp->conn->remote_caps)
-    return bgp->conn->remote_caps;
-
-  return NULL;
-}
-
-static const struct bgp_caps *
-bmp_get_bgp_remote_caps_ext(const struct bgp_proto *bgp)
-{
-  const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps(bgp);
-  if (remote_caps != NULL)
-    return remote_caps;
-
-  if (bgp->incoming_conn.remote_caps)
+  if (out_state == BS_ESTABLISHED)
   {
-    remote_caps = bgp->incoming_conn.remote_caps;
-  }
-  else if (bgp->outgoing_conn.remote_caps)
-  {
-    remote_caps = bgp->outgoing_conn.remote_caps;
+    const adata *loc_open = ea_get_adata(bgp_attr, &ea_bgp_out_conn_local_open_msg);
+    const adata *rem_open = ea_get_adata(bgp_attr, &ea_bgp_out_conn_remote_open_msg);
+    SKIP_BACK_DECLARE(struct bgp_conn_sk_ad, sk, ad, ea_get_adata(bgp_attr, &ea_bgp_out_conn_sk));
+
+    ASSERT_DIE(loc_open && rem_open);
+    bmp_peer_up_(p, bgp_attr, sync, loc_open, rem_open, sk);
+
+    return true;
   }
 
-  return remote_caps;
+  return false;
 }
 
 static bool
-bmp_is_peer_global_instance(const struct bgp_proto *bgp)
+bmp_is_peer_global_instance(ea_list *bgp)
 {
-  return (bgp->cf->peer_type != BGP_PT_EXTERNAL &&
-            bgp->cf->peer_type != BGP_PT_INTERNAL)
-              ? (bgp->local_as != bgp->remote_as)
-              : (bgp->cf->peer_type == BGP_PT_EXTERNAL);
+  int peer_type = ea_get_int(bgp, &ea_bgp_peer_type, 0);
+  int local_as = ea_get_int(bgp, &ea_bgp_loc_as, 0);
+  int remote_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
+
+  return (peer_type != BGP_PT_EXTERNAL &&
+            peer_type != BGP_PT_INTERNAL)
+              ? (local_as != remote_as)
+              : (peer_type == BGP_PT_EXTERNAL);
 }
 
 static void
-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)
+bmp_send_peer_up_notif_msg(struct bmp_proto *p, ea_list *bgp,
+    const adata *tx_data, const adata *rx_data, struct bgp_conn_sk_ad *sk)
 {
   ASSERT(p->started);
 
-  const struct birdsock *sk = bmp_get_birdsock_ext(bgp);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    sk,
-    "[BMP] No BGP socket"
-  );
-
+  const int rem_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
+  const int rem_id = ea_get_int(bgp, &ea_bgp_rem_id, 0);
   const bool is_global_instance_peer = bmp_is_peer_global_instance(bgp);
   buffer payload = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
+
   bmp_peer_up_notif_msg_serialize(&payload, is_global_instance_peer,
-    bgp->remote_as, bgp->remote_id, 1,
-    sk->saddr, sk->daddr, sk->sport, sk->dport, tx_data, tx_data_size,
-    rx_data, rx_data_size);
+    rem_as, rem_id, 1,
+    sk->saddr, sk->daddr, sk->sport, sk->dport, tx_data, rx_data);
   bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
   bmp_buffer_free(&payload);
 }
@@ -783,41 +820,50 @@ bmp_send_peer_up_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
 static void
 bmp_route_monitor_put_update(struct bmp_proto *p, struct bmp_stream *bs, const byte *data, size_t length, btime timestamp)
 {
-  struct bmp_data_node *upd_msg = mb_alloc(p->update_msg_mem_pool,
+  struct bmp_data_node *upd_msg = mb_allocz(p->update_msg_mem_pool,
                                sizeof (struct bmp_data_node));
   upd_msg->data = mb_alloc(p->update_msg_mem_pool, length);
   memcpy(upd_msg->data, data, length);
   upd_msg->data_size = length;
+
   add_tail(&p->update_msg_queue, &upd_msg->n);
 
   /* Save some metadata */
-  struct bgp_proto *bgp = bs->bgp;
-  upd_msg->remote_as = bgp->remote_as;
-  upd_msg->remote_id = bgp->remote_id;
-  upd_msg->remote_ip = bgp->remote_ip;
+  ea_list *bgp = bs->bgp;
+  upd_msg->remote_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
+  upd_msg->remote_id = ea_get_int(bgp, &ea_bgp_rem_id, 0);
+  upd_msg->remote_ip = ea_get_ip(bgp, &ea_bgp_rem_ip, IPA_NONE);
   upd_msg->timestamp = timestamp;
   upd_msg->global_peer = bmp_is_peer_global_instance(bgp);
   upd_msg->policy = bmp_stream_policy(bs);
 
   /* Kick the commit */
   if (!ev_active(p->update_ev))
-    ev_schedule(p->update_ev);
+    ev_send_loop(p->p.loop, p->update_ev);
 }
 
 static void
-bmp_route_monitor_notify(struct bmp_proto *p, struct bmp_stream *bs,
-			 const net_addr *n, const struct rte *new, const struct rte_src *src)
+bmp_route_monitor_notify(struct bmp_proto *p, struct bgp_proto *bgp_p, u32 afi, bool policy, const rte *new, ea_list *old)
 {
-  byte buf[BGP_MAX_EXT_MSG_LENGTH];
-  byte *end = bgp_bmp_encode_rte(bs->sender, buf, n, new, src);
+  /* Idempotent update */
+  if ((old == new->attrs) || old && new->attrs && ea_same(old, new->attrs))
+    return;
 
-  btime delta_t = new ? current_time() - new->lastmod : 0;
+  /* No stream, probably flushed already */
+  struct bmp_stream *bs = bmp_find_stream(p, bgp_p, afi, policy);
+  if (!bs)
+    return;
+
+  byte buf[BGP_MAX_EXT_MSG_LENGTH];
+  byte *end = bgp_bmp_encode_rte(bs->sender, bgp_p, buf, new);
+
+  btime delta_t = new->attrs ? current_time() - new->lastmod : 0;
   btime timestamp = current_real_time() - delta_t;
 
   if (end)
     bmp_route_monitor_put_update(p, bs, buf, end - buf, timestamp);
   else
-    log(L_WARN "%s: Cannot encode update for %N", p->p.name, n);
+    log(L_WARN "%s: Cannot encode update for %N", p->p.name, new->net);
 }
 
 static void
@@ -855,10 +901,10 @@ bmp_route_monitor_commit(void *p_)
 static void
 bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs)
 {
-  TRACE(D_PACKETS, "Sending END-OF-RIB for %s.%s", bs->bgp->p.name, bs->sender->c.name);
+  TRACE(D_PACKETS, "Sending END-OF-RIB for %s.%s", ea_get_adata(bs->bgp, &ea_name)->data, ea_get_adata(bs->sender, &ea_name)->data);
 
   byte rx_end_payload[DEFAULT_MEM_BLOCK_SIZE];
-  byte *pos = bgp_create_end_mark_(bs->sender, rx_end_payload + BGP_HEADER_LENGTH);
+  byte *pos = bgp_create_end_mark_ea_(bs->sender, rx_end_payload + BGP_HEADER_LENGTH);
   memset(rx_end_payload + BGP_MSG_HDR_MARKER_POS, 0xff,
 	 BGP_MSG_HDR_MARKER_SIZE); // BGP UPDATE MSG marker
   put_u16(rx_end_payload + BGP_MSG_HDR_LENGTH_POS, pos - rx_end_payload);
@@ -868,27 +914,42 @@ bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs)
 }
 
 static void
-bmp_send_peer_down_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
+bmp_send_peer_down_notif_msg(struct bmp_proto *p, ea_list *bgp,
   const byte *data, const size_t data_size)
 {
   ASSERT(p->started);
 
-  const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps_ext(bgp);
+  //const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps_ext(bgp);
+  int remote_caps = ea_get_int(bgp, &ea_bgp_as4_session, 0);
+  int in_state = ea_get_int(bgp, &ea_bgp_in_conn_state, 0);
+  int out_state = ea_get_int(bgp, &ea_bgp_out_conn_state, 0);
+  int in_as4 = ea_get_int(bgp, &ea_bgp_as4_in_conn, 0);
+  int out_as4 = ea_get_int(bgp, &ea_bgp_as4_out_conn, 0);
+
+  if (in_state && in_as4)
+    remote_caps = in_as4;
+  else if (out_state && out_as4)
+    remote_caps = out_as4;
+
   bool is_global_instance_peer = bmp_is_peer_global_instance(bgp);
-  buffer payload
-    = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
-  bmp_peer_down_notif_msg_serialize(&payload, is_global_instance_peer,
-    bgp->remote_as, bgp->remote_id,
-    remote_caps ? remote_caps->as4_support : bgp->as4_session,
-    bgp->remote_ip, data, data_size);
+  buffer payload = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
+  bmp_peer_down_notif_msg_serialize(
+      &payload,
+      is_global_instance_peer,
+      ea_get_int(bgp, &ea_bgp_rem_as, 0),
+      ea_get_int(bgp, &ea_bgp_rem_id, 0),
+      remote_caps,
+      *((ip_addr *) ea_get_adata(bgp, &ea_bgp_rem_ip)->data),
+      data,
+      data_size
+      );
   bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
 
   bmp_buffer_free(&payload);
 }
 
 static void
-bmp_peer_down_(struct bmp_proto *p, const struct bgp_proto *bgp,
-	       int err_class, int err_code, int err_subcode, const byte *data, int length)
+bmp_peer_down_(struct bmp_proto *p, ea_list *bgp, struct bgp_session_close_ad *bscad)
 {
   if (!p->started)
     return;
@@ -897,12 +958,12 @@ bmp_peer_down_(struct bmp_proto *p, const struct bgp_proto *bgp,
   if (!bp)
     return;
 
-  TRACE(D_STATES, "Peer down for %s", bgp->p.name);
+  TRACE(D_STATES, "Peer down for %s", ea_find(bgp, &ea_name)->u.ad->data);
 
   uint bmp_code = 0;
   uint fsm_code = 0;
 
-  switch (err_class)
+  switch (bscad->last_error_class)
   {
   case BE_BGP_RX:
     bmp_code = BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION;
@@ -916,10 +977,10 @@ bmp_peer_down_(struct bmp_proto *p, const struct bgp_proto *bgp,
 
   default:
     bmp_code = BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION;
-    length = 0;
     break;
   }
 
+  uint length = bscad->ad.length - sizeof *bscad + sizeof bscad->ad;
   buffer payload = bmp_buffer_alloc(p->buffer_mpool, 1 + BGP_HEADER_LENGTH + 2 + length);
   bmp_put_u8(&payload, bmp_code);
 
@@ -928,9 +989,9 @@ bmp_peer_down_(struct bmp_proto *p, const struct bgp_proto *bgp,
   case BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION:
   case BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION:
     bmp_put_bgp_hdr(&payload, BGP_HEADER_LENGTH + 2 + length, PKT_NOTIFICATION);
-    bmp_put_u8(&payload, err_code);
-    bmp_put_u8(&payload, err_subcode);
-    bmp_put_data(&payload, data, length);
+    bmp_put_u8(&payload, bscad->notify_code);
+    bmp_put_u8(&payload, bscad->notify_subcode);
+    bmp_put_data(&payload, bscad->data, length);
     break;
 
   case BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION:
@@ -945,14 +1006,6 @@ bmp_peer_down_(struct bmp_proto *p, const struct bgp_proto *bgp,
   bmp_remove_peer(p, bp);
 }
 
-void
-bmp_peer_down(const struct bgp_proto *bgp,
-	      int err_class, int code, int subcode, const byte *data, int length)
-{
-  struct bmp_proto *p; node *n;
-  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
-    bmp_peer_down_(p, bgp, err_class, code, subcode, data, length);
-}
 
 static void
 bmp_send_termination_msg(struct bmp_proto *p,
@@ -976,52 +1029,82 @@ bmp_send_termination_msg(struct bmp_proto *p,
   bmp_buffer_free(&stream);
 }
 
-int
-bmp_preexport(struct channel *C UNUSED, rte *e)
+static void
+bmp_split_policy(struct bmp_proto *p, const rte *new, const rte *old)
 {
-  /* Reject non-direct routes */
-  if (e->src->proto != e->sender->proto)
-    return -1;
+  rte loc = *(new ?: old);
 
-  /* Reject non-BGP routes */
-  if (e->sender->channel != &channel_bgp)
-    return -1;
+  struct proto *rte_proto = (struct proto*) SKIP_BACK(struct proto, sources, loc.src->owner);
+  struct bgp_proto *bgp = (struct bgp_proto *) rte_proto;
+  struct bgp_channel *src_ch = SKIP_BACK(struct bgp_channel, c.in_req, loc.sender->req);
 
-  return 1;
+  /* Ignore piped routes */
+  if (src_ch->c.proto != rte_proto)
+    return;
+
+  /* Ignore non-BGP routes */
+  if (rte_proto->proto != &proto_bgp)
+    return;
+
+  /* Checking the pre policy */
+  if (p->monitoring_rib.in_pre_policy)
+  {
+    /* Compute the pre policy attributes */
+    loc.attrs = new ? ea_strip_to(new->attrs, BIT32_ALL(EALS_PREIMPORT)) : NULL;
+    ea_list *old_attrs = old ? ea_strip_to(old->attrs, BIT32_ALL(EALS_PREIMPORT)) : NULL;
+
+    bmp_route_monitor_notify(p, bgp, src_ch->afi, false, &loc, old_attrs);
+  }
+
+  /* Checking the post policy */
+  if (p->monitoring_rib.in_post_policy)
+  {
+    /* Compute the post policy attributes */
+    loc.attrs = new ? ea_normalize(new->attrs, 0) : NULL;
+    ea_list *old_attrs = old ? ea_normalize(old->attrs, 0) : NULL;
+
+    bmp_route_monitor_notify(p, bgp, src_ch->afi, true, &loc, old_attrs);
+  }
 }
 
 static void
-bmp_rt_notify(struct proto *P, struct channel *c, struct network *net,
-		struct rte *new, struct rte *old)
+bmp_check_routes(void *bt_)
 {
-  struct bmp_proto *p = (void *) P;
+  struct bmp_table *bt = (struct bmp_table *)bt_;
+  struct bmp_proto *p = bt->p;
 
-  struct bgp_channel *src = (void *) (new ?: old)->sender;
-  struct bgp_proto *bgp = (void *) src->c.proto;
-  bool policy = (c->table == src->c.table);
+  RT_EXPORT_WALK(&bt->out_req, u)
+  {
+    switch (u->kind)
+    {
+      case RT_EXPORT_STOP:
+	bug("Main table export stopped");
 
-  /*
-   * We assume that we receive peer_up before the first route and peer_down
-   * synchronously with BGP session close. So if bmp_stream exists, the related
-   * BGP session is up and could be accessed. That may not be true in
-   * multithreaded setup.
-   */
+      case RT_EXPORT_FEED:
+	/* Send updates one after another */
+	for (uint i = 0; i < u->feed->count_routes; i++)
+	{
+	  rte *new = &u->feed->block[i];
+	  if (new->flags & REF_OBSOLETE)
+	    break;
 
-  struct bmp_stream *bs = bmp_find_stream(p, bgp, src->afi, policy);
-  if (!bs)
-    return;
+	  bmp_split_policy(p, new, NULL);
+	}
+	break;
 
-  bmp_route_monitor_notify(p, bs, net->n.addr, new, (new ?: old)->src);
+      case RT_EXPORT_UPDATE:
+	bmp_split_policy(p, u->update->new, u->update->old);
+	break;
+    }
+  }
 }
 
 static void
-bmp_feed_end(struct channel *c)
+bmp_feed_end(struct rt_export_request *req)
 {
-  struct bmp_proto *p = (void *) c->proto;
+  SKIP_BACK_DECLARE(struct bmp_table, bt, out_req, req);
 
-  struct bmp_table *bt = bmp_find_table(p, c->table);
-  if (!bt)
-    return;
+  struct bmp_proto *p = bt->p;
 
   /*
    * Unsynced streams are added in one moment during BMP session establishment,
@@ -1068,10 +1151,38 @@ bmp_startup(struct bmp_proto *p)
   bmp_buffer_free(&payload);
 
   /* Send Peer Up messages */
-  struct proto *peer;
-  WALK_LIST(peer, proto_list)
-    if ((peer->proto->class == PROTOCOL_BGP) && (peer->proto_state == PS_UP))
-      bmp_peer_init(p, (struct bgp_proto *) peer);
+  u32 length;
+  PST_LOCKED(ts) /* The size of protos field will never decrease, the inconsistency caused by growing is not important */
+    length = ts->length_states;
+
+  /* Subscribe to protocol state changes */
+  p->proto_state_reader = (struct lfjour_recipient) {
+    .event = &p->proto_state_changed,
+    .target = proto_event_list(&p->p),
+  };
+
+  p->proto_state_changed = (event) {
+    .hook = bmp_proto_state_changed,
+    .data = p,
+  };
+
+  proto_states_subscribe(&p->proto_state_reader);
+
+  /* Load protocol states */
+  for (u32 i = 0; i < length; i++)
+  {
+    ea_list *proto_attr = proto_get_state(i);
+    if (proto_attr == NULL)
+      continue;
+
+    struct protocol *proto = (struct protocol *) ea_get_ptr(proto_attr, &ea_protocol_type, 0);
+    const int state = ea_get_int(proto_attr, &ea_state, 0);
+
+    if (proto != &proto_bgp || state != PS_UP)
+      continue;
+
+    bmp_peer_up_inout(p, proto_attr, false);
+  }
 }
 
 /**
@@ -1089,6 +1200,9 @@ bmp_down(struct bmp_proto *p)
 
   TRACE(D_EVENTS, "BMP session closed");
 
+  proto_states_unsubscribe(&p->proto_state_reader);
+  ev_postpone(&p->proto_state_changed);
+
   /* Unregister existing peer structures */
   HASH_WALK_DELSAFE(p->peer_map, next, bp)
   {
@@ -1128,12 +1242,12 @@ bmp_connect(struct bmp_proto *p)
 
   TRACE(D_EVENTS, "Connecting to %I port %u", sk->daddr, sk->dport);
 
-  int rc = sk_open(sk);
+  int rc = sk_open(sk, p->p.loop);
 
   if (rc < 0)
     sk_log_error(sk, p->p.name);
 
-  tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);
+  tm_start_in(p->connect_retry_timer, CONNECT_RETRY_TIME, p->p.loop);
 }
 
 /* BMP connect successful event - switch from Connect to Established state */
@@ -1151,7 +1265,7 @@ bmp_connected(struct birdsock *sk)
   bmp_startup(p);
 }
 
-/* BMP socket error event - switch from any state to Idle state */
+/* BMPbmp_startup socket error event - switch from any state to Idle state */
 static void
 bmp_sock_err(sock *sk, int err)
 {
@@ -1168,8 +1282,9 @@ bmp_sock_err(sock *sk, int err)
     bmp_down(p);
 
   bmp_close_socket(p);
-  tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);
+  tm_start_in(p->connect_retry_timer, CONNECT_RETRY_TIME, p->p.loop);
 
+  if (p->p.proto_state == PS_UP)
     proto_notify_state(&p->p, PS_START);
 }
 
@@ -1210,6 +1325,40 @@ bmp_postconfig(struct proto_config *CF)
     cf_error("Station port number not specified");
 }
 
+
+static void
+bmp_process_proto_state_change(struct bmp_proto *p, struct lfjour_item *last_up)
+{
+  struct proto_pending_update *ppu = SKIP_BACK(struct proto_pending_update, li, last_up);
+  if (!ppu)
+    return;
+
+  if (bmp_peer_up_inout(p, ppu->proto_attr, true))
+    goto done;
+
+  SKIP_BACK_DECLARE(struct bgp_session_close_ad, bscad, ad, ea_get_adata(ppu->proto_attr, &ea_bgp_close_bmp));
+  if (bscad)
+  {
+    bmp_peer_down_(p, ppu->proto_attr, bscad);
+    goto done;
+  }
+
+done:
+  lfjour_release(&p->proto_state_reader, last_up);
+}
+
+static void
+bmp_proto_state_changed(void *_p)
+{
+  struct bmp_proto *p = _p;
+
+  ASSERT_DIE(birdloop_inside(p->p.loop));
+
+  struct lfjour_item *last_up;
+  while (last_up = lfjour_get(&p->proto_state_reader))
+    bmp_process_proto_state_change(p, last_up);
+}
+
 /** Configuration handle section **/
 static struct proto *
 bmp_init(struct proto_config *CF)
@@ -1218,9 +1367,9 @@ bmp_init(struct proto_config *CF)
   struct bmp_proto *p = (void *) P;
   struct bmp_config *cf = (void *) CF;
 
-  P->rt_notify = bmp_rt_notify;
-  P->preexport = bmp_preexport;
-  P->feed_end = bmp_feed_end;
+  ASSERT_DIE(birdloop_inside(&main_birdloop));
+  if (!bgp_next_hop_ea_class)
+    bgp_next_hop_ea_class = ea_class_find_by_name("bgp_next_hop");
 
   p->cf = cf;
   p->local_addr = cf->local_addr;
@@ -1243,10 +1392,10 @@ bmp_start(struct proto *P)
 {
   struct bmp_proto *p = (void *) P;
 
-  p->buffer_mpool = rp_new(P->pool, "BMP Buffer");
-  p->map_mem_pool = rp_new(P->pool, "BMP Map");
-  p->tx_mem_pool = rp_new(P->pool, "BMP Tx");
-  p->update_msg_mem_pool = rp_new(P->pool, "BMP Update");
+  p->buffer_mpool = rp_new(P->pool, proto_domain(&p->p), "BMP Buffer");
+  p->map_mem_pool = rp_new(P->pool, proto_domain(&p->p), "BMP Map");
+  p->tx_mem_pool = rp_new(P->pool, proto_domain(&p->p), "BMP Tx");
+  p->update_msg_mem_pool = rp_new(P->pool, proto_domain(&p->p), "BMP Update");
   p->tx_ev = ev_new_init(p->p.pool, bmp_fire_tx, p);
   p->update_ev = ev_new_init(p->p.pool, bmp_route_monitor_commit, p);
   p->connect_retry_timer = tm_new_init(p->p.pool, bmp_connection_retry, p, 0, 0);
@@ -1260,9 +1409,8 @@ bmp_start(struct proto *P)
   init_list(&p->update_msg_queue);
   p->started = false;
   p->sock_err = 0;
-  add_tail(&bmp_proto_list, &p->bmp_node);
 
-  tm_start(p->connect_retry_timer, CONNECT_INIT_TIME);
+  tm_start_in(p->connect_retry_timer, CONNECT_INIT_TIME, p->p.loop);
 
   return PS_START;
 }
@@ -1279,7 +1427,6 @@ bmp_shutdown(struct proto *P)
   }
 
   p->sock_err = 0;
-  rem_node(&p->bmp_node);
 
   return PS_FLUSH;
 }
@@ -1350,7 +1497,6 @@ bmp_show_proto_info(struct proto *P)
 struct protocol proto_bmp = {
   .name = "BMP",
   .template = "bmp%d",
-  .class = PROTOCOL_BMP,
   .proto_size = sizeof(struct bmp_proto),
   .config_size = sizeof(struct bmp_config),
   .postconfig = bmp_postconfig,

proto/bmp/bmp.h

@@ -47,7 +47,6 @@ struct bmp_proto;
 struct bmp_proto {
   struct proto p;                  // Parent proto
   const struct bmp_config *cf;     // Shortcut to BMP configuration
-  node bmp_node;                   // Node in bmp_proto_list
 
   HASH(struct bmp_peer) peer_map;
   HASH(struct bmp_stream) stream_map;
@@ -73,29 +72,38 @@ struct bmp_proto {
   list update_msg_queue;           // Stores all composed BGP UPDATE MSGs
   bool started;                    // Flag that stores running status of BMP instance
   int sock_err;                    // Last socket error code
+
+  struct lfjour_recipient proto_state_reader; // Reader of protocol states
+  event proto_state_changed;
 };
 
 struct bmp_peer {
-  struct bgp_proto *bgp;
+  ea_list *bgp;
   struct bmp_peer *next;
   list streams;
 };
 
 struct bmp_stream {
   node n;
-  struct bgp_proto *bgp;
+  ea_list *bgp;
   u32 key;
   bool sync;
+  bool shutting_down;
   struct bmp_stream *next;
   struct bmp_table *table;
-  struct bgp_channel *sender;
+  ea_list *sender;
+  int in_pre_policy;
 };
 
 struct bmp_table {
-  struct rtable *table;
+  rtable *table;
   struct bmp_table *next;
   struct channel *channel;
-  u32 uc;
+  struct rt_export_request out_req;
+  struct bmp_proto *p;
+  struct rt_export_feeder in_req;
+  event event;
+  atomic_int uc;
 };
 
 
@@ -105,7 +113,7 @@ struct bmp_table {
  * bmp_peer_up - send notification that BGP peer connection is established
  */
 void
-bmp_peer_up(struct bgp_proto *bgp,
+bmp_peer_up(struct ea_list *bgp,
 	    const byte *tx_open_msg, uint tx_open_length,
 	    const byte *rx_open_msg, uint rx_open_length);
 

proto/bmp/config.Y

@@ -25,6 +25,7 @@ proto: bmp_proto '}' ;
 
 bmp_proto_start: proto_start BMP {
      this_proto = proto_config_new(&proto_bmp, $1);
+     this_proto->loop_order = DOMAIN_ORDER(proto);
      BMP_CFG->sys_descr = "Not defined";
      BMP_CFG->sys_name = "Not defined";
    }