BMP: overflow checker made configurable

Also added documentation for these two options.

doc/bird.sgml

@@ -3739,6 +3739,22 @@ 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">tx buffer <m/number/</tag>
+	How big the TX buffer is for holding BMP data. Must be at least 16KB
+        long to have some slack for routes and open messages, as these may be
+        (post-policy) longer than the default 4KB. Recommended and default value
+	is 1 MB and you should not change it unless you know what you are doing.
+
+	<tag><label id="bmp-tx-buffer-count">tx buffer count <m/number/</tag>
+	How many TX buffers are held at most. This is effectively a multiplier
+        for the <cf/tx buffer/ option, capping the buffer sizes in case BMP
+	receiver gets clogged. When overflown, BMP restarts. Default: 1024.
+</descrip>
+
 <sect1>Example
 <label id="bmp-exam">
 
@@ -3752,6 +3768,9 @@ protocol bmp {
 
 	# Monitor accepted routes (passed import filters)
 	monitoring rib in post_policy;
+
+	# Allow only 64M of pending data
+	tx buffer count 64;
 }
 </code>
 

proto/bmp/bmp.c

@@ -230,7 +230,14 @@ struct bmp_tx_buffer {
   resource r;
   node n;
   byte *end;
-  byte buf[0xfff00];
+  byte buf[0];
+};
+
+struct bmp_tx_buffer_class {
+  resource r;
+  struct resclass class;
+  uint counter;
+  bool deprecated;
 };
 
 // Stores necessary any data in list
@@ -282,8 +289,33 @@ bmp_init_msg_serialize(buffer *stream, const char *sys_descr, const char *sys_na
   bmp_info_tlv_hdr_serialize(stream, BMP_INFO_TLV_TYPE_SYS_NAME, sys_name);
 }
 
-static void btb_free(resource *r UNUSED) {
+static void
+btbm_free(resource *r)
+{
+  struct bmp_tx_buffer_class *btbc = SKIP_BACK(struct bmp_tx_buffer_class, r, r);
+  ASSERT_DIE(btbc->counter == 0);
+}
+
+static void
+btbm_dump(resource *r)
+{
+  struct bmp_tx_buffer_class *btbc = SKIP_BACK(struct bmp_tx_buffer_class, r, r);
+  debug("allocated %u blocks%s\n", btbc->counter, btbc->deprecated ? " (deprecated)" : "");
+}
+
+static struct resclass bmp_tx_buffer_metaclass = {
+  .name = "BMP TX buffer meta",
+  .size = sizeof(struct bmp_tx_buffer_class),
+  .free = btbm_free,
+  .dump = btbm_dump,
+};
+
+static void btb_free(resource *r) {
   UNUSED struct bmp_tx_buffer *btb = SKIP_BACK(struct bmp_tx_buffer, r, r);
+  struct bmp_tx_buffer_class *class = SKIP_BACK(struct bmp_tx_buffer_class, class, r->class);
+  ASSERT_DIE(class->counter);
+  if (!--class->counter && class->deprecated)
+    rfree(class);
 }
 
 static void
@@ -293,28 +325,37 @@ btb_dump(resource *r)
   debug("used %u bytes\n", btb->end - btb->buf);
 }
 
+static uint
+btb_bufsize(struct bmp_tx_buffer *btb)
+{
+  return btb->r.class->size - sizeof *btb;
+}
+
 static struct resmem
 btb_memsize(resource *r)
 {
   struct bmp_tx_buffer *btb = SKIP_BACK(struct bmp_tx_buffer, r, r);
   return (struct resmem) {
     .effective = btb->end - btb->buf,
-    .overhead = (sizeof *btb) - (btb->end - btb->buf),
+    .overhead = r->class->size - (btb->end - btb->buf),
   };
 }
 
