wp bmp: this commit is just to safe work before vacation. It throws errors and contains unfinished parts

lib/event.c

@@ -333,6 +333,7 @@ ev_run_list_limited(event_list *l, uint limit)
       edlog(l, e, next, 7, EDL_RUN_LIST);
 
       /* Run the event */
+      log("lock service in event.c? locked %i", locking_stack.service);
       e->hook(e->data);
       tmp_flush();
 

lib/route.h

@@ -471,7 +471,8 @@ extern struct ea_class ea_name, ea_protocol_name, ea_protocol_type, ea_table,
        ea_state, ea_old_state, ea_last_modified, ea_info, ea_proto_id,
        ea_channel_id, ea_deleted, ea_bgp_conn, ea_bgp_in_conn, ea_bgp_out_conn, ea_rtable, ea_bgp_afi;
 /* bgp protocol journal attributes */
-extern struct ea_class ea_bgp_rem_id, ea_bgp_rem_as, ea_bgp_loc_as, ea_bgp_rem_ip, ea_bgp_peer_type;
+extern struct ea_class ea_bgp_rem_id, ea_bgp_rem_as, ea_bgp_loc_as, ea_bgp_rem_ip, ea_bgp_peer_type,
+       ea_bgp_local_open_msg, ea_bgp_remote_open_msg, ea_bgp_local_open_msg_len, ea_bgp_remote_open_msg_len;
 
 /* Source: An old method to devise the route source protocol and kind.
  * To be superseded in a near future by something more informative. */

nest/proto.c

@@ -61,6 +61,7 @@ static void channel_reset_limit(struct channel *c, struct limit *l, int dir);
 static void channel_stop_export(struct channel *c);
 static void channel_check_stopped(struct channel *c);
 void init_journals(void);
+void add_journal_channel(struct channel *ch);
 static inline void channel_reimport(struct channel *c, struct rt_feeding_request *rfr)
 {
   rt_export_refeed(&c->reimporter, rfr);
@@ -219,6 +220,7 @@ proto_find_channel_by_name(struct proto *p, const char *n)
 struct channel *
 proto_add_channel(struct proto *p, struct channel_config *cf)
 {
+  log("add channed to protocol %s id %i", p->name, p->id);
   struct channel *c = mb_allocz(proto_pool, cf->class->channel_size);
 
   c->name = cf->name;
@@ -246,6 +248,8 @@ proto_add_channel(struct proto *p, struct channel_config *cf)
   c->channel_state = CS_DOWN;
   c->last_state_change = current_time();
   c->reloadable = 1;
+  c->id = hmap_first_zero(proto_state_table->channel_id_maker);
+  hmap_set(proto_state_table->channel_id_maker, c->id);
 
   init_list(&c->roa_subscriptions);
 
@@ -254,6 +258,7 @@ proto_add_channel(struct proto *p, struct channel_config *cf)
   add_tail(&p->channels, &c->n);
 
   CD(c, "Connected to table %s", c->table->name);
+  add_journal_channel(c);
 
   return c;
 }
@@ -264,7 +269,7 @@ proto_add_main_channel(struct proto *p, struct channel_config *cf)
   p->main_channel = proto_add_channel(p, cf);
   ea_list *eal = proto_state_table->attrs[p->id];
   ea_set_attr(&eal, EA_LITERAL_STORE_STRING(&ea_table, 0, p->main_channel->table->name));
-  proto_journal_state_push(eal, p);
+  proto_journal_state_push(eal, p, 1);
   return p->main_channel;
 }
 
@@ -275,7 +280,7 @@ proto_remove_channel(struct proto *p UNUSED, struct channel *c)
 
   CD(c, "Removed", c->name);
 
-  ea_list *eal = *get_channel_ea(c)->attrs;
+  ea_list *eal = get_channel_ea(c)->attrs;
   ea_set_attr(&eal, EA_LITERAL_EMBEDDED(&ea_deleted, 0, 1));
   channel_journal_state_push(eal,c);
 
@@ -1299,6 +1304,13 @@ proto_new(struct proto_config *cf)
   p->hash_key = random_u32();
   cf->proto = p;
 
