buffers: major cleanup and improvements

This patch introduces following changes:
- deprecated free lists which are not used and not compatible
  with external buffer managers (i.e. DPDK)
- introduces native support for per-numa buffer pools
- significantly improves performance of buffer alloc and free

Change-Id: I4a8e723ae47056717afd6cac0efe87cb731b5be7
Signed-off-by: Damjan Marion <damarion@cisco.com>

6 files changed, 35 insertions, 193 deletions

diff --git a/src/plugins/dpdk/device/cli.c b/src/plugins/dpdk/device/cli.c
index bbe367f7ec5..209f4ed51ed 100644
--- a/src/plugins/dpdk/device/cli.c
+++ b/src/plugins/dpdk/device/cli.c
@@ -382,27 +382,27 @@ static clib_error_t *
 show_dpdk_buffer (vlib_main_t * vm, unformat_input_t * input,
 		  vlib_cli_command_t * cmd)
 {
-  struct rte_mempool *rmp;
-  int i;
+  vlib_buffer_main_t *bm = vm->buffer_main;
+  vlib_buffer_pool_t *bp;
 
-  for (i = 0; i < vec_len (dpdk_main.pktmbuf_pools); i++)
-    {
-      rmp = dpdk_main.pktmbuf_pools[i];
-      if (rmp)
-	{
-	  unsigned count = rte_mempool_avail_count (rmp);
-	  unsigned free_count = rte_mempool_in_use_count (rmp);
+  vec_foreach (bp, bm->buffer_pools)
+  {
+    struct rte_mempool *rmp = dpdk_mempool_by_buffer_pool_index[bp->index];
+    if (rmp)
+      {
+	unsigned count = rte_mempool_avail_count (rmp);
+	unsigned free_count = rte_mempool_in_use_count (rmp);
 
-	  vlib_cli_output (vm,
-			   "name=\"%s\"  available = %7d allocated = %7d total = %7d\n",
-			   rmp->name, (u32) count, (u32) free_count,
-			   (u32) (count + free_count));
-	}
-      else
-	{
-	  vlib_cli_output (vm, "rte_mempool is NULL (!)\n");
-	}
-    }
+	vlib_cli_output (vm,
+			 "name=\"%s\"  available = %7d allocated = %7d total = %7d\n",
+			 rmp->name, (u32) count, (u32) free_count,
+			 (u32) (count + free_count));
+      }
+    else
+      {
+	vlib_cli_output (vm, "rte_mempool is NULL (!)\n");
+      }
+  }
   return 0;
 }
 
@@ -2018,59 +2018,6 @@ VLIB_CLI_COMMAND (show_vpe_version_command, static) = {
 };
 /* *INDENT-ON* */
 
