svm: allow mq attachments at random offsets

Type: feature

Signed-off-by: Florin Coras <fcoras@cisco.com>
Change-Id: Ic373cd2c11272da539eb4b0db27227f36f2f9688

6 files changed, 77 insertions, 36 deletions

diff --git a/src/plugins/hs_apps/sapi/vpp_echo.c b/src/plugins/hs_apps/sapi/vpp_echo.c
index a47a4d455d8..19b5808c550 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo.c
+++ b/src/plugins/hs_apps/sapi/vpp_echo.c
@@ -556,16 +556,14 @@ session_accepted_handler (session_accepted_msg_t * mp)
   session = echo_session_new (em);
 
   if (echo_attach_session (mp->segment_handle, mp->server_rx_fifo,
-			   mp->server_tx_fifo, session))
+			   mp->server_tx_fifo, mp->vpp_event_queue_address,
+			   session))
     {
       ECHO_FAIL (ECHO_FAIL_ACCEPTED_WAIT_FOR_SEG_ALLOC,
 		 "accepted wait_for_segment_allocation errored");
       return;
     }
 
-  session->vpp_evt_q =
-    uword_to_pointer (mp->vpp_event_queue_address, svm_msg_q_t *);
-
   session->vpp_session_handle = mp->handle;
 
   /* session->transport needed by app_send_dgram */
@@ -617,14 +615,14 @@ session_connected_handler (session_connected_msg_t * mp)
   session = echo_session_new (em);
 
   if (echo_attach_session (mp->segment_handle, mp->server_rx_fifo,
-			   mp->server_tx_fifo, session))
+			   mp->server_tx_fifo, mp->vpp_event_queue_address,
+			   session))
     {
       ECHO_FAIL (ECHO_FAIL_CONNECTED_WAIT_FOR_SEG_ALLOC,
 		 "connected wait_for_segment_allocation errored");
       return;
     }
-  session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-					 svm_msg_q_t *);
+
   session->vpp_session_handle = mp->handle;
   session->start = clib_time_now (&em->clib_time);
   session->listener_index = listener_index;
