dpdk: dpdk-input optimizations and fixes

- fix issue caused by assumption that ethertype starts at 0
- intoduce buffer templates to speed-up vlib_buffer_t metadata
initialization
- avoid check for multiseg buffers inside loop if multiseg is disabled
- interleave prefetches to reduce load on L1 cache

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

diff --git a/src/plugins/dpdk.am b/src/plugins/dpdk.am
index 01383de..f1a37ae 100644 (file)
--- a/src/plugins/dpdk.am
+++ b/src/plugins/dpdk.am
@@ -16,6 +16,10 @@ vppplugins_LTLIBRARIES += dpdk_plugin.la
 
 dpdk_plugin_la_LDFLAGS = $(AM_LDFLAGS) -Wl,--whole-archive,-l:libdpdk.a,--no-whole-archive,-lm,-ldl
 
+# due to internal compiler error in GCC when compiling dpdk/device/node.c
+# debug sysmbols level reduced to 1. See GCC PR #79953 for details
+CFLAGS += -g1
+
 dpdk_plugin_la_SOURCES =                                       \
   dpdk/main.c                                                  \
   dpdk/buffer.c                                                        \

diff --git a/src/plugins/dpdk/device/dpdk.h b/src/plugins/dpdk/device/dpdk.h
index 092c7dc..84f86ae 100644 (file)
--- a/src/plugins/dpdk/device/dpdk.h
+++ b/src/plugins/dpdk/device/dpdk.h
@@ -338,6 +338,9 @@ typedef struct
   /* per-thread recycle lists */
   u32 **recycle;
 
+  /* per-thread buffer templates */
+  vlib_buffer_t *buffer_templates;
+
   /* buffer flags template, configurable to enable/disable tcp / udp cksum */
   u32 buffer_flags_template;
 

diff --git a/src/plugins/dpdk/device/init.c b/src/plugins/dpdk/device/init.c
index 85ecde2..110d745 100755 (executable)
--- a/src/plugins/dpdk/device/init.c
+++ b/src/plugins/dpdk/device/init.c
@@ -572,6 +572,21 @@ dpdk_lib_init (dpdk_main_t * dm)
     dm->buffer_flags_template &= ~(IP_BUFFER_L4_CHECKSUM_CORRECT
                                   | IP_BUFFER_L4_CHECKSUM_COMPUTED);
 