-#if CLI_DEBUG
-
-static clib_error_t *
-dpdk_validate_buffers_fn (vlib_main_t * vm, unformat_input_t * input,
-			  vlib_cli_command_t * cmd_arg)
-{
-  u32 n_invalid_bufs = 0, uninitialized = 0;
-  u32 is_poison = 0, is_test = 0;
-  while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
-    {
-      if (unformat (input, "poison"))
-	is_poison = 1;
-      else if (unformat (input, "trajectory"))
-	is_test = 1;
-      else
-	return clib_error_return (0, "unknown input `%U'",
-				  format_unformat_error, input);
-    }
-
-  if (VLIB_BUFFER_TRACE_TRAJECTORY == 0)
-    {
-      vlib_cli_output (vm, "Trajectory not enabled. Recompile with "
-		       "VLIB_BUFFER_TRACE_TRAJECTORY 1");
-      return 0;
-    }
-  if (is_poison)
-    {
-      dpdk_buffer_poison_trajectory_all ();
-    }
-  if (is_test)
-    {
-      n_invalid_bufs = dpdk_buffer_validate_trajectory_all (&uninitialized);
-      if (!n_invalid_bufs)
-	vlib_cli_output (vm, "All buffers are valid %d uninitialized",
-			 uninitialized);
-      else
-	vlib_cli_output (vm, "Found %d invalid buffers and %d uninitialized",
-			 n_invalid_bufs, uninitialized);
-    }
-  return 0;
-}
-
-/* *INDENT-OFF* */
-VLIB_CLI_COMMAND (test_dpdk_buffers_command, static) =
-{
-  .path = "test dpdk buffers",
-  .short_help = "test dpdk buffers [poison] [trajectory]",
-  .function = dpdk_validate_buffers_fn,
-};
-/* *INDENT-ON* */
-
-#endif
-
 clib_error_t *
 dpdk_cli_init (vlib_main_t * vm)
 {
diff --git a/src/plugins/dpdk/device/common.c b/src/plugins/dpdk/device/common.c
index 25d5802899f..b239616b7d9 100644
--- a/src/plugins/dpdk/device/common.c
+++ b/src/plugins/dpdk/device/common.c
@@ -40,6 +40,7 @@ void
 dpdk_device_setup (dpdk_device_t * xd)
 {
   dpdk_main_t *dm = &dpdk_main;
+  vlib_main_t *vm = vlib_get_main ();
   vnet_main_t *vnm = vnet_get_main ();
   vnet_sw_interface_t *sw = vnet_get_sw_interface (vnm, xd->sw_if_index);
   vnet_hw_interface_t *hi = vnet_get_hw_interface (vnm, xd->hw_if_index);
@@ -116,26 +117,23 @@ dpdk_device_setup (dpdk_device_t * xd)
 			CLIB_CACHE_LINE_BYTES);
   for (j = 0; j < xd->rx_q_used; j++)
     {
-      dpdk_mempool_private_t *privp;
       uword tidx = vnet_get_device_input_thread_index (dm->vnet_main,
 						       xd->hw_if_index, j);
       unsigned lcore = vlib_worker_threads[tidx].cpu_id;
       u16 socket_id = rte_lcore_to_socket_id (lcore);
+      u8 bpidx = vlib_buffer_pool_get_default_for_numa (vm, socket_id);
+      vlib_buffer_pool_t *bp = vlib_get_buffer_pool (vm, bpidx);
+      struct rte_mempool *mp = dpdk_mempool_by_buffer_pool_index[bpidx];
 
-      rv =
-	rte_eth_rx_queue_setup (xd->port_id, j, xd->nb_rx_desc,
-				xd->cpu_socket, 0,
-				dm->pktmbuf_pools[socket_id]);
+      rv = rte_eth_rx_queue_setup (xd->port_id, j, xd->nb_rx_desc,
+				   xd->cpu_socket, 0, mp);
 
       /* retry with any other CPU socket */
       if (rv < 0)
-	rv =
-	  rte_eth_rx_queue_setup (xd->port_id, j,
-				  xd->nb_rx_desc, SOCKET_ID_ANY, 0,
-				  dm->pktmbuf_pools[socket_id]);
+	rv = rte_eth_rx_queue_setup (xd->port_id, j, xd->nb_rx_desc,
+				     SOCKET_ID_ANY, 0, mp);
 
-      privp = rte_mempool_get_priv (dm->pktmbuf_pools[socket_id]);
-      xd->buffer_pool_for_queue[j] = privp->buffer_pool_index;
+      xd->buffer_pool_for_queue[j] = bp->index;
 
       if (rv < 0)
 	dpdk_device_error (xd, "rte_eth_rx_queue_setup", rv);
diff --git a/src/plugins/dpdk/device/device.c b/src/plugins/dpdk/device/device.c
index 86b9a50f41b..e3520fc720e 100644
--- a/src/plugins/dpdk/device/device.c
+++ b/src/plugins/dpdk/device/device.c
@@ -127,11 +127,9 @@ dpdk_validate_rte_mbuf (vlib_main_t * vm, vlib_buffer_t * b,
       mb->pkt_len = b->current_length;
       mb->data_off = VLIB_BUFFER_PRE_DATA_SIZE + b->current_data;
       first_mb->nb_segs++;
-      if (PREDICT_FALSE (b->n_add_refs))
-	{
-	  rte_mbuf_refcnt_update (mb, b->n_add_refs);
-	  b->n_add_refs = 0;
-	}
+      if (PREDICT_FALSE (b->ref_count > 1))
+	mb->pool =
+	  dpdk_no_cache_mempool_by_buffer_pool_index[b->buffer_pool_index];
     }
 }
 
diff --git a/src/plugins/dpdk/device/dpdk.h b/src/plugins/dpdk/device/dpdk.h
index 425cf265873..14630967ea5 100644
--- a/src/plugins/dpdk/device/dpdk.h
+++ b/src/plugins/dpdk/device/dpdk.h
@@ -56,8 +56,6 @@
 #include <vlib/pci/pci.h>
 #include <vnet/flow/flow.h>
 
-#define NB_MBUF   (16<<10)
-
 extern vnet_device_class_t dpdk_device_class;
 extern vlib_node_registration_t dpdk_input_node;
 extern vlib_node_registration_t admin_up_down_process_node;
@@ -364,7 +362,7 @@ typedef struct
   u8 nchannels_set_manually;
   u32 coremask;
   u32 nchannels;
-  u32 num_mbufs;
+  u32 num_crypto_mbufs;
 
   /*
    * format interface names ala xxxEthernet%d/%d/%d instead of
@@ -443,9 +441,6 @@ typedef struct
   vnet_main_t *vnet_main;
   dpdk_config_main_t *conf;
 
-  /* mempool */
-  struct rte_mempool **pktmbuf_pools;
-
   /* API message ID base */
   u16 msg_id_base;
 
diff --git a/src/plugins/dpdk/device/dpdk_priv.h b/src/plugins/dpdk/device/dpdk_priv.h
index 1956cca19c3..e0e06826935 100644
--- a/src/plugins/dpdk/device/dpdk_priv.h
+++ b/src/plugins/dpdk/device/dpdk_priv.h
@@ -49,14 +49,6 @@ _(file-prefix)                                  \
 _(vdev)                                         \
 _(log-level)
 
-typedef struct
-{
-  /* must be first */
-  struct rte_pktmbuf_pool_private mbp_priv;
-  u8 buffer_pool_index;
-} dpdk_mempool_private_t;
-
-
 static inline void
 dpdk_get_xstats (dpdk_device_t * xd)
 {
diff --git a/src/plugins/dpdk/device/init.c b/src/plugins/dpdk/device/init.c
index dcff0e52293..fa3b691fca4 100644
--- a/src/plugins/dpdk/device/init.c
+++ b/src/plugins/dpdk/device/init.c
@@ -151,60 +151,6 @@ dpdk_device_lock_init (dpdk_device_t * xd)
     }
 }
 
