session: don't wait indefinitely for apps to consume evts (VPP-1454)

[vpp.git] / src / vnet / session / session_api.c

diff --git a/src/vnet/session/session_api.c b/src/vnet/session/session_api.c
index 13a3376..e82be56 100755 (executable)
--- a/src/vnet/session/session_api.c
+++ b/src/vnet/session/session_api.c
@@ -147,7 +147,8 @@ send_del_segment_callback (u32 api_client_index, const ssvm_private_t * fs)
 }
 
 static int
-send_app_cut_through_registration_add (u32 api_client_index, u64 mq_addr,
+send_app_cut_through_registration_add (u32 api_client_index,
+                                      u32 wrk_map_index, u64 mq_addr,
                                       u64 peer_mq_addr)
 {
   vl_api_app_cut_through_registration_add_t *mp;
@@ -169,6 +170,7 @@ send_app_cut_through_registration_add (u32 api_client_index, u64 mq_addr,
 
   mp->evt_q_address = mq_addr;
   mp->peer_evt_q_address = peer_mq_addr;
+  mp->wrk_index = wrk_map_index;
 
   mq = uword_to_pointer (mq_addr, svm_msg_q_t *);
   peer_mq = uword_to_pointer (peer_mq_addr, svm_msg_q_t *);
@@ -253,7 +255,7 @@ send_session_accept_callback (stream_session_t * s)
        }
       else
        {
-         ll = application_get_local_listen_session (server_wrk,
+         ll = application_get_local_listen_session (server,
                                                     ls->listener_index);
          if (ll->transport_listener_index != ~0)
            {
@@ -291,14 +293,13 @@ mq_send_local_session_disconnected_cb (u32 app_wrk_index,
   app_mq = app_wrk->event_queue;
   svm_msg_q_lock_and_alloc_msg_w_ring (app_mq, SESSION_MQ_CTRL_EVT_RING,
                                       SVM_Q_WAIT, msg);
-  svm_msg_q_unlock (app_mq);
   evt = svm_msg_q_msg_data (app_mq, msg);
   memset (evt, 0, sizeof (*evt));
   evt->event_type = SESSION_CTRL_EVT_DISCONNECTED;
   mp = (session_disconnected_msg_t *) evt->data;
   mp->handle = application_local_session_handle (ls);
   mp->context = app->api_client_index;
-  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
+  svm_msg_q_add_and_unlock (app_mq, msg);
 }
 
 static void
@@ -420,6 +421,23 @@ static session_cb_vft_t session_cb_vft = {
   .del_segment_callback = send_del_segment_callback,
 };
 
+static int
+mq_try_lock_and_alloc_msg (svm_msg_q_t * app_mq, svm_msg_q_msg_t * msg)
+{
+  int rv;
+  u8 try = 0;
+  while (try < 100)
+    {
+      rv = svm_msg_q_lock_and_alloc_msg_w_ring (app_mq,
+                                               SESSION_MQ_CTRL_EVT_RING,
+                                               SVM_Q_NOWAIT, msg);
+      if (!rv)
+       return 0;
+      try++;
+    }
+  return -1;
+}
+
 static int
 mq_send_session_accepted_cb (stream_session_t * s)
 {
@@ -435,15 +453,14 @@ mq_send_session_accepted_cb (stream_session_t * s)
 
   app = application_get (app_wrk->app_index);
   app_mq = app_wrk->event_queue;
-  svm_msg_q_lock_and_alloc_msg_w_ring (app_mq, SESSION_MQ_CTRL_EVT_RING,
-                                      SVM_Q_WAIT, msg);
-  svm_msg_q_unlock (app_mq);
+  if (mq_try_lock_and_alloc_msg (app_mq, msg))
+    return -1;
 
   evt = svm_msg_q_msg_data (app_mq, msg);
   memset (evt, 0, sizeof (*evt));
   evt->event_type = SESSION_CTRL_EVT_ACCEPTED;
   mp = (session_accepted_msg_t *) evt->data;
-  mp->context = app_wrk->wrk_index;
+  mp->context = app->app_index;
   mp->server_rx_fifo = pointer_to_uword (s->server_rx_fifo);
   mp->server_tx_fifo = pointer_to_uword (s->server_tx_fifo);
 
@@ -475,6 +492,7 @@ mq_send_session_accepted_cb (stream_session_t * s)
       u8 main_thread = vlib_num_workers ()? 1 : 0;
 
       send_app_cut_through_registration_add (app->api_client_index,
+                                            app_wrk->wrk_map_index,
                                             ls->server_evt_q,
                                             ls->client_evt_q);
 
@@ -486,9 +504,7 @@ mq_send_session_accepted_cb (stream_session_t * s)
        }
       else
        {
-         ll =
-           application_get_local_listen_session (app_wrk,
-                                                 ls->listener_index);
+         ll = application_get_local_listen_session (app, ls->listener_index);
          if (ll->transport_listener_index != ~0)
            {
              listener = listen_session_get (ll->transport_listener_index);
@@ -507,7 +523,7 @@ mq_send_session_accepted_cb (stream_session_t * s)
       mp->client_event_queue_address = ls->client_evt_q;
       mp->server_event_queue_address = ls->server_evt_q;
     }
-  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
+  svm_msg_q_add_and_unlock (app_mq, msg);
 
   return 0;
 }
@@ -524,16 +540,15 @@ mq_send_session_disconnected_cb (stream_session_t * s)
 
   app = application_get (app_wrk->app_index);
   app_mq = app_wrk->event_queue;
-  svm_msg_q_lock_and_alloc_msg_w_ring (app_mq, SESSION_MQ_CTRL_EVT_RING,
-                                      SVM_Q_WAIT, msg);
-  svm_msg_q_unlock (app_mq);
+  if (mq_try_lock_and_alloc_msg (app_mq, msg))
+    return;
   evt = svm_msg_q_msg_data (app_mq, msg);
   memset (evt, 0, sizeof (*evt));
   evt->event_type = SESSION_CTRL_EVT_DISCONNECTED;
   mp = (session_disconnected_msg_t *) evt->data;
   mp->handle = session_handle (s);
   mp->context = app->api_client_index;
-  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
+  svm_msg_q_add_and_unlock (app_mq, msg);
 }
 
 static void
@@ -546,15 +561,14 @@ mq_send_session_reset_cb (stream_session_t * s)
   session_event_t *evt;
 
   app_mq = app->event_queue;
-  svm_msg_q_lock_and_alloc_msg_w_ring (app_mq, SESSION_MQ_CTRL_EVT_RING,
-                                      SVM_Q_WAIT, msg);
-  svm_msg_q_unlock (app_mq);
+  if (mq_try_lock_and_alloc_msg (app_mq, msg))
+    return;
   evt = svm_msg_q_msg_data (app_mq, msg);
   memset (evt, 0, sizeof (*evt));
   evt->event_type = SESSION_CTRL_EVT_RESET;
   mp = (session_reset_msg_t *) evt->data;
   mp->handle = session_handle (s);
-  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
+  svm_msg_q_add_and_unlock (app_mq, msg);
 }
 
 static int
@@ -579,9 +593,8 @@ mq_send_session_connected_cb (u32 app_wrk_index, u32 api_context,
       return -1;
     }
 
-  svm_msg_q_lock_and_alloc_msg_w_ring (app_mq, SESSION_MQ_CTRL_EVT_RING,
-                                      SVM_Q_WAIT, msg);
-  svm_msg_q_unlock (app_mq);
+  if (mq_try_lock_and_alloc_msg (app_mq, msg))
+    return -1;
   evt = svm_msg_q_msg_data (app_mq, msg);
   memset (evt, 0, sizeof (*evt));
   evt->event_type = SESSION_CTRL_EVT_CONNECTED;
@@ -615,6 +628,7 @@ mq_send_session_connected_cb (u32 app_wrk_index, u32 api_context,
       u8 main_thread = vlib_num_workers ()? 1 : 0;
 
       send_app_cut_through_registration_add (app->api_client_index,
+                                            app_wrk->wrk_map_index,
                                             ls->client_evt_q,
                                             ls->server_evt_q);
 
@@ -632,7 +646,7 @@ done:
   mp->retval = is_fail ?
     clib_host_to_net_u32 (VNET_API_ERROR_SESSION_CONNECT) : 0;
 
-  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
+  svm_msg_q_add_and_unlock (app_mq, msg);
   return 0;
 }
 
@@ -659,9 +673,9 @@ mq_send_session_bound_cb (u32 app_wrk_index, u32 api_context,
       return -1;
     }
 
-  svm_msg_q_lock_and_alloc_msg_w_ring (app_mq, SESSION_MQ_CTRL_EVT_RING,
-                                      SVM_Q_WAIT, msg);
-  svm_msg_q_unlock (app_mq);
+  if (mq_try_lock_and_alloc_msg (app_mq, msg))
+    return -1;
+
   evt = svm_msg_q_msg_data (app_mq, msg);
   memset (evt, 0, sizeof (*evt));
   evt->event_type = SESSION_CTRL_EVT_BOUND;
@@ -683,7 +697,7 @@ mq_send_session_bound_cb (u32 app_wrk_index, u32 api_context,
   else
     {
       local_session_t *local;
-      local = application_get_local_listen_session_from_handle (handle);
+      local = application_get_local_listener_w_handle (handle);
       mp->lcl_port = local->port;
       mp->lcl_is_ip4 = session_type_is_ip4 (local->session_type);
     }
@@ -698,7 +712,7 @@ mq_send_session_bound_cb (u32 app_wrk_index, u32 api_context,
 
 done:
   mp->retval = rv;
-  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
+  svm_msg_q_add_and_unlock (app_mq, msg);
   return 0;
 }
 
@@ -962,6 +976,7 @@ vl_api_connect_uri_t_handler (vl_api_connect_uri_t * mp)
   app = application_lookup (mp->client_index);
   if (app)
     {
+      memset (a, 0, sizeof (*a));
       a->uri = (char *) mp->uri;
       a->api_context = mp->context;
       a->app_index = app->app_index;
@@ -1235,6 +1250,7 @@ vl_api_unbind_sock_t_handler (vl_api_unbind_sock_t * mp)
     {
       a->app_index = app->app_index;
       a->handle = mp->handle;
+      a->wrk_map_index = mp->wrk_index;
       if ((error = vnet_unbind (a)))
        {
          rv = clib_error_get_code (error);
@@ -1251,7 +1267,7 @@ vl_api_connect_sock_t_handler (vl_api_connect_sock_t * mp)
 {
   vl_api_connect_session_reply_t *rmp;
   vnet_connect_args_t _a, *a = &_a;
-  application_t *app;
+  application_t *app = 0;
   clib_error_t *error = 0;
   int rv = 0;
 
@@ -1303,6 +1319,12 @@ vl_api_connect_sock_t_handler (vl_api_connect_sock_t * mp)
 
 done:
   REPLY_MACRO (VL_API_CONNECT_SESSION_REPLY);
+
+  if (app && application_use_mq_for_ctrl (app))
+    {
+      app_worker_t *app_wrk = application_get_worker (app, mp->wrk_index);
+      mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, 1);
+    }
 }
 
 static void
@@ -1325,7 +1347,7 @@ vl_api_app_worker_add_del_t_handler (vl_api_app_worker_add_del_t * mp)
   if (!reg)
     return;
 
-  app = application_lookup (mp->app_api_index);
+  app = application_lookup (clib_net_to_host_u32 (mp->app_api_index));
   if (!app)
     {
       rv = VNET_API_ERROR_INVALID_VALUE;
@@ -1345,8 +1367,8 @@ vl_api_app_worker_add_del_t_handler (vl_api_app_worker_add_del_t * mp)
       goto done;
     }
 
-  /* Make coverity happy */
-  ASSERT (args.evt_q && args.segment);
+  if (!mp->is_add)
+    goto done;
 
   /* Send fifo segment fd if needed */
   if (ssvm_type (args.segment) == SSVM_SEGMENT_MEMFD)
@@ -1366,9 +1388,9 @@ vl_api_app_worker_add_del_t_handler (vl_api_app_worker_add_del_t * mp)
 done:
   REPLY_MACRO2 (VL_API_APP_WORKER_ADD_DEL_REPLY, ({
     rmp->is_add = mp->is_add;
-    if (!rv)
+    rmp->wrk_index = clib_host_to_net_u32 (args.wrk_index);
+    if (!rv && mp->is_add)
       {
-       rmp->wrk_index = clib_host_to_net_u32 (args.wrk_index);
        if (vec_len (args.segment->name))
          {
            memcpy (rmp->segment_name, args.segment->name,