+  /* vlib_buffer_t template */
+  vec_validate_aligned (dm->buffer_templates, tm->n_vlib_mains - 1,
+                       CLIB_CACHE_LINE_BYTES);
+  for (i = 0; i < tm->n_vlib_mains; i++)
+    {
+      vlib_buffer_free_list_t *fl;
+      vlib_buffer_t *bt = vec_elt_at_index (dm->buffer_templates, i);
+      fl = vlib_buffer_get_free_list (vm,
+                                     VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
+      vlib_buffer_init_for_free_list (bt, fl);
+      bt->flags = dm->buffer_flags_template;
+      bt->current_data = -RTE_PKTMBUF_HEADROOM;
+      vnet_buffer (bt)->sw_if_index[VLIB_TX] = (u32) ~ 0;
+    }
+
   for (i = 0; i < nports; i++)
     {
       u8 addr[6];

diff --git a/src/plugins/dpdk/device/node.c b/src/plugins/dpdk/device/node.c
index 2120069..e8d502c 100644 (file)
--- a/src/plugins/dpdk/device/node.c
+++ b/src/plugins/dpdk/device/node.c
@@ -37,21 +37,21 @@ static char *dpdk_error_strings[] = {
 always_inline int
 vlib_buffer_is_ip4 (vlib_buffer_t * b)
 {
-  ethernet_header_t *h = (ethernet_header_t *) b->data;
+  ethernet_header_t *h = (ethernet_header_t *) vlib_buffer_get_current (b);
   return (h->type == clib_host_to_net_u16 (ETHERNET_TYPE_IP4));
 }
 
 always_inline int
 vlib_buffer_is_ip6 (vlib_buffer_t * b)
 {
-  ethernet_header_t *h = (ethernet_header_t *) b->data;
+  ethernet_header_t *h = (ethernet_header_t *) vlib_buffer_get_current (b);
   return (h->type == clib_host_to_net_u16 (ETHERNET_TYPE_IP6));
 }
 
 always_inline int
 vlib_buffer_is_mpls (vlib_buffer_t * b)
 {
-  ethernet_header_t *h = (ethernet_header_t *) b->data;
+  ethernet_header_t *h = (ethernet_header_t *) vlib_buffer_get_current (b);
   return (h->type == clib_host_to_net_u16 (ETHERNET_TYPE_MPLS_UNICAST));
 }
 
@@ -217,10 +217,35 @@ dpdk_prefetch_buffer (struct rte_mbuf *mb)
 }
 
 static_always_inline void
-dpdk_prefetch_buffer_data (struct rte_mbuf *mb)
+dpdk_prefetch_ethertype (struct rte_mbuf *mb)
 {
-  vlib_buffer_t *b = vlib_buffer_from_rte_mbuf (mb);
-  CLIB_PREFETCH (b->data, CLIB_CACHE_LINE_BYTES, STORE);
+  CLIB_PREFETCH (mb->buf_addr + mb->data_off +
+                STRUCT_OFFSET_OF (ethernet_header_t, type),
+                CLIB_CACHE_LINE_BYTES, LOAD);
+}
+
+
+/*
+   This function should fill 1st cacheline of vlib_buffer_t metadata with data
+   from buffer template. Instead of filling field by field, we construct
+   template and then use 128/256 bit vector instruction to copy data.
+   This code first loads whole cacheline into 4 128-bit registers (xmm)
+   or two 256 bit registers (ymm) and then stores data into all 4 buffers
+   efectively saving on register load operations.
+*/
+
+static_always_inline void
+dpdk_buffer_init_from_template (void *d0, void *d1, void *d2, void *d3,
+                               void *s)
+{
+  int i;
+  for (i = 0; i < 2; i++)
+    {
+      *(u8x32 *) (((u8 *) d0) + i * 32) =
+       *(u8x32 *) (((u8 *) d1) + i * 32) =
+       *(u8x32 *) (((u8 *) d2) + i * 32) =
+       *(u8x32 *) (((u8 *) d3) + i * 32) = *(u8x32 *) (((u8 *) s) + i * 32);
+    }
 }
 
 /*
@@ -229,7 +254,8 @@ dpdk_prefetch_buffer_data (struct rte_mbuf *mb)
  */
 static_always_inline u32
 dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
-                  vlib_node_runtime_t * node, u32 cpu_index, u16 queue_id)
+                  vlib_node_runtime_t * node, u32 cpu_index, u16 queue_id,
+                  int maybe_multiseg)
 {
   u32 n_buffers;
   u32 next_index = VNET_DEVICE_INPUT_NEXT_ETHERNET_INPUT;
@@ -239,7 +265,7 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
   uword n_rx_bytes = 0;
   u32 n_trace, trace_cnt __attribute__ ((unused));
   vlib_buffer_free_list_t *fl;
-  u32 buffer_flags_template;
+  vlib_buffer_t *bt = vec_elt_at_index (dm->buffer_templates, cpu_index);
 
   if ((xd->flags & DPDK_DEVICE_FLAG_ADMIN_UP) == 0)
     return 0;
@@ -251,8 +277,6 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
       return 0;
     }
 
-  buffer_flags_template = dm->buffer_flags_template;
-
   vec_reset_length (xd->d_trace_buffers[cpu_index]);
   trace_cnt = n_trace = vlib_get_trace_count (vm, node);
 