@@ -806,7 +804,7 @@ 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;
+  svm_msg_q_t *mq = &em->rpc_msq_queue;
 
   while (em->state < STATE_DATA_DONE && !em->time_to_stop)
     {
@@ -1321,7 +1319,7 @@ main (int argc, char **argv)
   cfg->n_rings = 1;
   cfg->q_nitems = rpc_queue_size;
   cfg->ring_cfgs = rc;
-  em->rpc_msq_queue = svm_msg_q_alloc (cfg);
+  svm_msg_q_attach (&em->rpc_msq_queue, svm_msg_q_alloc (cfg));
 
   signal (SIGINT, stop_signal);
   signal (SIGQUIT, stop_signal);
diff --git a/src/plugins/hs_apps/sapi/vpp_echo_bapi.c b/src/plugins/hs_apps/sapi/vpp_echo_bapi.c
index c643cec2ce3..6ad825d7872 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo_bapi.c
+++ b/src/plugins/hs_apps/sapi/vpp_echo_bapi.c
@@ -264,12 +264,12 @@ echo_segment_detach (u64 segment_handle)
 
 int
 echo_attach_session (uword segment_handle, uword rxf_offset, uword txf_offset,
-		     echo_session_t *s)
+		     uword mq_offset, echo_session_t *s)
 {
   svm_fifo_shared_t *rx_fifo, *tx_fifo;
   echo_main_t *em = &echo_main;
+  u32 fs_index, eqs_index;
   fifo_segment_t *fs;
-  u32 fs_index;
 
   fs_index = echo_segment_lookup (segment_handle);
   if (fs_index == (u32) ~0)
@@ -279,6 +279,12 @@ echo_attach_session (uword segment_handle, uword rxf_offset, uword txf_offset,
       return -1;
     }
 
+  if (mq_offset != (uword) ~0)
+    {
+      eqs_index = echo_segment_lookup (ECHO_MQ_SEG_HANDLE);
+      ASSERT (eqs_index != (u32) ~0);
+    }
+
   rx_fifo = uword_to_pointer (rxf_offset, svm_fifo_shared_t *);
   tx_fifo = uword_to_pointer (txf_offset, svm_fifo_shared_t *);
   rx_fifo->client_session_index = s->session_index;
@@ -290,6 +296,39 @@ echo_attach_session (uword segment_handle, uword rxf_offset, uword txf_offset,
   s->rx_fifo = fifo_segment_alloc_fifo_w_shared (fs, rx_fifo);
   s->tx_fifo = fifo_segment_alloc_fifo_w_shared (fs, tx_fifo);
 
+  if (mq_offset != (uword) ~0)
+    {
+      fs = fifo_segment_get_segment (&em->segment_main, eqs_index);
+      s->vpp_evt_q =
+	fifo_segment_msg_q_attach (fs, mq_offset, rx_fifo->slice_index);
+    }
+
+  clib_spinlock_unlock (&em->segment_handles_lock);
+
+  return 0;
+}
+
+int
+echo_segment_attach_mq (uword segment_handle, uword mq_offset, u32 mq_index,
+			svm_msg_q_t **mq)
+{
+  echo_main_t *em = &echo_main;
+  fifo_segment_t *fs;
+  u32 fs_index;
+
+  fs_index = echo_segment_lookup (segment_handle);
+  if (fs_index == (u32) ~0)
+    {
+      ECHO_LOG (0, "ERROR: mq segment %lx for is not attached!",
+		segment_handle);
+      return -1;
+    }
+
+  clib_spinlock_lock (&em->segment_handles_lock);
+
+  fs = fifo_segment_get_segment (&em->segment_main, fs_index);
+  *mq = fifo_segment_msg_q_attach (fs, mq_offset, mq_index);
+
   clib_spinlock_unlock (&em->segment_handles_lock);
 
   return 0;
@@ -338,8 +377,6 @@ static void
   em->state = STATE_CLEANED_CERT_KEY;
 }
 
-#define ECHO_MQ_SEG_HANDLE ((u64) ~0 - 1)
-
 static void
 vl_api_app_attach_reply_t_handler (vl_api_app_attach_reply_t * mp)
 {
@@ -364,8 +401,6 @@ vl_api_app_attach_reply_t_handler (vl_api_app_attach_reply_t * mp)
       ECHO_FAIL (ECHO_FAIL_VL_API_NULL_APP_MQ, "NULL app_mq");
       return;
     }
-  em->app_mq = uword_to_pointer (mp->app_mq, svm_msg_q_t *);
-  em->ctrl_mq = uword_to_pointer (mp->vpp_ctrl_mq, svm_msg_q_t *);
 
   if (mp->n_fds)
     {
@@ -385,6 +420,8 @@ vl_api_app_attach_reply_t_handler (vl_api_app_attach_reply_t * mp)
 		       "svm_fifo_segment_attach failed on SSVM_SEGMENT_MEMFD");
 	    goto failed;
 	  }
+      echo_segment_attach_mq (ECHO_MQ_SEG_HANDLE, mp->vpp_ctrl_mq,
+			      mp->vpp_ctrl_mq_thread, &em->ctrl_mq);
 
       if (mp->fd_flags & SESSION_FD_F_MEMFD_SEGMENT)
 	{
@@ -401,6 +438,7 @@ vl_api_app_attach_reply_t_handler (vl_api_app_attach_reply_t * mp)
 	    }
 	  vec_free (segment_name);
 	}
+      echo_segment_attach_mq (segment_handle, mp->app_mq, 0, &em->app_mq);
 
       if (mp->fd_flags & SESSION_FD_F_MQ_EVENTFD)
 	svm_msg_q_set_consumer_eventfd (em->app_mq, fds[n_fds++]);
