From 46b91ebb5c350ceeccec97d8e09524f708512342 Mon Sep 17 00:00:00 2001 From: Nathan Skrzypczak Date: Fri, 27 Sep 2019 17:03:15 +0200 Subject: hsa: fix vpp_echo mq locking Type: fix Change-Id: I18d2cde0baaed4134e8378c09aaa88693fb997f8 Signed-off-by: Nathan Skrzypczak --- src/plugins/hs_apps/sapi/vpp_echo.c | 77 +++++++++++++++++++++++------- src/plugins/hs_apps/sapi/vpp_echo_bapi.c | 2 +- src/plugins/hs_apps/sapi/vpp_echo_common.h | 1 + 3 files changed, 62 insertions(+), 18 deletions(-) (limited to 'src/plugins/hs_apps') 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; -- cgit 1.2.3-korg