diff --git a/configure.ac b/configure.ac index 9c809201..25e1b159 100644 --- a/configure.ac +++ b/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 diff --git a/nest/proto.c b/nest/proto.c index 12e06be8..44927720 100644 --- a/nest/proto.c +++ b/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", diff --git a/nest/protocol.h b/nest/protocol.h index dc9aa68d..d643d8ba 100644 --- a/nest/protocol.h +++ b/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, diff --git a/proto/bgp/attrs.c b/proto/bgp/attrs.c index 42d36228..597ada4d 100644 --- a/proto/bgp/attrs.c +++ b/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); diff --git a/proto/bgp/bgp.h b/proto/bgp/bgp.h index e3293bab..0d06065f 100644 --- a/proto/bgp/bgp.h +++ b/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 */ diff --git a/proto/bgp/packets.c b/proto/bgp/packets.c index 2ad75c82..7a01dc00 100644 --- a/proto/bgp/packets.c +++ b/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) { diff --git a/proto/bmp/bmp.c b/proto/bmp/bmp.c index 6187914d..c72891c2 100644 --- a/proto/bmp/bmp.c +++ b/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); - channel_set_state(bt->channel, CS_DOWN); - proto_remove_channel(&p->p, bt->channel); + if (bt->channel) + { + channel_set_state(bt->channel, CS_STOP); + channel_set_state(bt->channel, CS_DOWN); + } + 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,9 +1282,10 @@ 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); - proto_notify_state(&p->p, PS_START); + if (p->p.proto_state == PS_UP) + proto_notify_state(&p->p, PS_START); } /* BMP connect timeout event - switch from Idle/Connect state to Connect state */ @@ -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, diff --git a/proto/bmp/bmp.h b/proto/bmp/bmp.h index d69aaafb..b7867f0d 100644 --- a/proto/bmp/bmp.h +++ b/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); diff --git a/proto/bmp/config.Y b/proto/bmp/config.Y index acb0c4d9..55633943 100644 --- a/proto/bmp/config.Y +++ b/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"; }