Merge commit 'e6a100b31a7637ee739338e4b933367707ec931f' into thread-next

doc/bird.sgml

@@ -3844,6 +3844,15 @@ by default and have to be enabled during installation by the configure option
 routes (in <ref id="bgp-import-table" name="BGP import tables">) and post-policy
 routes (in regular routing tables). All BGP protocols are monitored automatically.
 
+<sect1>Configuration (incomplete)
+<label id="bmp-config">
+
+<p><descrip>
+	<tag><label id="bmp-tx-buffer-limit">tx buffer limit <m/number/</tag>
+	How much data we are going to queue before we call the session stuck
+	and restart it, in megabytes. Default value: 1024 (effectively 1 gigabyte).
+</descrip>
+
 <sect1>Example
 <label id="bmp-exam">
 
@@ -3857,6 +3866,9 @@ protocol bmp {
 
 	# Monitor accepted routes (passed import filters)
 	monitoring rib in post_policy;
+
+	# Allow only 64M of pending data
+	tx buffer limit 64;
 }
 </code>
 

proto/bgp/bgp.h

@@ -703,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(ea_list *c, struct bgp_proto *bgp_p, byte *buf, const struct rte *new);
+byte * bgp_bmp_encode_rte(ea_list *c, struct bgp_proto *bgp_p, byte *buf, byte *end, const struct rte *new);
 
 #define BGP_AIGP_METRIC		1
 #define BGP_AIGP_MAX		U64(0xffffffffffffffff)

proto/bgp/packets.c