-static struct resclass bmp_tx_buffer_class = {
+static struct bmp_tx_buffer_class *
+btbm_new(struct bmp_proto *p, uint size)
+{
+  ASSERT_DIE(size > sizeof (struct bmp_tx_buffer) + 4096);
+  struct bmp_tx_buffer_class *c = ralloc(p->p.pool, &bmp_tx_buffer_metaclass);
+  c->class = (struct resclass) {
     .name = "BMP TX buffer",
-  .size = sizeof(struct bmp_tx_buffer),
+    .size = size,
     .free = btb_free,
     .dump = btb_dump,
     .memsize = btb_memsize,
-};
+  };
-
+  c->counter = 0;
-static bool
+  c->deprecated = false;
-bmp_check_tx_overflow(u64 cnt)
+  return c;
-{
-  return cnt * bmp_tx_buffer_class.size > (1ULL << 30);
 }
 
 static void
@@ -326,15 +367,23 @@ bmp_schedule_tx_packet(struct bmp_proto *p, const byte *payload, const size_t si
 
   struct bmp_tx_buffer *btb = EMPTY_LIST(p->tx_queue) ? NULL :
     SKIP_BACK(struct bmp_tx_buffer, n, TAIL(p->tx_queue));
-  if (btb && (btb->end + size > btb->buf + sizeof btb->buf))
+  if (btb && (size + btb->end - btb->buf > btb_bufsize(btb)))
     btb = NULL;
 
+  if (size + sizeof *btb > p->tx_pcls->class.size)
+  {
+    log(L_ERR "%s: configured too small buffer (%u) for message size %lu",
+	p->tx_pcls->class.size, size + sizeof *btb);
+    return;
+  }
+
   if (!btb)
   {
-    if (bmp_check_tx_overflow(++p->tx_pending_count))
+    if (++p->tx_pending_count >= p->tx_pending_limit)
       ev_schedule(p->tx_overflow_event);
 
-    btb = ralloc(p->tx_mem_pool, &bmp_tx_buffer_class);
+    btb = ralloc(p->tx_mem_pool, &p->tx_pcls->class);
+    p->tx_pcls->counter++;
     btb->end = btb->buf;
     btb->n = (node) {};
     add_tail(&p->tx_queue, &btb->n);
@@ -1215,7 +1264,7 @@ static void
 bmp_tx_overflow(void *_p)
 {
   struct bmp_proto *p = _p;
-  if (!bmp_check_tx_overflow(p->tx_pending_count))
+  if (p->tx_pending_count < p->tx_pending_limit)
     return;
 
   p->sock_err = 0;
@@ -1306,6 +1355,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;
+  p->tx_pcls = btbm_new(p, cf->tx_bufsize);
+  p->tx_pending_limit = cf->tx_pending_limit;
 
   return P;
 }
@@ -1381,6 +1432,15 @@ 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 */
+  if (new->tx_bufsize != p->tx_pcls->class.size)
+  {
+    p->tx_pcls->deprecated = true;
+    p->tx_pcls = btbm_new(p, new->tx_bufsize);
+  }
+
+  p->tx_pending_limit = new->tx_pending_limit;
+
   return 1;
 }
 

proto/bmp/bmp.h

@@ -38,6 +38,8 @@ 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_bufsize;		      // Pending TX buffer size
+  uint tx_pending_limit;	      // Maximum on pending TX buffer count
 };
 
 /* Forward declarations */
@@ -68,7 +70,9 @@ struct bmp_proto {
   pool *tx_mem_pool;               // Memory pool used for packet allocations designated to BMP collector
   list tx_queue;                   // Stores queued packets going to be sent
   struct bmp_tx_buffer *tx_pending;// This buffer waits for socket to flush
+  struct bmp_tx_buffer_class *tx_pcls; // Class used for tx_pending allocations
   uint tx_pending_count;	   // How many buffers waiting for flush
+  uint tx_pending_limit;	   // Maximum on buffer count
   event *tx_overflow_event;	   // Too many buffers waiting for flush
   timer *connect_retry_timer;      // Timer for retrying connection to the BMP collector
   bool started;                    // Flag that stores running status of BMP instance
@@ -98,7 +102,6 @@ struct bmp_table {
   u32 uc;
 };
 
-
 #ifdef CONFIG_BMP
 
 /**

proto/bmp/config.Y

@@ -27,6 +27,8 @@ bmp_proto_start: proto_start BMP {
      this_proto = proto_config_new(&proto_bmp, $1);
      BMP_CFG->sys_descr = "Not defined";
      BMP_CFG->sys_name = "Not defined";
+     BMP_CFG->tx_bufsize = 1048576;
+     BMP_CFG->tx_pending_limit = 1024;
    }
  ;
 
@@ -71,6 +73,14 @@ bmp_proto:
  | bmp_proto MONITORING RIB IN POST_POLICY bool ';' {
      BMP_CFG->monitoring_rib_in_post_policy = $6;
    }
+ | bmp_proto TX BUFFER expr ';' {
+     BMP_CFG->tx_bufsize = $4;
+     if ($4 < 16384) cf_error("BMP TX buffer must be at least 16384 bytes long");
+   }
+ | bmp_proto TX BUFFER COUNT expr ';' {
+     BMP_CFG->tx_pending_limit = $5;
+     if ($5 < 2) cf_error("BMP must allow at least 2 pending TX buffers");
+   }
  ;
 
 CF_CODE