@@ -272,33 +296,44 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
 
   fl = vlib_buffer_get_free_list (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
 
+  /* Update buffer template */
+  vnet_buffer (bt)->sw_if_index[VLIB_RX] = xd->vlib_sw_if_index;
+  bt->error = node->errors[DPDK_ERROR_NONE];
+
   mb_index = 0;
 
   while (n_buffers > 0)
     {
       vlib_buffer_t *b0, *b1, *b2, *b3;
-      u32 bi0, next0, l3_offset0;
-      u32 bi1, next1, l3_offset1;
-      u32 bi2, next2, l3_offset2;
-      u32 bi3, next3, l3_offset3;
+      u32 bi0, next0;
+      u32 bi1, next1;
+      u32 bi2, next2;
+      u32 bi3, next3;
       u8 error0, error1, error2, error3;
       u64 or_ol_flags;
 
       vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
 
-      while (n_buffers > 8 && n_left_to_next > 4)
+      while (n_buffers >= 12 && n_left_to_next >= 4)
        {
-         struct rte_mbuf *mb0 = xd->rx_vectors[queue_id][mb_index];
-         struct rte_mbuf *mb1 = xd->rx_vectors[queue_id][mb_index + 1];
-         struct rte_mbuf *mb2 = xd->rx_vectors[queue_id][mb_index + 2];
-         struct rte_mbuf *mb3 = xd->rx_vectors[queue_id][mb_index + 3];
+         struct rte_mbuf *mb0, *mb1, *mb2, *mb3;
 
-         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 4]);
-         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 5]);
-         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 6]);
-         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 7]);
+         /* prefetches are interleaved with the rest of the code to reduce
+            pressure on L1 cache */
+         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 8]);
+         dpdk_prefetch_ethertype (xd->rx_vectors[queue_id][mb_index + 4]);
 
-         if (xd->flags & DPDK_DEVICE_FLAG_MAYBE_MULTISEG)
+         mb0 = xd->rx_vectors[queue_id][mb_index];
+         mb1 = xd->rx_vectors[queue_id][mb_index + 1];
+         mb2 = xd->rx_vectors[queue_id][mb_index + 2];
+         mb3 = xd->rx_vectors[queue_id][mb_index + 3];
+
+         ASSERT (mb0);
+         ASSERT (mb1);
+         ASSERT (mb2);
+         ASSERT (mb3);
+
+         if (maybe_multiseg)
            {
              if (PREDICT_FALSE (mb0->nb_segs > 1))
                dpdk_prefetch_buffer (mb0->next);
@@ -310,22 +345,29 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
                dpdk_prefetch_buffer (mb3->next);
            }
 
-         ASSERT (mb0);
-         ASSERT (mb1);
-         ASSERT (mb2);
-         ASSERT (mb3);
-
-         or_ol_flags = (mb0->ol_flags | mb1->ol_flags |
-                        mb2->ol_flags | mb3->ol_flags);
          b0 = vlib_buffer_from_rte_mbuf (mb0);
          b1 = vlib_buffer_from_rte_mbuf (mb1);
          b2 = vlib_buffer_from_rte_mbuf (mb2);
          b3 = vlib_buffer_from_rte_mbuf (mb3);
 
-         vlib_buffer_init_for_free_list (b0, fl);
-         vlib_buffer_init_for_free_list (b1, fl);
-         vlib_buffer_init_for_free_list (b2, fl);
-         vlib_buffer_init_for_free_list (b3, fl);
+         dpdk_buffer_init_from_template (b0, b1, b2, b3, bt);
+
+         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 9]);
+         dpdk_prefetch_ethertype (xd->rx_vectors[queue_id][mb_index + 5]);
+
+         /* current_data must be set to -RTE_PKTMBUF_HEADROOM in template */
+         b0->current_data += mb0->data_off;
+         b1->current_data += mb1->data_off;
+         b2->current_data += mb2->data_off;
+         b3->current_data += mb3->data_off;
+
+         b0->current_length = mb0->data_len;
+         b1->current_length = mb1->data_len;
+         b2->current_length = mb2->data_len;
+         b3->current_length = mb3->data_len;
+
+         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 10]);
+         dpdk_prefetch_ethertype (xd->rx_vectors[queue_id][mb_index + 7]);
 
          bi0 = vlib_get_buffer_index (vm, b0);
          bi1 = vlib_get_buffer_index (vm, b1);