+  p->id = hmap_first_zero(proto_state_table->proto_id_maker);
+  hmap_set(proto_state_table->proto_id_maker, p->id);
+  log("protocol %s has id %i", p->name, p->id);
+  if (p->id >= proto_state_table->length)
+    protos_attr_field_grow();
+  init_list(&proto_state_table->channels_attrs[p->id]);
+
   init_list(&p->channels);
 
   return p;
@@ -1321,13 +1333,9 @@ proto_init(struct proto_config *c, struct proto *after)
 
   PD(p, "Initializing%s", p->disabled ? " [disabled]" : "");
 
-  p->id = hmap_first_zero(proto_state_table->proto_id_maker);
-  hmap_set(proto_state_table->proto_id_maker, p->id);
-  if (p->id >= proto_state_table->length)
-    protos_attr_field_grow();
   ea_list *eal = proto_state_to_eattr(p, old_state, 0);
 
-  proto_journal_state_push(eal, p);
+  proto_journal_state_push(eal, p, 1);
 
   return p;
 }
@@ -1753,7 +1761,7 @@ proto_rethink_goal(struct proto *p)
 
     ea_list *eal = proto_state_table->attrs[p->id];
     ea_set_attr(&eal, EA_LITERAL_EMBEDDED(&ea_deleted, 0, 1));
-    proto_journal_state_push(eal, p);
+    proto_journal_state_push(eal, p, 1);
     log("deleting %i", p->id);
     hmap_clear(proto_state_table->proto_id_maker, p->id);
     atomic_store(&proto_state_table->attrs[p->id], NULL);
@@ -2354,7 +2362,7 @@ proto_notify_state(struct proto *p, uint state)
   ea_list *eal = proto_state_table->attrs[p->id];
   ea_set_attr(&eal, EA_LITERAL_EMBEDDED(&ea_state, 0, p->proto_state));
   ea_set_attr(&eal, EA_LITERAL_EMBEDDED(&ea_old_state, 0, old_state));