diff --git a/src/plugins/hs_apps/sapi/vpp_echo_common.c b/src/plugins/hs_apps/sapi/vpp_echo_common.c
index 497f56c3e1e..e24629b783e 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo_common.c
+++ b/src/plugins/hs_apps/sapi/vpp_echo_common.c
@@ -543,23 +543,23 @@ echo_send_rpc (echo_main_t * em, void *fp, echo_rpc_args_t * args)
 {
   svm_msg_q_msg_t msg;
   echo_rpc_msg_t *evt;
-  if (PREDICT_FALSE (svm_msg_q_lock (em->rpc_msq_queue)))
+  if (PREDICT_FALSE (svm_msg_q_lock (&em->rpc_msq_queue)))
     {
       ECHO_FAIL (ECHO_FAIL_RPC_SIZE, "RPC lock failed");
       return -1;
     }
-  if (PREDICT_FALSE (svm_msg_q_ring_is_full (em->rpc_msq_queue, 0)))
+  if (PREDICT_FALSE (svm_msg_q_ring_is_full (&em->rpc_msq_queue, 0)))
     {
-      svm_msg_q_unlock (em->rpc_msq_queue);
+      svm_msg_q_unlock (&em->rpc_msq_queue);
       ECHO_FAIL (ECHO_FAIL_RPC_SIZE, "RPC ring is full");
       return -2;
     }
-  msg = svm_msg_q_alloc_msg_w_ring (em->rpc_msq_queue, 0);
-  evt = (echo_rpc_msg_t *) svm_msg_q_msg_data (em->rpc_msq_queue, &msg);
+  msg = svm_msg_q_alloc_msg_w_ring (&em->rpc_msq_queue, 0);
+  evt = (echo_rpc_msg_t *) svm_msg_q_msg_data (&em->rpc_msq_queue, &msg);
   evt->fp = fp;
   clib_memcpy (&evt->args, args, sizeof (evt->args));
 
-  svm_msg_q_add_and_unlock (em->rpc_msq_queue, &msg);
+  svm_msg_q_add_and_unlock (&em->rpc_msq_queue, &msg);
   return 0;
 }
 
diff --git a/src/plugins/hs_apps/sapi/vpp_echo_common.h b/src/plugins/hs_apps/sapi/vpp_echo_common.h
index cd2bbb6038c..dc5f7dfb9b5 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo_common.h
+++ b/src/plugins/hs_apps/sapi/vpp_echo_common.h
@@ -38,6 +38,7 @@
 #define TIMEOUT 10.0
 #define LOGGING_BATCH (100)
 #define LOG_EVERY_N_IDLE_CYCLES (1e8)
+#define ECHO_MQ_SEG_HANDLE	((u64) ~0 - 1)
 
 #define foreach_echo_fail_code                                          \
   _(ECHO_FAIL_NONE, "ECHO_FAIL_NONE")                                   \
@@ -300,7 +301,7 @@ typedef struct
   uword *shared_segment_handles;	/* Hash table : segment_names -> 1 */
   clib_spinlock_t segment_handles_lock;	/* Hash table lock */
   echo_proto_cb_vft_t *proto_cb_vft;
-  svm_msg_q_t *rpc_msq_queue;	/* MQ between quic_echo threads */
+  svm_msg_q_t rpc_msq_queue; /* MQ between quic_echo threads */
   fifo_segment_main_t segment_main;
 
   /* State of the connection, shared between msg RX thread and main thread */
@@ -444,7 +445,9 @@ int echo_segment_attach (u64 segment_handle, char *name,
 u32 echo_segment_lookup (u64 segment_handle);
 void echo_segment_detach (u64 segment_handle);
 int echo_attach_session (uword segment_handle, uword rxf_offset,
-			 uword txf_offset, echo_session_t *s);
+			 uword mq_offset, uword txf_offset, echo_session_t *s);
+int echo_segment_attach_mq (uword segment_handle, uword mq_offset,
+			    u32 mq_index, svm_msg_q_t **mq);
 
 /* Binary API */
 
diff --git a/src/plugins/hs_apps/sapi/vpp_echo_proto_udp.c b/src/plugins/hs_apps/sapi/vpp_echo_proto_udp.c
index 9689a83d76b..10dfcf02067 100644
--- a/src/plugins/hs_apps/sapi/vpp_echo_proto_udp.c
+++ b/src/plugins/hs_apps/sapi/vpp_echo_proto_udp.c
@@ -132,14 +132,13 @@ udp_echo_bound_uri_cb (session_bound_msg_t * mp, echo_session_t * session)
     return;
 
   if (echo_attach_session (mp->segment_handle, mp->rx_fifo, mp->tx_fifo,
-			   session))
+			   mp->vpp_evt_q, session))
     {
       ECHO_FAIL (ECHO_FAIL_ACCEPTED_WAIT_FOR_SEG_ALLOC,
 		 "accepted wait_for_segment_allocation errored");
       return;
     }
 
