threads: add support for multiple worker handoff queues

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

3 files changed, 16 insertions, 123 deletions

diff --git a/vnet/vnet/devices/dpdk/cli.c b/vnet/vnet/devices/dpdk/cli.c
index 8f99b272ad8..5e53a98beae 100644
--- a/vnet/vnet/devices/dpdk/cli.c
+++ b/vnet/vnet/devices/dpdk/cli.c
@@ -346,9 +346,7 @@ show_efd (vlib_main_t * vm,
   else if (unformat (input, "worker"))
     {
       vlib_thread_main_t *tm = vlib_get_thread_main ();
-      vlib_frame_queue_t *fq;
       vlib_thread_registration_t *tr;
-      int thread_id;
       u32 num_workers = 0;
       u32 first_worker_index = 0;
       uword *p;
@@ -364,27 +362,9 @@ show_efd (vlib_main_t * vm,
 
       vlib_cli_output (vm,
 		       "num_workers               %d\n"
-		       "first_worker_index        %d\n"
-		       "vlib_frame_queues[%d]:\n",
-		       num_workers, first_worker_index, tm->n_vlib_mains);
+		       "first_worker_index        %d\n",
+		       num_workers, first_worker_index);
 
-      for (thread_id = 0; thread_id < tm->n_vlib_mains; thread_id++)
-	{
-	  fq = vlib_frame_queues[thread_id];
-	  if (fq)
-	    {
-	      vlib_cli_output (vm,
-			       "%2d: frames_queued         %u\n"
-			       "    frames_queued_hint    %u\n"
-			       "    enqueue_full_events   %u\n"
-			       "    enqueue_efd_discards  %u\n",
-			       thread_id,
-			       (fq->tail - fq->head),
-			       (fq->tail - fq->head_hint),
-			       fq->enqueue_full_events,
-			       fq->enqueue_efd_discards);
-	    }
-	}
     }
   else if (unformat (input, "help"))
     {
@@ -413,9 +393,6 @@ clear_efd (vlib_main_t * vm,
 {
   dpdk_main_t *dm = &dpdk_main;
   dpdk_device_t *xd;
-  vlib_thread_main_t *tm = vlib_get_thread_main ();
-  vlib_frame_queue_t *fq;
-  int thread_id;
 
     /* *INDENT-OFF* */
     vec_foreach (xd, dm->devices)
@@ -432,16 +409,6 @@ clear_efd (vlib_main_t * vm,
       }
     /* *INDENT-ON* */
 
-  for (thread_id = 0; thread_id < tm->n_vlib_mains; thread_id++)
-    {
-      fq = vlib_frame_queues[thread_id];
-      if (fq)
-	{
-	  fq->enqueue_full_events = 0;
-	  fq->enqueue_efd_discards = 0;
-	}
-    }
-
   return 0;
 }
 
diff --git a/vnet/vnet/handoff.c b/vnet/vnet/handoff.c
index 5593aa74942..22d2ea98df7 100644
--- a/vnet/vnet/handoff.c
+++ b/vnet/vnet/handoff.c
@@ -34,12 +34,16 @@ typedef struct
 
   per_inteface_handoff_data_t *if_data;
 
+  /* Worker handoff index */
+  u32 frame_queue_index;
+
   /* convenience variables */
   vlib_main_t *vlib_main;
   vnet_main_t *vnet_main;
 } handoff_main_t;
 
 handoff_main_t handoff_main;
+vlib_node_registration_t handoff_dispatch_node;
 
 typedef struct
 {
@@ -147,8 +151,9 @@ worker_handoff_node_fn (vlib_main_t * vm,
 	  if (hf)
 	    hf->n_vectors = VLIB_FRAME_SIZE - n_left_to_next_worker;
 
-	  hf = dpdk_get_handoff_queue_elt (next_worker_index,
-					   handoff_queue_elt_by_worker_index);
+	  hf = vlib_get_worker_handoff_queue_elt (hm->frame_queue_index,
+						  next_worker_index,
+						  handoff_queue_elt_by_worker_index);
 
 	  n_left_to_next_worker = VLIB_FRAME_SIZE - hf->n_vectors;
 	  to_next_worker = &hf->buffer_index[hf->n_vectors];
@@ -163,7 +168,7 @@ worker_handoff_node_fn (vlib_main_t * vm,
       if (n_left_to_next_worker == 0)
 	{
 	  hf->n_vectors = VLIB_FRAME_SIZE;
-	  vlib_put_handoff_queue_elt (hf);
+	  vlib_put_frame_queue_elt (hf);
 	  current_worker_index = ~0;
 	  handoff_queue_elt_by_worker_index[next_worker_index] = 0;
 	  hf = 0;
@@ -196,7 +201,7 @@ worker_handoff_node_fn (vlib_main_t * vm,
 	   */
 	  if (1 || hf->n_vectors == hf->last_n_vectors)
 	    {
-	      vlib_put_handoff_queue_elt (hf);
+	      vlib_put_frame_queue_elt (hf);
 	      handoff_queue_elt_by_worker_index[i] = 0;
 	    }
 	  else
@@ -247,6 +252,10 @@ interface_handoff_enable_disable (vlib_main_t * vm, u32 sw_if_index,
   if (clib_bitmap_last_set (bitmap) >= hm->num_workers)
     return VNET_API_ERROR_INVALID_WORKER;
 
+  if (hm->frame_queue_index == ~0)
+    hm->frame_queue_index =
+      vlib_frame_queue_main_init (handoff_dispatch_node.index, 0);
+
   vec_validate (hm->if_data, sw_if_index);
   d = vec_elt_at_index (hm->if_data, sw_if_index);
 
@@ -356,9 +365,6 @@ format_handoff_dispatch_trace (u8 * s, va_list * args)
   return s;
 }
 
-
-vlib_node_registration_t handoff_dispatch_node;
-
 #define foreach_handoff_dispatch_error \
 _(EXAMPLE, "example packets")
 
@@ -556,8 +562,7 @@ handoff_init (vlib_main_t * vm)
   hm->vlib_main = vm;
   hm->vnet_main = &vnet_main;
 
-  ASSERT (tm->handoff_dispatch_node_index == ~0);
-  tm->handoff_dispatch_node_index = handoff_dispatch_node.index;
+  hm->frame_queue_index = ~0;
 
   return 0;
 }
diff --git a/vnet/vnet/handoff.h b/vnet/vnet/handoff.h
index 9320f5602b5..4fefb369475 100644
--- a/vnet/vnet/handoff.h
+++ b/vnet/vnet/handoff.h
@@ -32,85 +32,6 @@ typedef enum
   HANDOFF_DISPATCH_N_NEXT,
 } handoff_dispatch_next_t;
 
-static inline void
-vlib_put_handoff_queue_elt (vlib_frame_queue_elt_t * hf)
-{
-  CLIB_MEMORY_BARRIER ();
-  hf->valid = 1;
-}
-
-static inline vlib_frame_queue_elt_t *
-vlib_get_handoff_queue_elt (u32 vlib_worker_index)
-{
-  vlib_frame_queue_t *fq;
-  vlib_frame_queue_elt_t *elt;
-  u64 new_tail;
-
-  fq = vlib_frame_queues[vlib_worker_index];
-  ASSERT (fq);
-
-  new_tail = __sync_add_and_fetch (&fq->tail, 1);
-
-  /* Wait until a ring slot is available */
-  while (new_tail >= fq->head_hint + fq->nelts)
-    vlib_worker_thread_barrier_check ();
-
-  elt = fq->elts + (new_tail & (fq->nelts - 1));
-
-  /* this would be very bad... */
-  while (elt->valid)
-    ;
-
-  elt->msg_type = VLIB_FRAME_QUEUE_ELT_DISPATCH_FRAME;
-  elt->last_n_vectors = elt->n_vectors = 0;
-
-  return elt;
-}
-
-static inline vlib_frame_queue_t *
-is_vlib_handoff_queue_congested (u32 vlib_worker_index,
-				 u32 queue_hi_thresh,
-				 vlib_frame_queue_t **
-				 handoff_queue_by_worker_index)
-{
-  vlib_frame_queue_t *fq;
-
-  fq = handoff_queue_by_worker_index[vlib_worker_index];
-  if (fq != (vlib_frame_queue_t *) (~0))
-    return fq;
-
-  fq = vlib_frame_queues[vlib_worker_index];
-  ASSERT (fq);
-
-  if (PREDICT_FALSE (fq->tail >= (fq->head_hint + queue_hi_thresh)))
-    {
-      /* a valid entry in the array will indicate the queue has reached
-       * the specified threshold and is congested
-       */
-      handoff_queue_by_worker_index[vlib_worker_index] = fq;
-      fq->enqueue_full_events++;
-      return fq;
-    }
-
-  return NULL;
-}
-
-static inline vlib_frame_queue_elt_t *
-dpdk_get_handoff_queue_elt (u32 vlib_worker_index,
-			    vlib_frame_queue_elt_t **
-			    handoff_queue_elt_by_worker_index)
-{
-  vlib_frame_queue_elt_t *elt;
-
-  if (handoff_queue_elt_by_worker_index[vlib_worker_index])
-    return handoff_queue_elt_by_worker_index[vlib_worker_index];
-
-  elt = vlib_get_handoff_queue_elt (vlib_worker_index);
-
-  handoff_queue_elt_by_worker_index[vlib_worker_index] = elt;
-
-  return elt;
-}
 
 static inline u64
 ipv4_get_key (ip4_header_t * ip)