-  proto_journal_state_push(eal, p);
+  proto_journal_state_push(eal, p, 1);
 
   switch (state)
   {
@@ -2896,13 +2904,15 @@ proto_iterate_named(struct symbol *sym, struct protocol *proto, struct proto *ol
 void
 protos_attr_field_init(void)
 {
+  int init_length = 16;
   proto_state_table = mb_allocz(&root_pool, sizeof(struct proto_attrs));
-  proto_state_table->attrs = mb_allocz(&root_pool, sizeof(ea_list *_Atomic)*16);
+  proto_state_table->attrs = mb_allocz(&root_pool, sizeof(ea_list)*init_length);
-  proto_state_table->length = 16;
+  proto_state_table->channels_attrs = mb_allocz(&root_pool, sizeof(list)*init_length);
+  proto_state_table->length = init_length;
   proto_state_table->proto_id_maker = mb_allocz(&root_pool, sizeof(struct hmap));
   proto_state_table->channel_id_maker = mb_allocz(&root_pool, sizeof(struct hmap));
-  hmap_init(proto_state_table->proto_id_maker, &root_pool, 16);
+  hmap_init(proto_state_table->proto_id_maker, &root_pool, init_length);
-  hmap_init(proto_state_table->channel_id_maker, &root_pool, 16*2);
+  hmap_init(proto_state_table->channel_id_maker, &root_pool, init_length*2);
   //TODO free? or have self pool?
 }
 
@@ -2911,8 +2921,11 @@ protos_attr_field_grow(void)
 {
   log("grow");
   ea_list *_Atomic * new_field = mb_allocz(&root_pool, sizeof(ea_list *_Atomic)*proto_state_table->length*2);
+  list * new_chann =  mb_allocz(&root_pool, sizeof(list)*proto_state_table->length*2);
   memcpy(new_field, proto_state_table->attrs, proto_state_table->length*(sizeof(ea_list* _Atomic)));
+  memcpy(new_chann, proto_state_table->channels_attrs, proto_state_table->length*(sizeof(list)));
   atomic_store(&proto_state_table->attrs, new_field);
+  proto_state_table->channels_attrs = new_chann; //TODO atomic
   atomic_store(&proto_state_table->length, (proto_state_table->length*2));
 }
 
@@ -2921,9 +2934,11 @@ cleanup_journal_item(struct lfjour * journal UNUSED, struct lfjour_item *i)
 {
   log("cleanup_journal_item");
   struct proto_pending_update *pupdate = SKIP_BACK(struct proto_pending_update, li, i);
-  ea_free_later(pupdate->old_attr);
+  //ea_free_later(pupdate->old_attr);
-  eattr *new_ea = ea_find(pupdate->proto_attr, &ea_deleted);
+  int deleting = ea_get_int(pupdate->proto_attr, &ea_deleted, 0);
-  if (new_ea->u.i)
+  //TODO temporal eatters
+
+  if (deleting)
   {
     log("try to delete");
     ea_free_later(pupdate->proto_attr);
@@ -2958,9 +2973,10 @@ init_journals(void)
 ea_list *
 proto_state_to_eattr(struct proto *p, int old_state, int proto_deleting)
 {
+  log("protocol %s to eattr", p->name);
   struct {
 	ea_list l;
-	eattr a[9+9];
+	eattr a[9+15];
   } eattrs;
 
   eattrs.l = (ea_list) {};
@@ -2988,6 +3004,9 @@ channel_state_to_eattr(struct channel *ch, int proto_deleting)
 	ea_list l;
 	eattr a[5];
   } eattrs;
+  
+  eattrs.l = (ea_list) {};
+
   eattrs.a[eattrs.l.count++] = EA_LITERAL_STORE_STRING(&ea_name, 0, ch->name);
   eattrs.a[eattrs.l.count++] = EA_LITERAL_EMBEDDED(&ea_proto_id, 0, ch->proto->id);
   eattrs.a[eattrs.l.count++] = EA_LITERAL_EMBEDDED(&ea_deleted, 0, proto_deleting);
@@ -2998,19 +3017,27 @@ channel_state_to_eattr(struct channel *ch, int proto_deleting)
     struct bgp_channel *bc = (struct bgp_channel *) ch;
     eattrs.a[eattrs.l.count++] = EA_LITERAL_EMBEDDED(&ea_bgp_afi, 0, bc->afi);
   }
+  log("to lookup goes %i (%i)", eattrs.l, &eattrs.l);
   return ea_lookup_slow(&eattrs.l, 0, EALS_CUSTOM);
 }
 
 //mpls!!!!! ch nonskip
 
 void
-proto_journal_state_push(ea_list *attr, struct proto *p)
+proto_journal_state_push(ea_list *attr, struct proto *p, int save_to_jour)
 {
+  log("proto_journal_state_push %i", p->id);
+  log("push state for %s", ea_get_adata(attr, &ea_name)->data);
   ea_set_attr(&attr, EA_LITERAL_STORE_ADATA(&ea_last_modified, 0, &p->last_state_change, sizeof(btime)));
   
   attr = ea_lookup(attr, 0, EALS_CUSTOM);
-  atomic_store(&proto_state_table->attrs[p->id], attr);
   ea_list *old_attr = proto_state_table->attrs[p->id];
+  if (save_to_jour)
+  {
+    atomic_store(&proto_state_table->attrs[p->id], attr);
+    ASSERT(ea_get_int(attr, &ea_bgp_remote_open_msg_len, 0) == 0);
+  }
+
   LOCK_DOMAIN(rtable, proto_journal_domain);
   struct proto_pending_update *pupdate = SKIP_BACK(struct proto_pending_update, li, lfjour_push_prepare(proto_journal));
   if (!pupdate)
@@ -3045,7 +3072,7 @@ get_channel_ea(struct channel *ch)
   struct channel_attrs *chan_att;
   WALK_LIST(chan_att, proto_state_table->channels_attrs[ch->proto->id])
   {
-    const int id = ea_get_int(*chan_att->attrs, &ea_channel_id, 0);
+    const int id = ea_get_int(chan_att->attrs, &ea_channel_id, 0);
     if (ch->id == id)
       return chan_att;
   }
@@ -3055,7 +3082,8 @@ get_channel_ea(struct channel *ch)
 void
 channel_journal_state_push(ea_list *attr, struct channel *ch)
 {
-  ea_list *old_attr = *get_channel_ea(ch)->attrs;
+  attr = ea_lookup(attr, 0, EALS_CUSTOM);
+  ea_list *old_attr = get_channel_ea(ch)->attrs;
   LOCK_DOMAIN(rtable, proto_journal_domain);
   struct channel_pending_update *pupdate = SKIP_BACK(struct channel_pending_update, li, lfjour_push_prepare(channel_journal));
   if (!pupdate)
@@ -3063,6 +3091,8 @@ channel_journal_state_push(ea_list *attr, struct channel *ch)
     UNLOCK_DOMAIN(rtable, proto_journal_domain);
     return;
   }
+  struct channel_attrs *ch_attr = get_channel_ea(ch);
+  atomic_store(&ch_attr->attrs, attr);
   *pupdate = (struct channel_pending_update) {
     .li = pupdate->li,	/* Keep the item's internal state */
     .channel_attr = attr,
@@ -3083,21 +3113,23 @@ channel_journal_state_change(struct channel *ch, int old_state, int new_state)
 }*/
 
 void
-add_journal_channel(struct channel *ch, int new_state)
+add_journal_channel(struct channel *ch)
 {
-  ea_list *eal = channel_state_to_eattr(ch, new_state);
+  log("adding channel to %s %i", ch->proto->name, ch->proto->id);
+  ea_list *eal = channel_state_to_eattr(ch, 0);
+  log("eal %i", eal);
   struct channel_attrs *attr = mb_allocz(&root_pool, sizeof(struct channel_attrs));
-  atomic_store(attr->attrs, eal);
+  //atomic_store(attr->attrs, eal);
-  ASSERT_DIE(proto_state_table->attrs[ch->proto->id] != NULL);
+  attr->attrs = eal;
-  add_tail(&proto_state_table->channels_attrs[ch->proto->id], &ch->n);
+  add_tail(&proto_state_table->channels_attrs[ch->proto->id], &attr->n);
-  
+  log("really, added %i to id %i (attr %i attrs %i)", proto_state_table->channels_attrs[ch->proto->id], ch->proto->id, attr, attr->attrs);
 }
 
 void dummy_log_proto_attr_list(void)
 {
 //debugging function
   ea_list *eal;
-  for(u32 i = 0; i < proto_state_table->length; i++)
+  for (u32 i = 0; i < proto_state_table->length; i++)
   {
     eal = proto_get_state_list(i);
     if (eal)

nest/protocol.h

@@ -413,10 +413,11 @@ struct proto_attrs {
 extern struct lfjour *proto_journal;
 extern struct lfjour *channel_journal;
 extern struct proto_attrs *proto_state_table;
+extern DOMAIN(rtable) proto_journal_domain;
 
 struct channel_attrs {
   node n;
-  ea_list *_Atomic *attrs;
+  ea_list *_Atomic attrs;
 };
 
 struct proto_pending_update {
@@ -433,7 +434,7 @@ struct channel_pending_update {
   struct channel *channel;
 };
 
-void proto_journal_state_push(ea_list *attr, struct proto *p);
+void proto_journal_state_push(ea_list *attr, struct proto *p,  int save_to_jour);
 void channel_journal_state_push(ea_list *attr, struct channel *ch);
 ea_list *proto_state_to_eattr(struct proto *p, int old_state, int protocol_deleting);
 struct channel_attrs *get_channel_ea(struct channel *ch);

nest/rt-attr.c

@@ -1965,6 +1965,26 @@ struct ea_class ea_rtable = {
   .type = T_PTR,
 };
 
+struct ea_class ea_bgp_local_open_msg_len = {
+  .name = "bgp_local_open_msg_len",
+  .type = T_INT,
+};
+
+struct ea_class ea_bgp_remote_open_msg_len = {
+  .name = "bgp_remote_open_msg_len",
+  .type = T_INT,
+};
+
+struct ea_class ea_bgp_local_open_msg = {
+  .name = "bgp_local_open_msg",
+  .type = T_BYTESTRING,
+};
+
+struct ea_class ea_bgp_remote_open_msg = {
+  .name = "bgp_remote_open_msg",
+  .type = T_BYTESTRING,
+};
+
 /**
  * rta_init - initialize route attribute cache
  *
@@ -2030,6 +2050,10 @@ rta_init(void)
   ea_register_init(&ea_rtable);
   ea_register_init(&ea_bgp_peer_type);
   ea_register_init(&ea_bgp_afi);
+  ea_register_init(&ea_bgp_local_open_msg);
+  ea_register_init(&ea_bgp_remote_open_msg);
+  ea_register_init(&ea_bgp_local_open_msg_len);
+  ea_register_init(&ea_bgp_remote_open_msg_len);
 }
 
 /*

proto/bgp/attrs.c

@@ -1316,6 +1316,7 @@ bgp_export_attrs(struct bgp_export_state *s, ea_list *a)
 static inline int
 bgp_encode_attr(struct bgp_write_state *s, eattr *a, byte *buf, uint size)
 {
+  log("size %i", size);
   const union bgp_attr_desc *desc = bgp_find_attr_desc(a);
   if (s->ignore_non_bgp_attrs == 0)
     ASSERT_DIE(desc);

proto/bgp/bgp.c

@@ -678,7 +678,7 @@ bgp_conn_enter_established_state(struct bgp_conn *conn)
   p->conn = conn;
   ea_list *eal = proto_state_table->attrs[p->p.id];
   ea_set_attr(&eal, EA_LITERAL_STORE_PTR(&ea_bgp_conn, 0, p->conn));
-  proto_journal_state_push(eal, &p->p);
+  proto_journal_state_push(eal, &p->p, 1);
   p->last_error_class = 0;
   p->last_error_code = 0;
 
@@ -795,9 +795,21 @@ bgp_conn_enter_established_state(struct bgp_conn *conn)
   bgp_conn_set_state(conn, BS_ESTABLISHED);
   proto_notify_state(&p->p, PS_UP);
   
+  log("bgp domain %i, locked %i", p->p.pool->domain, DG_IS_LOCKED(p->p.pool->domain));
+
 #ifdef CONFIG_BMP
-  bmp_peer_up(proto_state_table->attrs[p->p.id], conn->local_open_msg, conn->local_open_length,
+  //TODO not to repeat done eatters
-	      conn->remote_open_msg, conn->remote_open_length);
+  log("in bgp nesmysl, protocol %s id %i", p->p.name, p->p.id);
+  ea_list *ea = proto_state_table->attrs[p->p.id];
+  ea_set_attr(&ea, EA_LITERAL_STORE_ADATA(&ea_bgp_local_open_msg, 0, conn->local_open_msg, conn->local_open_length));
+  ea_set_attr(&ea, EA_LITERAL_STORE_ADATA(&ea_bgp_remote_open_msg, 0, conn->remote_open_msg, conn->remote_open_length));
+  ea_set_attr(&ea, EA_LITERAL_EMBEDDED(&ea_bgp_local_open_msg_len, 0, conn->local_open_length));
+  ea_set_attr(&ea, EA_LITERAL_EMBEDDED(&ea_bgp_remote_open_msg_len, 0, conn->remote_open_length));
+  ea = ea_lookup(ea, 0, EALS_CUSTOM);
+  proto_journal_state_push(ea, &p->p, 0);
+  log("send to bmp locked %i (p %i)", locking_stack.service, p->p.id);
+  //bmp_peer_up(proto_state_table->attrs[p->p.id], conn->local_open_msg, conn->local_open_length,
+  //         conn->remote_open_msg, conn->remote_open_length);
 #endif
 }
 
@@ -2367,7 +2379,7 @@ bgp_reconfigure(struct proto *P, struct proto_config *CF)
   p->cf = new;
   ea_list *eal = proto_state_table->attrs[p->p.id];
   ea_set_attr(&eal, EA_LITERAL_EMBEDDED(&ea_bgp_peer_type, 0, p->cf->peer_type));
-  proto_journal_state_push(eal, &p->p);
+  proto_journal_state_push(eal, &p->p, 1);
 
   /* Check whether existing connections are compatible with required capabilities */
   struct bgp_conn *ci = &p->incoming_conn;
@@ -2588,6 +2600,16 @@ bgp_state_to_eattr(struct proto *P, ea_list *l, eattr *attributes)
     attributes[l->count++] = EA_LITERAL_STORE_PTR(&ea_bgp_in_conn, 0, &p->incoming_conn);
     attributes[l->count++] = EA_LITERAL_STORE_PTR(&ea_bgp_out_conn, 0, &p->outgoing_conn);
   }
+  else
+  {
+    attributes[l->count++] = EA_LITERAL_STORE_PTR(&ea_bgp_conn, 0, NULL);
+    attributes[l->count++] = EA_LITERAL_STORE_PTR(&ea_bgp_in_conn, 0, NULL);
+    attributes[l->count++] = EA_LITERAL_STORE_PTR(&ea_bgp_out_conn, 0, NULL);
+  }
+  attributes[l->count++] = EA_LITERAL_STORE_ADATA(&ea_bgp_local_open_msg, 0, NULL, 0);
+  attributes[l->count++] = EA_LITERAL_STORE_ADATA(&ea_bgp_remote_open_msg, 0, NULL, 0);
+  attributes[l->count++] = EA_LITERAL_EMBEDDED(&ea_bgp_local_open_msg_len, 0, 0);
+  attributes[l->count++] = EA_LITERAL_EMBEDDED(&ea_bgp_remote_open_msg_len, 0, 0);
 }
 
 static void

proto/bmp/bmp.c

@@ -279,8 +279,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));
+  struct bmp_data_node *tx_data = mb_allocz(p->tx_mem_pool, sizeof (struct bmp_data_node));
-  tx_data->data = mb_alloc(p->tx_mem_pool, size);
+  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);
@@ -602,7 +602,16 @@ bmp_find_peer(struct bmp_proto *p, ea_list *bgp_attr)
 static struct bmp_peer *
 bmp_add_peer(struct bmp_proto *p, ea_list *bgp_attr)
 {
-  struct bmp_peer *bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
+  log("domain is locked %i, service is locked %i", DG_IS_LOCKED(p->p.pool->domain), locking_stack.service);
+  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);
@@ -612,9 +621,12 @@ bmp_add_peer(struct bmp_proto *p, ea_list *bgp_attr)
   int proto_id = ea_get_int(bgp_attr, &ea_proto_id, 0);
   
   ea_list *chan_attr;
+  log("before while id %i, eattrs %i", proto_id, proto_state_table->channels_attrs[proto_id]);
   WALK_LIST(chan_attr, proto_state_table->channels_attrs[proto_id])
   {
-    rtable *ch_table = (rtable *) ea_get_adata(chan_attr, &ea_rtable)->data;
+    log("chan_attr in bmp_add_peer %i", chan_attr);
+    rtable *ch_table = (rtable *) ea_get_ptr(chan_attr, &ea_rtable, 0);
+    log("has table ptr %i", ch_table);
     if (p->monitoring_rib.in_pre_policy && ch_table)
       bmp_add_stream(p, bp, ea_get_int(chan_attr, &ea_bgp_afi, 0), false, ch_table, chan_attr, 1);
 
@@ -1079,6 +1091,7 @@ bmp_startup(struct bmp_proto *p)
 
   TRACE(D_EVENTS, "BMP session established");
 
+  log("bmp startup %s %i", p->p.name, p->p.id);
   proto_notify_state(&p->p, PS_UP);
 
   /* Send initiation message */
@@ -1241,10 +1254,57 @@ bmp_postconfig(struct proto_config *CF)
     cf_error("Station port number not specified");
 }
 
+void
+fc_for_bmp_recipient(void *rec)
+{
+  log("received update, locked %i", locking_stack.service);
+  struct lfjour_item *last_up;
+  struct proto_pending_update *pupdate;
+  while (last_up = lfjour_get((struct lfjour_recipient *)rec))
+  {
+    pupdate = SKIP_BACK(struct proto_pending_update, li, last_up);
+    const byte *tx_open_msg = ea_get_adata(pupdate->proto_attr, &ea_bgp_local_open_msg)->data;
+
+    int id = ea_get_int(pupdate->proto_attr, &ea_proto_id, 0);
+    log("id %i, tx = %i, l %i", id, tx_open_msg, ea_get_int(pupdate->proto_attr, &ea_bgp_local_open_msg_len, 0)); //FIXME why WHY are bmp eattrs able to give not null pointer?
+    if (ea_get_int(pupdate->proto_attr, &ea_bgp_local_open_msg_len, 0))
+    {
+      struct protocol *proto = (struct protocol *) ea_get_ptr(pupdate->proto_attr, &ea_protocol_type, 0);
+      //if (proto != &proto_bgp)
+      //{
+        log("protocol %s is bgp %i", proto->name, proto == &proto_bgp);
+        //bug("wrong protocol");
+        //}
+      //int id = ea_get_int(pupdate->proto_attr, &ea_proto_id, 0);
+      const byte *rx_open_msg = ea_get_adata(pupdate->proto_attr, &ea_bgp_remote_open_msg)->data;
+      int l_len = ea_get_int(pupdate->proto_attr, &ea_bgp_remote_open_msg_len, 0);
+      int r_len = ea_get_int(pupdate->proto_attr, &ea_bgp_remote_open_msg_len, 0);
+      bmp_peer_up(proto_state_table->attrs[id], tx_open_msg, l_len, rx_open_msg, r_len);
+    }
+	    
+	    
+    lfjour_release(rec);
+  }
+}
+
+void
+create_bmp_recipient(void)
+{
+  struct lfjour_recipient *r = mb_allocz(&root_pool, sizeof(struct lfjour_recipient));
+  r->event = ev_new_init(&root_pool, fc_for_bmp_recipient, r);
+  struct birdloop *loop = birdloop_new(&root_pool, DOMAIN_ORDER(service), 1, "bmp recipient loop");
+  r->target = birdloop_event_list(loop);
+  
+  LOCK_DOMAIN(rtable, proto_journal_domain);
+  lfjour_register(proto_journal, r);
+  UNLOCK_DOMAIN(rtable, proto_journal_domain);
+}
+
 /** Configuration handle section **/
 static struct proto *
 bmp_init(struct proto_config *CF)
 {
+  log("start init locked %i", locking_stack.service);
   struct proto *P = proto_new(CF);
   struct bmp_proto *p = (void *) P;
   struct bmp_config *cf = (void *) CF;
@@ -1261,6 +1321,8 @@ bmp_init(struct proto_config *CF)
   strcpy(p->sys_name, cf->sys_name);
   p->monitoring_rib.in_pre_policy = cf->monitoring_rib_in_pre_policy;
   p->monitoring_rib.in_post_policy = cf->monitoring_rib_in_post_policy;
+  create_bmp_recipient();
+  log("new proto created locked %i", locking_stack.service);
 
   return P;
 }
@@ -1272,6 +1334,7 @@ bmp_init(struct proto_config *CF)
 static int
 bmp_start(struct proto *P)
 {
+  log("start locked %i", locking_stack.service);
   struct bmp_proto *p = (void *) P;
 
   p->buffer_mpool = rp_new(P->pool, proto_domain(&p->p), "BMP Buffer");
@@ -1295,6 +1358,7 @@ bmp_start(struct proto *P)
 
   tm_start(p->connect_retry_timer, CONNECT_INIT_TIME);
 
+  log("end of start locked %i", locking_stack.service);
   return PS_START;
 }
 

proto/mrt/mrt.c

@@ -395,9 +395,9 @@ mrt_peer_table_dump(struct mrt_table_dump_state *s)
       continue;
     }
     rcu_read_unlock();
-    struct protocol **type = (struct protocol **)ea_get_ptr(eal, &ea_protocol_type, 0);
+    struct protocol *type = (struct protocol *)ea_get_ptr(eal, &ea_protocol_type, 0);
     int state = ea_get_int(eal, &ea_state, 0);
-    if ((*type == &proto_bgp) && (state != PS_DOWN))
+    if ((type == &proto_bgp) && (state != PS_DOWN))
     {
       int rem_id = ea_get_int(eal, &ea_bgp_rem_id, 0);
       int rem_as = ea_get_int(eal, &ea_bgp_rem_as, 0);