@@ -345,21 +387,17 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
            }
          else
            {
-             /* prefetch packet data for faster access to the ethertype */
-             dpdk_prefetch_buffer_data (xd->rx_vectors[queue_id]
-                                        [mb_index + 4]);
-             dpdk_prefetch_buffer_data (xd->rx_vectors[queue_id]
-                                        [mb_index + 5]);
-             dpdk_prefetch_buffer_data (xd->rx_vectors[queue_id]
-                                        [mb_index + 6]);
-             dpdk_prefetch_buffer_data (xd->rx_vectors[queue_id]
-                                        [mb_index + 7]);
              next0 = dpdk_rx_next_from_etype (mb0, b0);
              next1 = dpdk_rx_next_from_etype (mb1, b1);
              next2 = dpdk_rx_next_from_etype (mb2, b2);
              next3 = dpdk_rx_next_from_etype (mb3, b3);
            }
 
+         dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 11]);
+         dpdk_prefetch_ethertype (xd->rx_vectors[queue_id][mb_index + 6]);
+
+         or_ol_flags = (mb0->ol_flags | mb1->ol_flags |
+                        mb2->ol_flags | mb3->ol_flags);
          if (PREDICT_FALSE (or_ol_flags & PKT_RX_IP_CKSUM_BAD))
            {
              dpdk_rx_error_from_mb (mb0, &next0, &error0);
@@ -371,46 +409,11 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
              b2->error = node->errors[error2];
              b3->error = node->errors[error3];
            }
-         else
-           {
-             b0->error = b1->error = node->errors[DPDK_ERROR_NONE];
-             b2->error = b3->error = node->errors[DPDK_ERROR_NONE];
-           }
 
-         l3_offset0 = device_input_next_node_advance[next0];
-         l3_offset1 = device_input_next_node_advance[next1];
-         l3_offset2 = device_input_next_node_advance[next2];
-         l3_offset3 = device_input_next_node_advance[next3];
-
-         b0->current_data = l3_offset0 + mb0->data_off;
-         b1->current_data = l3_offset1 + mb1->data_off;
-         b2->current_data = l3_offset2 + mb2->data_off;
-         b3->current_data = l3_offset3 + mb3->data_off;
-
-         b0->current_data -= RTE_PKTMBUF_HEADROOM;
-         b1->current_data -= RTE_PKTMBUF_HEADROOM;
-         b2->current_data -= RTE_PKTMBUF_HEADROOM;
-         b3->current_data -= RTE_PKTMBUF_HEADROOM;
-
-         b0->current_length = mb0->data_len - l3_offset0;
-         b1->current_length = mb1->data_len - l3_offset1;
-         b2->current_length = mb2->data_len - l3_offset2;
-         b3->current_length = mb3->data_len - l3_offset3;
-
-         b0->flags = buffer_flags_template;
-         b1->flags = buffer_flags_template;
-         b2->flags = buffer_flags_template;
-         b3->flags = buffer_flags_template;
-
-         vnet_buffer (b0)->sw_if_index[VLIB_RX] = xd->vlib_sw_if_index;
-         vnet_buffer (b1)->sw_if_index[VLIB_RX] = xd->vlib_sw_if_index;
-         vnet_buffer (b2)->sw_if_index[VLIB_RX] = xd->vlib_sw_if_index;
-         vnet_buffer (b3)->sw_if_index[VLIB_RX] = xd->vlib_sw_if_index;
-
-         vnet_buffer (b0)->sw_if_index[VLIB_TX] = (u32) ~ 0;
-         vnet_buffer (b1)->sw_if_index[VLIB_TX] = (u32) ~ 0;
-         vnet_buffer (b2)->sw_if_index[VLIB_TX] = (u32) ~ 0;
-         vnet_buffer (b3)->sw_if_index[VLIB_TX] = (u32) ~ 0;
+         vlib_buffer_advance (b0, device_input_next_node_advance[next0]);
+         vlib_buffer_advance (b1, device_input_next_node_advance[next1]);
+         vlib_buffer_advance (b2, device_input_next_node_advance[next2]);
+         vlib_buffer_advance (b3, device_input_next_node_advance[next3]);
 
          n_rx_bytes += mb0->pkt_len;
          n_rx_bytes += mb1->pkt_len;