-static struct rte_mempool_ops *
-get_ops_by_name (char *ops_name)
-{
-  u32 i;
-
-  for (i = 0; i < rte_mempool_ops_table.num_ops; i++)
-    {
-      if (!strcmp (ops_name, rte_mempool_ops_table.ops[i].name))
-	return &rte_mempool_ops_table.ops[i];
-    }
-
-  return 0;
-}
-
-static int
-dpdk_ring_alloc (struct rte_mempool *mp)
-{
-  u32 rg_flags = 0, count;
-  i32 ret;
-  char rg_name[RTE_RING_NAMESIZE];
-  struct rte_ring *r;
-
-  ret = snprintf (rg_name, sizeof (rg_name), RTE_MEMPOOL_MZ_FORMAT, mp->name);
-  if (ret < 0 || ret >= (i32) sizeof (rg_name))
-    return -ENAMETOOLONG;
-
-  /* ring flags */
-  if (mp->flags & MEMPOOL_F_SP_PUT)
-    rg_flags |= RING_F_SP_ENQ;
-  if (mp->flags & MEMPOOL_F_SC_GET)
-    rg_flags |= RING_F_SC_DEQ;
-
-  count = rte_align32pow2 (mp->size + 1);
-  /*
-   * Allocate the ring that will be used to store objects.
-   * Ring functions will return appropriate errors if we are
-   * running as a secondary process etc., so no checks made
-   * in this function for that condition.
-   */
-  /* XXX can we get memory from the right socket? */
-  r = clib_mem_alloc_aligned (rte_ring_get_memsize (count),
-			      CLIB_CACHE_LINE_BYTES);
-
-  /* XXX rte_ring_lookup will not work */
-
-  ret = rte_ring_init (r, rg_name, count, rg_flags);
-  if (ret)
-    return ret;
-
-  mp->pool_data = r;
-
-  return 0;
-}
-
 static int
 dpdk_port_crc_strip_enabled (dpdk_device_t * xd)
 {
@@ -220,7 +166,6 @@ dpdk_lib_init (dpdk_main_t * dm)
 {
   u32 nports;
   u32 mtu, max_rx_frame;
-  u32 nb_desc = 0;
   int i;
   clib_error_t *error;
   vlib_main_t *vm = vlib_get_main ();
@@ -631,9 +576,6 @@ dpdk_lib_init (dpdk_main_t * dm)
 	  dq->queue_id = 0;
 	}
 
-      /* count the number of descriptors used for this device */
-      nb_desc += xd->nb_rx_desc + xd->nb_tx_desc * xd->tx_q_used;
-
       error = ethernet_register_interface
 	(dm->vnet_main, dpdk_device_class.index, xd->device_index,
 	 /* ethernet address */ addr,
@@ -811,10 +753,6 @@ dpdk_lib_init (dpdk_main_t * dm)
     }
   /* *INDENT-ON* */
 
-  if (nb_desc > dm->conf->num_mbufs)
-    dpdk_log_err ("%d mbufs allocated but total rx/tx ring size is %d\n",
-		  dm->conf->num_mbufs, nb_desc);
-
   return 0;
 }
 
