hsa: fix vpp_echo mq locking

Type: fix

Change-Id: I18d2cde0baaed4134e8378c09aaa88693fb997f8
Signed-off-by: Nathan Skrzypczak <nathan.skrzypczak@gmail.com>

3 files changed, 62 insertions, 18 deletions

diff --git a/src/plugins/hs_apps/sapi/vpp_echo.c b/src/plugins/hs_apps/sapi/vpp_echo.c
index 79b71d1611e..407ab3f3816 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo.c
+++ b/src/plugins/hs_apps/sapi/vpp_echo.c
@@ -399,7 +399,7 @@ echo_handle_data (echo_main_t * em, echo_session_t * s, u8 * rx_buf)
 static void *
 echo_data_thread_fn (void *arg)
 {
-  clib_mem_set_thread_index ();
+  clib_mem_set_thread_index ();	/* First thing to do in client thread */
   echo_main_t *em = &echo_main;
   u32 N = em->n_clients;
   u32 n = (N + em->n_rx_threads - 1) / em->n_rx_threads;
@@ -661,44 +661,83 @@ echo_process_rpcs (echo_main_t * em)
 {
   echo_rpc_msg_t *rpc;
   svm_msg_q_msg_t msg;
+  svm_msg_q_t *mq = em->rpc_msq_queue;
+
   while (em->state < STATE_DATA_DONE && !em->time_to_stop)
     {
-      if (svm_msg_q_sub (em->rpc_msq_queue, &msg, SVM_Q_TIMEDWAIT, 1))
-	continue;
-      rpc = svm_msg_q_msg_data (em->rpc_msq_queue, &msg);
+      svm_msg_q_lock (mq);
+      if (svm_msg_q_is_empty (mq) && svm_msg_q_timedwait (mq, 1))
+	{
+	  svm_msg_q_unlock (mq);
+	  continue;
+	}
+      svm_msg_q_sub_w_lock (mq, &msg);
+      rpc = svm_msg_q_msg_data (mq, &msg);
+      svm_msg_q_unlock (mq);
       ((echo_rpc_t) rpc->fp) (rpc->arg, rpc->opaque);
-      svm_msg_q_free_msg (em->rpc_msq_queue, &msg);
+      svm_msg_q_free_msg (mq, &msg);
+    }
+}
+
+static inline int
+echo_mq_dequeue_batch (svm_msg_q_t * mq, svm_msg_q_msg_t * msg_vec,
+		       u32 n_max_msg)
+{
+  svm_msg_q_msg_t *msg;
+  u32 n_msgs;
+  int i;
+
+  n_msgs = clib_min (svm_msg_q_size (mq), n_max_msg);
+  for (i = 0; i < n_msgs; i++)
+    {
+      vec_add2 (msg_vec, msg, 1);
+      svm_msg_q_sub_w_lock (mq, msg);
     }
+  return n_msgs;
 }
 
 static void *
 echo_mq_thread_fn (void *arg)
 {
-  clib_mem_set_thread_index ();
+  clib_mem_set_thread_index ();	/* First thing to do in client thread */
+  svm_msg_q_msg_t *msg_vec = 0;
   echo_main_t *em = &echo_main;
   session_event_t *e;
-  svm_msg_q_msg_t msg;
-  int rv;
+  svm_msg_q_msg_t *msg;
+  svm_msg_q_t *mq;
+  int i;
+
+  vec_validate (msg_vec, em->evt_q_size);
+  vec_reset_length (msg_vec);
   wait_for_state_change (em, STATE_ATTACHED, 0);
-  if (em->state < STATE_ATTACHED || !em->our_event_queue)
+  mq = em->our_event_queue;
+  if (em->state < STATE_ATTACHED || !mq)
     {
       ECHO_FAIL (ECHO_FAIL_APP_ATTACH, "Application failed to attach");
       pthread_exit (0);
     }
 
-  while (1)
+  while (em->state < STATE_DETACHED && !em->time_to_stop)
     {
-      if (!(rv = svm_msg_q_sub (em->our_event_queue,
-				&msg, SVM_Q_TIMEDWAIT, 1)))
+      svm_msg_q_lock (mq);
+      if (svm_msg_q_is_empty (mq) && svm_msg_q_timedwait (mq, 1))
 	{
-	  e = svm_msg_q_msg_data (em->our_event_queue, &msg);
+	  svm_msg_q_unlock (mq);
+	  continue;
+	}
+      echo_mq_dequeue_batch (mq, msg_vec, ~0);
+      svm_msg_q_unlock (mq);
+
+      for (i = 0; i < vec_len (msg_vec); i++)
+	{
+	  msg = vec_elt_at_index (msg_vec, i);
+	  e = svm_msg_q_msg_data (mq, msg);
 	  handle_mq_event (e);
-	  svm_msg_q_free_msg (em->our_event_queue, &msg);
+	  svm_msg_q_free_msg (mq, msg);	/* No lock, single thread dequeuing */
 	}
-      if (rv == ETIMEDOUT
-	  && (em->time_to_stop || em->state == STATE_DETACHED))
-	break;
+      vec_reset_length (msg_vec);
     }
+  vec_free (msg_vec);
   pthread_exit (0);
 }
 
@@ -744,6 +783,7 @@ print_usage_and_exit (void)
 	   "  use-svm-api         Use SVM API to connect to VPP\n"
 	   "  test-bytes[:assert] Check data correctness when receiving (assert fails on first error)\n"
 	   "  fifo-size N         Use N Kb fifos\n"
+	   "  mq-size N           Use N event slots for vpp_echo <-> vpp events\n"
 	   "  rx-buf N[Kb|Mb|GB]  Use N[Kb|Mb|GB] RX buffer\n"
 	   "  tx-buf N[Kb|Mb|GB]  Use N[Kb|Mb|GB] TX test buffer\n"
 	   "  appns NAMESPACE     Use the namespace NAMESPACE\n"
@@ -845,6 +885,8 @@ echo_process_opts (int argc, char **argv)
 	;
       else if (unformat (a, "tx-buf %U", unformat_data, &em->tx_buf_size))
 	;
+      else if (unformat (a, "mq-size %d", &em->evt_q_size))
+	;
       else if (unformat (a, "nclients %d", &em->n_clients))
 	{
 	  em->n_sessions = em->n_clients + 1;
@@ -969,6 +1011,7 @@ main (int argc, char **argv)
   em->time_to_stop = 0;
   em->i_am_master = 1;
   em->n_rx_threads = 4;
+  em->evt_q_size = 256;
   em->test_return_packets = RETURN_PACKETS_NOTEST;
   em->timing.start_event = ECHO_EVT_FIRST_QCONNECT;
   em->timing.end_event = ECHO_EVT_LAST_BYTE;
diff --git a/src/plugins/hs_apps/sapi/vpp_echo_bapi.c b/src/plugins/hs_apps/sapi/vpp_echo_bapi.c
index 6a958cb59d6..1894b5de840 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo_bapi.c
+++ b/src/plugins/hs_apps/sapi/vpp_echo_bapi.c
@@ -44,7 +44,7 @@ echo_send_attach (echo_main_t * em)
   bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = em->fifo_size;
   bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
   bmp->options[APP_OPTIONS_SEGMENT_SIZE] = 256 << 20;
-  bmp->options[APP_OPTIONS_EVT_QUEUE_SIZE] = 256;
+  bmp->options[APP_OPTIONS_EVT_QUEUE_SIZE] = em->evt_q_size;
   if (em->appns_id)
     {
       bmp->namespace_id_len = vec_len (em->appns_id);
diff --git a/src/plugins/hs_apps/sapi/vpp_echo_common.h b/src/plugins/hs_apps/sapi/vpp_echo_common.h
index 17a39732a2d..adcfd780f54 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo_common.h
+++ b/src/plugins/hs_apps/sapi/vpp_echo_common.h
@@ -287,6 +287,7 @@ typedef struct
   u8 output_json;		/* Output stats as JSON */
   u8 log_lvl;			/* Verbosity of the logging */
   int max_test_msg;		/* Limit the number of incorrect data messages */
+  u32 evt_q_size;		/* Size of the vpp MQ (app<->vpp events) */
 
   u8 *appns_id;
   u64 appns_flags;