-  session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *);
   session->transport.is_ip4 = mp->lcl_is_ip4;
   clib_memcpy_fast (&session->transport.lcl_ip, mp->lcl_ip,
 		    sizeof (ip46_address_t));
diff --git a/src/plugins/unittest/session_test.c b/src/plugins/unittest/session_test.c
index f54ed9ff1ac..68605b2cbd7 100644
--- a/src/plugins/unittest/session_test.c
+++ b/src/plugins/unittest/session_test.c
@@ -1909,8 +1909,9 @@ session_test_mq_basic (vlib_main_t * vm, unformat_input_t * input)
   svm_msg_q_cfg_t _cfg, *cfg = &_cfg;
   svm_msg_q_msg_t msg1, msg2, msg[12];
   int __clib_unused verbose, i, rv;
-  svm_msg_q_t *mq;
+  svm_msg_q_shared_t *smq;
   svm_msg_q_ring_t *ring;
+  svm_msg_q_t _mq = { 0 }, *mq = &_mq;
   u8 *rings_ptr;
 
   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
@@ -1933,28 +1934,30 @@ session_test_mq_basic (vlib_main_t * vm, unformat_input_t * input)
   cfg->q_nitems = 16;
   cfg->ring_cfgs = rc;
 
-  mq = svm_msg_q_alloc (cfg);
+  smq = svm_msg_q_alloc (cfg);
+  svm_msg_q_attach (mq, smq);
   SESSION_TEST (mq != 0, "svm_msg_q_alloc");
   SESSION_TEST (vec_len (mq->rings) == 2, "ring allocation");
-  rings_ptr = (u8 *) mq->rings + vec_bytes (mq->rings);
+  rings_ptr = (u8 *) mq->rings[0].shr->data;
   vec_foreach (ring, mq->rings)
   {
-    SESSION_TEST (ring->data == rings_ptr, "ring data");
+    SESSION_TEST (ring->shr->data == rings_ptr, "ring data");
     rings_ptr += (uword) ring->nitems * ring->elsize;
+    rings_ptr += sizeof (svm_msg_q_ring_shared_t);
   }
 
   msg1 = svm_msg_q_alloc_msg (mq, 8);
-  rv = (mq->rings[0].cursize != 1
-	|| msg1.ring_index != 0 || msg1.elt_index != 0);
+  rv = (mq->rings[0].shr->cursize != 1 || msg1.ring_index != 0 ||
+	msg1.elt_index != 0);
   SESSION_TEST (rv == 0, "msg alloc1");
 
   msg2 = svm_msg_q_alloc_msg (mq, 15);
-  rv = (mq->rings[1].cursize != 1
-	|| msg2.ring_index != 1 || msg2.elt_index != 0);
+  rv = (mq->rings[1].shr->cursize != 1 || msg2.ring_index != 1 ||
+	msg2.elt_index != 0);
   SESSION_TEST (rv == 0, "msg alloc2");
 
   svm_msg_q_free_msg (mq, &msg1);
-  SESSION_TEST (mq->rings[0].cursize == 0, "free msg");
+  SESSION_TEST (mq->rings[0].shr->cursize == 0, "free msg");
 
   for (i = 0; i < 12; i++)
     {
@@ -1962,7 +1965,7 @@ session_test_mq_basic (vlib_main_t * vm, unformat_input_t * input)
       *(u32 *) svm_msg_q_msg_data (mq, &msg[i]) = i;
     }
 
-  rv = (mq->rings[0].cursize != 8 || mq->rings[1].cursize != 5);
+  rv = (mq->rings[0].shr->cursize != 8 || mq->rings[1].shr->cursize != 5);
   SESSION_TEST (rv == 0, "msg alloc3");
 
   *(u32 *) svm_msg_q_msg_data (mq, &msg2) = 123;
@@ -1998,7 +2001,7 @@ session_test_mq_basic (vlib_main_t * vm, unformat_input_t * input)
 	SESSION_TEST (0, "dequeue2 wrong data");
       svm_msg_q_free_msg (mq, &msg[i]);
     }
-  rv = (mq->rings[0].cursize == 0 && mq->rings[1].cursize == 0);
+  rv = (mq->rings[0].shr->cursize == 0 && mq->rings[1].shr->cursize == 0);
   SESSION_TEST (rv, "post dequeue");
 
   return 0;