@@ -1209,7 +1147,8 @@ dpdk_config (vlib_main_t * vm, unformat_input_t * input)
 	}
       else if (unformat (input, "num-mem-channels %d", &conf->nchannels))
 	conf->nchannels_set_manually = 0;
-      else if (unformat (input, "num-mbufs %d", &conf->num_mbufs))
+      else if (unformat (input, "num-crypto-mbufs %d",
+			 &conf->num_crypto_mbufs))
 	;
       else if (unformat (input, "uio-driver %s", &conf->uio_driver_name))
 	;
@@ -1452,36 +1391,10 @@ dpdk_config (vlib_main_t * vm, unformat_input_t * input)
   if (ret < 0)
     return clib_error_return (0, "rte_eal_init returned %d", ret);
 
-  /* set custom ring memory allocator */
-  {
-    struct rte_mempool_ops *ops = NULL;
-
-    ops = get_ops_by_name ("ring_sp_sc");
-    ops->alloc = dpdk_ring_alloc;
-
-    ops = get_ops_by_name ("ring_mp_sc");
-    ops->alloc = dpdk_ring_alloc;
-
-    ops = get_ops_by_name ("ring_sp_mc");
-    ops->alloc = dpdk_ring_alloc;
-
-    ops = get_ops_by_name ("ring_mp_mc");
-    ops->alloc = dpdk_ring_alloc;
-  }
-
   /* main thread 1st */
-  error = dpdk_buffer_pool_create (vm, conf->num_mbufs, rte_socket_id ());
-  if (error)
+  if ((error = dpdk_buffer_pools_create (vm)))
     return error;
 
-  for (i = 0; i < RTE_MAX_LCORE; i++)
-    {
-      error = dpdk_buffer_pool_create (vm, conf->num_mbufs,
-				       rte_lcore_to_socket_id (i));
-      if (error)
-	return error;
-    }
-
 done:
   return error;
 }
@@ -1768,7 +1681,6 @@ dpdk_init (vlib_main_t * vm)
   dm->conf = &dpdk_config_main;
 
   dm->conf->nchannels = 4;
-  dm->conf->num_mbufs = dm->conf->num_mbufs ? dm->conf->num_mbufs : NB_MBUF;
   vec_add1 (dm->conf->eal_init_args, (u8 *) "vnet");
   vec_add1 (dm->conf->eal_init_args, (u8 *) "--in-memory");