@@ -2480,7 +2480,7 @@ bgp_create_mp_unreach(struct bgp_write_state *s, struct bgp_bucket *buck, byte *
 #ifdef CONFIG_BMP
 
 static byte *
-bgp_create_update_bmp(ea_list *channel_ea, struct bgp_proto *bgp_p, byte *buf, struct bgp_bucket *buck, bool update)
+bgp_create_update_bmp(ea_list *channel_ea, struct bgp_proto *bgp_p, byte *buf, byte *end, struct bgp_bucket *buck, bool update)
 {
   struct bgp_channel *c;
   u32 c_id = ea_get_int(channel_ea, &ea_channel_id, 0);
@@ -2488,9 +2488,7 @@ bgp_create_update_bmp(ea_list *channel_ea, struct bgp_proto *bgp_p, byte *buf, s
     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 = bgp_p->conn->remote_caps;
   const struct bgp_af_caps *rem = bgp_find_af_caps(peer, c->afi);
@@ -2538,7 +2536,7 @@ bgp_bmp_prepare_bgp_hdr(byte *buf, const u16 msg_size, const u8 msg_type)
 }
 
 byte *
-bgp_bmp_encode_rte(ea_list *c, struct bgp_proto *bgp_p, byte *buf, const struct rte *new)
+bgp_bmp_encode_rte(ea_list *c, struct bgp_proto *bgp_p, byte *buf, byte *end, const struct rte *new)
 {
   byte *pkt = buf + BGP_HEADER_LENGTH;
 
@@ -2562,7 +2560,7 @@ bgp_bmp_encode_rte(ea_list *c, struct bgp_proto *bgp_p, byte *buf, const struct
   px->ni = NET_TO_INDEX(new->net);
   add_tail(&b->prefixes, &px->buck_node);
 
-  byte *end = bgp_create_update_bmp(c, bgp_p, pkt, b, !!new->attrs);
+  end = bgp_create_update_bmp(c, bgp_p, pkt, end, b, !!new->attrs);
 
   if (end)
     bgp_bmp_prepare_bgp_hdr(buf, end - buf, PKT_UPDATE);

proto/bmp/Makefile

@@ -1,4 +1,4 @@
-src := bmp.c buffer.c map.c
+src := bmp.c map.c
 obj := $(src-o-files)
 $(all-daemon)
 $(cf-local)

proto/bmp/bmp.c

@@ -29,7 +29,6 @@
  */
 
 #include "proto/bmp/bmp.h"
-#include "proto/bmp/buffer.h"
 #include "proto/bmp/map.h"
 
 #include <sys/socket.h>
@@ -93,16 +92,10 @@ enum bmp_message_type {
   BMP_ROUTE_MIRROR_MSG = 6 // Route Mirroring Message
 };
 
-// Total size of Common Header
-#define BMP_COMMON_HDR_SIZE 6
 // Defines size of padding when IPv4 address is going to be put into field
 // which can accept also IPv6 address
 #define BMP_PADDING_IP4_ADDR_SIZE 12
 
-/* BMP Per-Peer Header [RFC 7854 - Section 4.2] */
-// Total size of Per-Peer Header
-#define BMP_PER_PEER_HDR_SIZE 42
-
 enum bmp_peer_type {
   BMP_PEER_TYPE_GLOBAL_INSTANCE = 0,
   BMP_PEER_TYPE_RD_INSTANCE = 1,
@@ -165,6 +158,15 @@ enum bmp_peer_down_notif_reason {
   BMP_PEER_DOWN_REASON_PEER_DE_CONFIGURED = 5
 };
 
+struct bmp_peer_down_info {
+  u8 reason;
+  u8 fsm_code;
+  u8 err_code;
+  u8 err_subcode;
+  const byte *data;
+  int length;
+};
+
 /* BMP Termination Message [RFC 7854 - Section 4.5] */
 #define BMP_TERM_INFO_TYPE_SIZE 2
 enum bmp_term_info_type {
@@ -197,20 +199,42 @@ enum bmp_term_reason {
 #define IP4_MAX_TTL 255
 
 
-#define IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(expr, msg, rv...)     \
+#define bmp_buffer_need(b, sz)  ASSERT_DIE((b)->pos + (sz) <= (b)->end)
-  do {                                                                      \
+
-    if ((expr))                                                             \
+// Idea for following macros has been taken from |proto/mrt/mrt.c|
+#define BMP_DEFINE_PUT_FUNC(S, T)                               \
+  static inline void                                            \
+  bmp_put_##S(buffer *b, const T x)                             \
   {                                                             \
-      log(L_WARN "[BMP] " msg);                                             \
+    bmp_buffer_need(b, sizeof(T));			   \
-      return rv;                                                            \
+    put_##S(b->pos, x);                                    \
-    }                                                                       \
+    b->pos += sizeof(T);                                   \
-  } while (0)
+  }
 
+BMP_DEFINE_PUT_FUNC(u8, u8)
+BMP_DEFINE_PUT_FUNC(u16, u16)
+BMP_DEFINE_PUT_FUNC(u32, u32)
+BMP_DEFINE_PUT_FUNC(u64, u64)
+BMP_DEFINE_PUT_FUNC(ip4, ip4_addr)
+BMP_DEFINE_PUT_FUNC(ip6, ip6_addr)
 
-#define IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(p, msg, rv...)	\
+static inline void
-  do {									\
+bmp_put_data(buffer *b, const void *src, const size_t len)
-    IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(!(p), msg, rv);	\
+{
-  } while (0)
+  ASSERT_DIE(b->pos + len <= b->end);
+  memcpy(b->pos, src, len);
+  b->pos += len;
+}
+
+static inline buffer
+bmp_default_buffer(struct bmp_proto *p)
+{
+  return (buffer) {
+    .start = p->msgbuf,
+    .pos = p->msgbuf,
+    .end = p->msgbuf + sizeof p->msgbuf,
+  };
+}
 
 static const struct ea_class *bgp_next_hop_ea_class = NULL;
 
@@ -228,18 +252,44 @@ bmp_send_peer_up_notif_msg(struct bmp_proto *p, ea_list *bgp,
 
 static void bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs);
 
-// Stores necessary any data in list
+// Stores TX data
-struct bmp_data_node {
+struct bmp_tx_buffer {
-  node n;
+  struct bmp_tx_buffer *next;
-  byte *data;
+  byte *pos;
-  size_t data_size;
+  byte data[];
+};
 
-  u32 remote_as;
+#define bmp_tx_remains(b) (((byte *) (b) + page_size) - (b)->pos)
-  u32 remote_id;
+
-  ip_addr remote_ip;
+/* A dummy resource to accurately show memory pages allocated for pending TX */
-  btime timestamp;
+struct bmp_tx_resource {
-  bool global_peer;
+  resource r;
-  bool policy;
+  struct bmp_proto *p;
+};
+
+static void
+bmp_tx_resource_free(resource *r UNUSED) {}
+
+static void
+bmp_tx_resource_dump(resource *r UNUSED, uint indent UNUSED) {}
+
+static struct resmem
+bmp_tx_resource_memsize(resource *r)
+{
+  struct bmp_proto *p = SKIP_BACK(struct bmp_tx_resource, r, r)->p;
+
+  return (struct resmem) {
+    .effective = p->tx_pending_count * page_size,
+    .overhead = sizeof(struct bmp_tx_resource),
+  };
+}
+
+static struct resclass bmp_tx_resource_class = {
+  .name = "BMP TX buffers",
+  .size = sizeof(struct bmp_tx_resource),
+  .free = bmp_tx_resource_free,
+  .dump = bmp_tx_resource_dump,
+  .memsize = bmp_tx_resource_memsize,
 };
 
 static void
@@ -271,6 +321,7 @@ bmp_init_msg_serialize(buffer *stream, const char *sys_descr, const char *sys_na
   // We include MIB-II sysDescr and sysName in BMP INIT MSG so that's why
   // allocated 2x BMP_INFO_TLV_FIX_SIZE memory pool size
   const size_t data_size = (2 * BMP_INFO_TLV_FIX_SIZE) + sys_descr_len + sys_name_len;
+
   bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
   bmp_common_hdr_serialize(stream, BMP_INIT_MSG, data_size);
   bmp_info_tlv_hdr_serialize(stream, BMP_INFO_TLV_TYPE_SYS_DESCR, sys_descr);
@@ -278,21 +329,62 @@ bmp_init_msg_serialize(buffer *stream, const char *sys_descr, const char *sys_na
 }
 
 static void
-bmp_schedule_tx_packet(struct bmp_proto *p, const byte *payload, const size_t size)
+bmp_schedule_tx_packet(struct bmp_proto *p, const byte *payload, size_t size)
 {
   ASSERT(p->started);
 
-  struct bmp_data_node *tx_data = mb_allocz(p->tx_mem_pool, sizeof (struct bmp_data_node));
+  while (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);
-
-  if (sk_tx_buffer_empty(p->sk)
-      && !ev_active(p->tx_ev))
   {
-    ev_send_loop(p->p.loop, p->tx_ev);
+    if (!p->tx_last || !bmp_tx_remains(p->tx_last))
+    {
+      if (p->tx_pending_count >= p->tx_pending_limit)
+	return ev_send_loop(p->p.loop, p->tx_overflow_event);
+
+      p->tx_pending_count++;
+
+      struct bmp_tx_buffer *btb = alloc_page();
+      btb->pos = btb->data;
+      btb->next = NULL;
+
+      if (p->tx_last)
+      {
+	ASSERT_DIE(!p->tx_last->next);
+	p->tx_last->next = btb;
       }
+      else
+	ASSERT_DIE(p->tx_pending_count == 1);
+
+      p->tx_last = btb;
+
+      if (!p->tx_pending)
+	p->tx_pending = btb;
+    }
+
+    size_t cpylen = bmp_tx_remains(p->tx_last);
+    if (size < cpylen)
+      cpylen = size;
+
+    memcpy(p->tx_last->pos, payload, cpylen);
+    p->tx_last->pos += cpylen;
+
+    payload += cpylen;
+    size -= cpylen;
+  }
+
+  if (!p->sk->tbuf && !ev_active(p->tx_ev))
+    ev_send_loop(p->p.loop, p->tx_ev);
+}
+
+static void
+bmp_tx_buffer_free(struct bmp_proto *p, struct bmp_tx_buffer *btb)
+{
+  if (btb == p->tx_last)
+  {
+    p->tx_last = NULL;
+    ASSERT_DIE(!p->tx_pending_count);
+  }
+
+  free_page(btb);
 }
 
 static void
@@ -303,40 +395,33 @@ bmp_fire_tx(void *p_)
   if (!p->started)
     return;
 
-  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
+  int cnt = 0;
-    EMPTY_LIST(p->tx_queue),
+  for (struct bmp_tx_buffer *btb; btb = p->tx_pending; )
-    "Called BMP TX event handler when there is not any data to send"
-  );
-
-  size_t cnt = 0; // Counts max packets which we want to send per TX slot
-  struct bmp_data_node *tx_data;
-  struct bmp_data_node *tx_data_next;
-  WALK_LIST_DELSAFE(tx_data, tx_data_next, p->tx_queue)
   {
-    if (tx_data->data_size > p->sk->tbsize)
+    ASSERT_DIE(!p->sk->tbuf);
-    {
-      sk_set_tbsize(p->sk, tx_data->data_size);
-    }
 
-    size_t data_size = tx_data->data_size;
+    p->sk->tbuf = btb->data;
-    memcpy(p->sk->tbuf, tx_data->data, data_size);
+    u64 sz = btb->pos - btb->data;
-    mb_free(tx_data->data);
-    rem_node((node *) tx_data);
-    mb_free(tx_data);
 
-    if (sk_send(p->sk, data_size) <= 0)
+    p->tx_sent += sz;
+    p->tx_sent_total += sz;
+
+    if (p->tx_pending == p->tx_last)
+      p->tx_last = NULL;
+
+    p->tx_pending = btb->next;
+    p->tx_pending_count--;
+
+    if (sk_send(p->sk, sz) <= 0)
       return;
 
-    // BMP packets should be treat with lowest priority when scheduling sending
+    p->sk->tbuf = NULL;
-    // packets to target. That's why we want to send max. 32 packets per event
+    bmp_tx_buffer_free(p, btb);
-    // call
+
-    if (++cnt > 32)
+    if (cnt++ > 1024)
     {
       if (!ev_active(p->tx_ev))
-      {
         ev_send_loop(p->p.loop, p->tx_ev);
-      }
-
       return;
     }
   }
@@ -345,6 +430,13 @@ bmp_fire_tx(void *p_)
 static void
 bmp_tx(struct birdsock *sk)
 {
+  struct bmp_proto *p = sk->data;
+
+  struct bmp_tx_buffer *btb = SKIP_BACK(struct bmp_tx_buffer, data, sk->tbuf);
+  bmp_tx_buffer_free(p, btb);
+
+  sk->tbuf = NULL;
+
   bmp_fire_tx(sk->data);
 }
 
@@ -423,21 +515,31 @@ bmp_per_peer_hdr_serialize(buffer *stream, const bool is_global_instance_peer,
 }
 
 /* [4.6] Route Monitoring */
-static void
+static byte *
-bmp_route_monitor_msg_serialize(buffer *stream, const bool is_peer_global,
+bmp_route_monitor_msg_serialize(struct bmp_proto *p, const bool is_peer_global,
   const bool table_in_post_policy, const u32 peer_as, const u32 peer_bgp_id,
-  const bool as4_support, const ip_addr remote_addr, const byte *update_msg,
+  const bool as4_support, const ip_addr remote_addr, byte *update_msg,
   const size_t update_msg_size, btime timestamp)
 {
+  ASSERT_DIE(update_msg < &p->msgbuf[sizeof p->msgbuf]);
+
+  buffer stream;
+  STACK_BUFFER_INIT(stream, BMP_PER_PEER_HDR_SIZE + BMP_COMMON_HDR_SIZE);
+
   const size_t data_size = BMP_PER_PEER_HDR_SIZE + update_msg_size;
   u32 ts_sec = timestamp TO_S;
   u32 ts_usec = timestamp - (ts_sec S);
 
-  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
+  bmp_common_hdr_serialize(&stream, BMP_ROUTE_MONITOR, data_size);
-  bmp_common_hdr_serialize(stream, BMP_ROUTE_MONITOR, data_size);
+  bmp_per_peer_hdr_serialize(&stream, is_peer_global, table_in_post_policy,
-  bmp_per_peer_hdr_serialize(stream, is_peer_global, table_in_post_policy,
     as4_support, remote_addr, peer_as, peer_bgp_id, ts_sec, ts_usec);
-  bmp_put_data(stream, update_msg, update_msg_size);
+
+  size_t hdr_sz = stream.pos - stream.start;
+  ASSERT_DIE(update_msg >= &p->msgbuf[hdr_sz]);
+
+  byte *begin = &update_msg[-hdr_sz];
+  memcpy(begin, stream.start, hdr_sz);
+  return begin;
 }
 
 static void
@@ -455,6 +557,7 @@ bmp_peer_up_notif_msg_serialize(buffer *stream, const bool is_peer_global,
   bmp_per_peer_hdr_serialize(stream, is_peer_global,
     false /* TODO: Hardcoded pre-policy Adj-RIB-In */, as4_support, remote_addr,
     peer_as, peer_bgp_id, 0, 0); // 0, 0 - No timestamp provided
+
   bmp_put_ipa(stream, local_addr);
   bmp_put_u16(stream, local_port);
   bmp_put_u16(stream, remote_port);
@@ -467,15 +570,37 @@ bmp_peer_up_notif_msg_serialize(buffer *stream, const bool is_peer_global,
 static void
 bmp_peer_down_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 remote_addr, const byte *data, const size_t data_size)
+  const ip_addr remote_addr, const struct bmp_peer_down_info *info)
 {
-  const size_t payload_size = BMP_PER_PEER_HDR_SIZE + data_size;
+  const size_t data_size = BMP_PER_PEER_HDR_SIZE + 1 +
-  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + payload_size);
+    (((info->reason == BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION) ||
-  bmp_common_hdr_serialize(stream, BMP_PEER_DOWN_NOTIF, payload_size);
+      (info->reason == BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION)) ? (BGP_HEADER_LENGTH + 2 + info->length) :
+     (info->reason == BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION) ? 2 : 0);
+
+  bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
+  bmp_common_hdr_serialize(stream, BMP_PEER_DOWN_NOTIF, data_size);
   bmp_per_peer_hdr_serialize(stream, is_peer_global,
     false /* TODO: Hardcoded pre-policy adj RIB IN */,  as4_support, remote_addr,
     peer_as, peer_bgp_id, 0, 0); // 0, 0 - No timestamp provided
-  bmp_put_data(stream, data, data_size);
+
+  bmp_put_u8(stream, info->reason);
+
+  switch (info->reason)
+  {
+  case BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION:
+  case BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION:;
+    uint bgp_msg_length = BGP_HEADER_LENGTH + 2 + info->length;
+    bmp_buffer_need(stream, bgp_msg_length);
+    bmp_put_bgp_hdr(stream, bgp_msg_length, PKT_NOTIFICATION);
+    bmp_put_u8(stream, info->err_code);
+    bmp_put_u8(stream, info->err_subcode);
+    bmp_put_data(stream, info->data, info->length);
+    break;
+
+  case BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION:
+    bmp_put_u16(stream, info->fsm_code);
+    break;
+  }
 }
 
 
@@ -808,38 +933,31 @@ bmp_send_peer_up_notif_msg(struct bmp_proto *p, ea_list *bgp,
   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);
 
+  buffer payload = bmp_default_buffer(p);
   bmp_peer_up_notif_msg_serialize(&payload, is_global_instance_peer,
     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_schedule_tx_packet(p, payload.start, payload.pos - payload.start);
-  bmp_buffer_free(&payload);
 }
 
 static void
-bmp_route_monitor_put_update(struct bmp_proto *p, struct bmp_stream *bs, const byte *data, size_t length, btime timestamp)
+bmp_route_monitor_put_update(struct bmp_proto *p, struct bmp_stream *bs, byte *data, size_t length, btime timestamp)
 {
-  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 */
   ea_list *bgp = bs->bgp;
-  upd_msg->remote_as = ea_get_int(bgp, &ea_bgp_rem_as, 0);
+  const byte *start = bmp_route_monitor_msg_serialize(p,
-  upd_msg->remote_id = ea_get_int(bgp, &ea_bgp_rem_id, 0);
+      bmp_is_peer_global_instance(bgp),
-  upd_msg->remote_ip = ea_get_ip(bgp, &ea_bgp_rem_ip, IPA_NONE);
+      bmp_stream_policy(bs),
-  upd_msg->timestamp = timestamp;
+      ea_get_int(bgp, &ea_bgp_rem_as, 0),
-  upd_msg->global_peer = bmp_is_peer_global_instance(bgp);
+      ea_get_int(bgp, &ea_bgp_rem_id, 0),
-  upd_msg->policy = bmp_stream_policy(bs);
+      true,
+      ea_get_ip(bgp, &ea_bgp_rem_ip, IPA_NONE),
+      data,
+      length,
+      timestamp
+      );
 
-  /* Kick the commit */
+  bmp_schedule_tx_packet(p, start, (data - start) + length);
-  if (!ev_active(p->update_ev))
-    ev_send_loop(p->p.loop, p->update_ev);
 }
 
 static void
@@ -854,56 +972,25 @@ bmp_route_monitor_notify(struct bmp_proto *p, struct bgp_proto *bgp_p, u32 afi,
   if (!bs)
     return;
 
-  byte buf[BGP_MAX_EXT_MSG_LENGTH];
+  byte *bufend = &p->msgbuf[sizeof p->msgbuf];
-  byte *end = bgp_bmp_encode_rte(bs->sender, bgp_p, buf, new);
+  byte *begin = bufend - BGP_MAX_EXT_MSG_LENGTH;
+  byte *end = bgp_bmp_encode_rte(bs->sender, bgp_p, begin, bufend, 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);
+    bmp_route_monitor_put_update(p, bs, begin, end - begin, timestamp);
   else
     log(L_WARN "%s: Cannot encode update for %N", p->p.name, new->net);
 }
 
-static void
-bmp_route_monitor_commit(void *p_)
-{
-  struct bmp_proto *p = p_;
-
-  if (!p->started)
-    return;
-
-  buffer payload
-    = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
-
-  struct bmp_data_node *data, *data_next;
-  WALK_LIST_DELSAFE(data, data_next, p->update_msg_queue)
-  {
-    bmp_route_monitor_msg_serialize(&payload,
-      data->global_peer, data->policy,
-      data->remote_as, data->remote_id, true,
-      data->remote_ip, data->data, data->data_size,
-      data->timestamp);
-
-    bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
-
-    bmp_buffer_flush(&payload);
-
-    mb_free(data->data);
-    rem_node(&data->n);
-    mb_free(data);
-  }
-
-  bmp_buffer_free(&payload);
-}
-
 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", 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 *rx_end_payload = p->msgbuf + BMP_PER_PEER_HDR_SIZE + BMP_COMMON_HDR_SIZE;
   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
@@ -915,7 +1002,7 @@ 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, ea_list *bgp,
-  const byte *data, const size_t data_size)
+			     const struct bmp_peer_down_info *info)
 {
   ASSERT(p->started);
 
@@ -932,7 +1019,7 @@ bmp_send_peer_down_notif_msg(struct bmp_proto *p, ea_list *bgp,
     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);
+  buffer payload = bmp_default_buffer(p);
   bmp_peer_down_notif_msg_serialize(
       &payload,
       is_global_instance_peer,
@@ -940,12 +1027,9 @@ bmp_send_peer_down_notif_msg(struct bmp_proto *p, ea_list *bgp,
       ea_get_int(bgp, &ea_bgp_rem_id, 0),
       remote_caps,
       *((ip_addr *) ea_get_adata(bgp, &ea_bgp_rem_ip)->data),
-      data,
+      info
-      data_size
       );
-  bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
+  bmp_schedule_tx_packet(p, payload.start, payload.pos - payload.start);
-
-  bmp_buffer_free(&payload);
 }
 
 static void
@@ -960,48 +1044,32 @@ bmp_peer_down_(struct bmp_proto *p, ea_list *bgp, struct bgp_session_close_ad *b
 
   TRACE(D_STATES, "Peer down for %s", ea_find(bgp, &ea_name)->u.ad->data);
 
-  uint bmp_code = 0;
+  struct bmp_peer_down_info info = {
-  uint fsm_code = 0;
+    .err_code = bscad->notify_code,
+    .err_subcode = bscad->notify_subcode,
+    .data = bscad->data,
+    .length = bscad->ad.length - sizeof *bscad + sizeof bscad->ad,
+  };
 
   switch (bscad->last_error_class)
   {
   case BE_BGP_RX:
-    bmp_code = BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION;
+    info.reason = BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION;
     break;
 
   case BE_BGP_TX:
   case BE_AUTO_DOWN:
   case BE_MAN_DOWN:
-    bmp_code = BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION;
+    info.reason = BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION;
     break;
 
   default:
-    bmp_code = BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION;
+    info.reason = BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION;
+    info.length = 0;
     break;
   }
 
-  uint length = bscad->ad.length - sizeof *bscad + sizeof bscad->ad;
+  bmp_send_peer_down_notif_msg(p, bgp, &info);
-  buffer payload = bmp_buffer_alloc(p->buffer_mpool, 1 + BGP_HEADER_LENGTH + 2 + length);
-  bmp_put_u8(&payload, bmp_code);
-
-  switch (bmp_code)
-  {
-  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, 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:
-    bmp_put_u16(&payload, fsm_code);
-    break;
-  }
-
-  bmp_send_peer_down_notif_msg(p, bgp, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
-
-  bmp_buffer_free(&payload);
 
   bmp_remove_peer(p, bp);
 }
@@ -1015,18 +1083,21 @@ bmp_send_termination_msg(struct bmp_proto *p,
                                      + BMP_TERM_INFO_LEN_FIELD_SIZE
                                      + BMP_TERM_REASON_CODE_SIZE;
   const size_t term_msg_size = BMP_COMMON_HDR_SIZE + term_msg_hdr_size;
-  buffer stream = bmp_buffer_alloc(p->buffer_mpool, term_msg_size);
+  buffer stream = bmp_default_buffer(p);
+  bmp_buffer_need(&stream, term_msg_size);
+
   bmp_common_hdr_serialize(&stream, BMP_TERM_MSG, term_msg_hdr_size);
   bmp_put_u16(&stream, BMP_TERM_INFO_REASON);
   bmp_put_u16(&stream, BMP_TERM_REASON_CODE_SIZE); // 2-byte code indication the reason
   bmp_put_u16(&stream, reason);
-  memcpy(p->sk->tbuf, bmp_buffer_data(&stream), bmp_buffer_pos(&stream));
-  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    sk_send(p->sk, bmp_buffer_pos(&stream)) < 0,
-    "Failed to send BMP termination message"
-    );
 
-  bmp_buffer_free(&stream);
+  if (p->sk->tbuf)
+    bmp_tx_buffer_free(p, SKIP_BACK(struct bmp_tx_buffer, data, p->sk->tbuf));
+
+  p->sk->tbuf = stream.start;
+  if (sk_send(p->sk, stream.pos - stream.start) < 0)
+    log(L_WARN "%s: Cannot send BMP termination message", p->p.name);
+  p->sk->tbuf = NULL;
 }
 
 static void
@@ -1145,10 +1216,9 @@ bmp_startup(struct bmp_proto *p)
   proto_notify_state(&p->p, PS_UP);
 
   /* Send initiation message */
-  buffer payload = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
+  buffer payload = bmp_default_buffer(p);
   bmp_init_msg_serialize(&payload, p->sys_descr, p->sys_name);
-  bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
+  bmp_schedule_tx_packet(p, payload.start, payload.pos - payload.start);
-  bmp_buffer_free(&payload);
 
   /* Send Peer Up messages */
   u32 length;
@@ -1197,6 +1267,7 @@ bmp_down(struct bmp_proto *p)
 {
   ASSERT(p->started);
   p->started = false;
+  p->tx_sent = 0;
 
   TRACE(D_EVENTS, "BMP session closed");
 
@@ -1233,7 +1304,6 @@ bmp_connect(struct bmp_proto *p)
   sk->dport = p->station_port;
   sk->ttl = IP4_MAX_TTL;
   sk->tos = IP_PREC_INTERNET_CONTROL;
-  sk->tbsize = BGP_TX_BUFFER_EXT_SIZE;
   sk->tx_hook = bmp_connected;
   sk->err_hook = bmp_sock_err;
 
@@ -1288,6 +1358,26 @@ bmp_sock_err(sock *sk, int err)
     proto_notify_state(&p->p, PS_START);
 }
 
+static void
+bmp_tx_overflow(void *_p)
+{
+  struct bmp_proto *p = _p;
+  if (p->tx_pending_count < p->tx_pending_limit)
+    return;
+
+  p->sock_err = 0;
+
+  log(L_ERR "%s: Connection stalled", p->p.name);
+
+  if (p->started)
+    bmp_down(p);
+
+  bmp_close_socket(p);
+  tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);
+
+  proto_notify_state(&p->p, PS_START);
+}
+
 /* BMP connect timeout event - switch from Idle/Connect state to Connect state */
 static void
 bmp_connection_retry(timer *t)
@@ -1304,6 +1394,24 @@ bmp_connection_retry(timer *t)
 static void
 bmp_close_socket(struct bmp_proto *p)
 {
+  if (p->sk && p->sk->tbuf)
+    bmp_tx_buffer_free(p, SKIP_BACK(struct bmp_tx_buffer, data, p->sk->tbuf));
+
+  struct bmp_tx_buffer *btb = p->tx_pending;
+  while (btb)
+  {
+    p->tx_pending_count--;
+
+    struct bmp_tx_buffer *next = btb->next;
+    bmp_tx_buffer_free(p, btb);
+    btb = next;
+  }
+
+  p->tx_pending = NULL;
+
+  ASSERT_DIE(!p->tx_last);
+  ASSERT_DIE(!p->tx_pending_count);
+
   rfree(p->sk);
   p->sk = NULL;
 }
@@ -1379,6 +1487,7 @@ 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;
+  p->tx_pending_limit = cf->tx_pending_limit;
 
   return P;
 }
@@ -1392,20 +1501,20 @@ bmp_start(struct proto *P)
 {
   struct bmp_proto *p = (void *) P;
 
-  p->buffer_mpool = rp_new(P->pool, proto_domain(&p->p), "BMP Buffer");
-  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->tx_pending = NULL;
+  p->tx_pending_count = 0;
+  p->tx_overflow_event = ev_new_init(p->p.pool, bmp_tx_overflow, p);
   p->connect_retry_timer = tm_new_init(p->p.pool, bmp_connection_retry, p, 0, 0);
   p->sk = NULL;
 
+  resource *r = ralloc(P->pool, &bmp_tx_resource_class);
+  SKIP_BACK(struct bmp_tx_resource, r, r)->p = p;
+
   HASH_INIT(p->peer_map, P->pool, 4);
   HASH_INIT(p->stream_map, P->pool, 4);
   HASH_INIT(p->table_map, P->pool, 4);
 
-  init_list(&p->tx_queue);
-  init_list(&p->update_msg_queue);
   p->started = false;
   p->sock_err = 0;
 
@@ -1423,6 +1532,7 @@ bmp_shutdown(struct proto *P)
   {
     bmp_send_termination_msg(p, BMP_TERM_REASON_ADM);
     bmp_down(p);
+    bmp_close_socket(p);
   }
 
   p->sock_err = 0;
@@ -1452,6 +1562,9 @@ bmp_reconfigure(struct proto *P, struct proto_config *CF)
   /* We must update our copy of configuration ptr */
   p->cf = new;
 
+  /* Reconfigure tx buffer size limits */
+  p->tx_pending_limit = new->tx_pending_limit;
+
   return 1;
 }
 
@@ -1490,6 +1603,13 @@ bmp_show_proto_info(struct proto *P)
 
     if (p->sock_err)
       cli_msg(-1006, "  %-19s %M", "Last error:", p->sock_err);
+
+    cli_msg(-1006, "  %-19s % 9sB (limit %sB)", "Pending TX:",
+	fmt_order(p->tx_pending_count * (u64) page_size, 1, 10000),
+	fmt_order(p->tx_pending_limit * (u64) page_size, 1, 10000));
+
+    cli_msg(-1006, "  %-19s % 9sB", "Session TX:", fmt_order(p->tx_sent, 1, 10000));
+    cli_msg(-1006, "  %-19s % 9sB", "Total TX:", fmt_order(p->tx_sent_total, 1, 10000));
   }
 }
 

proto/bmp/bmp.h

@@ -16,11 +16,22 @@
 #include "lib/event.h"
 #include "lib/hash.h"
 #include "lib/socket.h"
+#include "proto/bgp/bgp.h"
 #include "proto/bmp/map.h"
 
 // Max length of MIB-II description object
 #define MIB_II_STR_LEN 255
 
+// Total size of Common Header
+#define BMP_COMMON_HDR_SIZE 6
+
+/* BMP Per-Peer Header [RFC 7854 - Section 4.2] */
+// Total size of Per-Peer Header
+#define BMP_PER_PEER_HDR_SIZE 42
+
+// Maximum length of BMP message altogether
+#define BMP_MSGBUF_LEN (BGP_MAX_EXT_MSG_LENGTH + BMP_PER_PEER_HDR_SIZE + BMP_COMMON_HDR_SIZE + 1)
+
 // The following fields of this structure controls whether there will be put
 // specific routes into Route Monitoring message and send to BMP collector
 struct monitoring_rib {
@@ -38,6 +49,7 @@ struct bmp_config {
   u16 station_port;                   // Monitoring station TCP port
   bool monitoring_rib_in_pre_policy;  // Route monitoring pre-policy Adj-Rib-In
   bool monitoring_rib_in_post_policy;  // Route monitoring post-policy Adj-Rib-In
+  uint tx_pending_limit;	      // Maximum on pending TX buffer count
 };
 
 /* Forward declarations */
@@ -63,17 +75,20 @@ struct bmp_proto {
   struct monitoring_rib monitoring_rib;
   // Below fields are for internal use
   // struct bmp_peer_map bgp_peers;   // Stores 'bgp_proto' structure per BGP peer
-  pool *buffer_mpool;              // Memory pool used for BMP buffer allocations
+  struct bmp_tx_buffer *tx_pending;// This buffer waits for socket to flush
-  pool *tx_mem_pool;               // Memory pool used for packet allocations designated to BMP collector
+  struct bmp_tx_buffer *tx_last;   // This buffer is the last to flush 
-  pool *update_msg_mem_pool;       // Memory pool used for BPG UPDATE MSG allocations
+  uint tx_pending_count;	   // How many buffers waiting for flush
-  list tx_queue;                   // Stores queued packets going to be sent
+  uint tx_pending_limit;	   // Maximum on buffer count
+  u64 tx_sent;			   // Amount of data sent
+  u64 tx_sent_total;		   // Amount of data sent accumulated over reconnections
+  event *tx_overflow_event;	   // Too many buffers waiting for flush
   timer *connect_retry_timer;      // Timer for retrying connection to the BMP collector
-  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;
+  byte msgbuf[BMP_MSGBUF_LEN];     // Buffer for preparing the messages before sending them out
 };
 
 struct bmp_peer {

proto/bmp/buffer.c

@@ -1,65 +0,0 @@
-/*
- *	BIRD -- The BGP Monitoring Protocol (BMP)
- *
- *	(c) 2020 Akamai Technologies, Inc. (Pawel Maslanka, pmaslank@akamai.com)
- *
- *	Can be freely distributed and used under the terms of the GNU GPL.
- */
-
-#include "proto/bmp/buffer.h"
-
-buffer
-bmp_buffer_alloc(pool *ppool, const size_t n)
-{
-  buffer buf;
-  buf.start = mb_alloc(ppool, n);
-  buf.pos = buf.start;
-  buf.end = buf.start + n;
-  return buf;
-}
-
-void
-bmp_buffer_free(buffer *buf)
-{
-  mb_free(buf->start);
-  buf->start = buf->pos = buf->end = NULL;
-}
-
-/**
- * @brief bmp_buffer_grow
- * @param buf - buffer to grow
- * @param n   - required amount of available space
- * Resize buffer in a way that there is at least @n bytes of available space.
- */
-static void
-bmp_buffer_grow(buffer *buf, const size_t n)
-{
-  size_t pos = bmp_buffer_pos(buf);
-  size_t size = bmp_buffer_size(buf);
-  size_t req = pos + n;
-
-  while (size < req)
-    size = size * 3 / 2;
-
-  buf->start = mb_realloc(buf->start, size);
-  buf->pos = buf->start + pos;
-  buf->end = buf->start + size;
-}
-
-void
-bmp_buffer_need(buffer *buf, const size_t n)
-{
-  if (bmp_buffer_avail(buf) < n)
-    bmp_buffer_grow(buf, n);
-}
-
-void
-bmp_put_data(buffer *buf, const void *src, const size_t n)
-{
-  if (!n)
-    return;
-
-  bmp_buffer_need(buf, n);
-  memcpy(buf->pos, src, n);
-  buf->pos += n;
-}

proto/bmp/buffer.h

@@ -1,77 +0,0 @@
-/*
- *	BIRD -- The BGP Monitoring Protocol (BMP)
- *
- *	(c) 2020 Akamai Technologies, Inc. (Pawel Maslanka, pmaslank@akamai.com)
- *
- *	Can be freely distributed and used under the terms of the GNU GPL.
- */
-
-#ifndef _BIRD_BMP_BUFFER_H_
-#define _BIRD_BMP_BUFFER_H_
-
-#include "proto/bmp/bmp.h"
-
-#include <stdlib.h>
-
-#include "lib/resource.h"
-
-buffer
-bmp_buffer_alloc(pool *ppool, const size_t n);
-
-void
-bmp_buffer_free(buffer *buf);
-
-static inline void
-bmp_buffer_flush(buffer *buf)
-{
-  buf->pos = buf->start;
-}
-
-static inline size_t
-bmp_buffer_size(const buffer *buf)
-{
-  return buf->end - buf->start;
-}
-
-static inline size_t
-bmp_buffer_avail(const buffer *buf)
-{
-  return buf->end - buf->pos;
-}
-
-static inline size_t
-bmp_buffer_pos(const buffer *buf)
-{
-  return buf->pos - buf->start;
-}
-
-static inline byte *
-bmp_buffer_data(const buffer *buf)
-{
-  return buf->start;
-}
-
-void
-bmp_buffer_need(buffer *buf, const size_t n);
-
-// Idea for following macros has been taken from |proto/mrt/mrt.c|
-#define BMP_DEFINE_PUT_FUNC(S, T)                               \
-  static inline void                                            \
-  bmp_put_##S(buffer *b, const T x)                             \
-  {                                                             \
-    bmp_buffer_need(b, sizeof(T));                              \
-    put_##S(b->pos, x);                                    \
-    b->pos += sizeof(T);                                   \
-  }
-
-BMP_DEFINE_PUT_FUNC(u8, u8)
-BMP_DEFINE_PUT_FUNC(u16, u16)
-BMP_DEFINE_PUT_FUNC(u32, u32)
-BMP_DEFINE_PUT_FUNC(u64, u64)
-BMP_DEFINE_PUT_FUNC(ip4, ip4_addr)
-BMP_DEFINE_PUT_FUNC(ip6, ip6_addr)
-
-void
-bmp_put_data(buffer *buf, const void *src, const size_t n);
-
-#endif /* _BIRD_BMP_BUFFER_H_ */

proto/bmp/config.Y

@@ -28,6 +28,7 @@ bmp_proto_start: proto_start BMP {
      this_proto->loop_order = DOMAIN_ORDER(proto);
      BMP_CFG->sys_descr = "Not defined";
      BMP_CFG->sys_name = "Not defined";
+     BMP_CFG->tx_pending_limit = (1 << 30) / page_size;
    }
  ;
 
@@ -72,6 +73,10 @@ bmp_proto:
  | bmp_proto MONITORING RIB IN POST_POLICY bool ';' {
      BMP_CFG->monitoring_rib_in_post_policy = $6;
    }
+ | bmp_proto TX BUFFER LIMIT expr ';' {
+     BMP_CFG->tx_pending_limit = $5 * (u64) (1 << 20) / page_size;
+     if ($5 < 1) cf_error("BMP TX buffer limit must be at least 1 megabyte");
+   }
  ;
 
 CF_CODE