@@ -418,7 +421,7 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
          n_rx_bytes += mb3->pkt_len;
 
          /* Process subsequent segments of multi-segment packets */
-         if (xd->flags & DPDK_DEVICE_FLAG_MAYBE_MULTISEG)
+         if (maybe_multiseg)
            {
              dpdk_process_subseq_segs (vm, b0, mb0, fl);
              dpdk_process_subseq_segs (vm, b1, mb1, fl);
@@ -452,6 +455,13 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
        {
          struct rte_mbuf *mb0 = xd->rx_vectors[queue_id][mb_index];
 
+         if (PREDICT_TRUE (n_buffers > 3))
+           {
+             dpdk_prefetch_buffer (xd->rx_vectors[queue_id][mb_index + 2]);
+             dpdk_prefetch_ethertype (xd->rx_vectors[queue_id]
+                                      [mb_index + 1]);
+           }
+
          ASSERT (mb0);
 
          b0 = vlib_buffer_from_rte_mbuf (mb0);
@@ -460,7 +470,11 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
          if (PREDICT_FALSE (mb0->nb_segs > 1))
            dpdk_prefetch_buffer (mb0->next);
 
-         vlib_buffer_init_for_free_list (b0, fl);
+         clib_memcpy (b0, bt, CLIB_CACHE_LINE_BYTES);
+
+         ASSERT (b0->current_data == -RTE_PKTMBUF_HEADROOM);
+         b0->current_data += mb0->data_off;
+         b0->current_length = mb0->data_len;
 
          bi0 = vlib_get_buffer_index (vm, b0);
 
@@ -474,18 +488,9 @@ dpdk_device_input (dpdk_main_t * dm, dpdk_device_t * xd,
            next0 = dpdk_rx_next_from_etype (mb0, b0);
 
          dpdk_rx_error_from_mb (mb0, &next0, &error0);
-         b0->error = node->errors[error0];
-
-         l3_offset0 = device_input_next_node_advance[next0];
 
-         b0->current_data = l3_offset0;
-         b0->current_data += mb0->data_off - RTE_PKTMBUF_HEADROOM;
-         b0->current_length = mb0->data_len - l3_offset0;
+         vlib_buffer_advance (b0, device_input_next_node_advance[next0]);
 
-         b0->flags = buffer_flags_template;
-
-         vnet_buffer (b0)->sw_if_index[VLIB_RX] = xd->vlib_sw_if_index;
-         vnet_buffer (b0)->sw_if_index[VLIB_TX] = (u32) ~ 0;
          n_rx_bytes += mb0->pkt_len;
 
          /* Process subsequent segments of multi-segment packets */
@@ -604,7 +609,10 @@ dpdk_input (vlib_main_t * vm, vlib_node_runtime_t * node, vlib_frame_t * f)
   vec_foreach (dq, dm->devices_by_cpu[cpu_index])
     {
       xd = vec_elt_at_index(dm->devices, dq->device);
-      n_rx_packets += dpdk_device_input (dm, xd, node, cpu_index, dq->queue_id);
+      if (xd->flags & DPDK_DEVICE_FLAG_MAYBE_MULTISEG)
+        n_rx_packets += dpdk_device_input (dm, xd, node, cpu_index, dq->queue_id, /* maybe_multiseg */ 1);
+      else
+        n_rx_packets += dpdk_device_input (dm, xd, node, cpu_index, dq->queue_id, /* maybe_multiseg */ 0);
     }
   /* *INDENT-ON* */