session: lock app worker mq for io events

Also fixes vcl client/server stats and closing procedure.

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

4 files changed, 25 insertions, 29 deletions

diff --git a/src/vnet/session/application.c b/src/vnet/session/application.c
index 3811091f05c..b58e73eb4f8 100644
--- a/src/vnet/session/application.c
+++ b/src/vnet/session/application.c
@@ -1319,9 +1319,8 @@ static app_send_evt_handler_fn * const app_send_evt_handler_fns[3] = {
 /**
  * Send event to application
  *
- * Logic from queue perspective is non-blocking. That is, if there's
- * not enough space to enqueue a message, we return. However, if the lock
- * flag is set, we do wait for queue mutex.
+ * Logic from queue perspective is non-blocking. If there's
+ * not enough space to enqueue a message, we return.
  */
 int
 app_worker_send_event (app_worker_t * app, stream_session_t * s, u8 evt_type)
@@ -1330,6 +1329,12 @@ app_worker_send_event (app_worker_t * app, stream_session_t * s, u8 evt_type)
   return app_send_evt_handler_fns[evt_type] (app, s, 0 /* lock */ );
 }
 
+/**
+ * Send event to application
+ *
+ * Logic from queue perspective is blocking. However, if queue is full,
+ * we return.
+ */
 int
 app_worker_lock_and_send_event (app_worker_t * app, stream_session_t * s,
 				u8 evt_type)
diff --git a/src/vnet/session/application_interface.c b/src/vnet/session/application_interface.c
index 6c517e8f92b..6b012bb845d 100644
--- a/src/vnet/session/application_interface.c
+++ b/src/vnet/session/application_interface.c
@@ -612,9 +612,14 @@ vnet_disconnect_session (vnet_disconnect_args_t * a)
     }
   else
     {
+      app_worker_t *app_wrk;
       stream_session_t *s;
+
       s = session_get_from_handle_if_valid (a->handle);
-      if (!s || s->app_wrk_index != a->app_index)
+      if (!s)
+	return VNET_API_ERROR_INVALID_VALUE;
+      app_wrk = app_worker_get (s->app_wrk_index);
+      if (app_wrk->app_index != a->app_index)
 	return VNET_API_ERROR_INVALID_VALUE;
 
       /* We're peeking into another's thread pool. Make sure */
diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index c5b2124acbd..472b38baae1 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -486,7 +486,7 @@ stream_session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
  * @return 0 on success or negative number if failed to send notification.
  */
 static inline int
-session_enqueue_notify (stream_session_t * s, u8 lock)
+session_enqueue_notify (stream_session_t * s)
 {
   app_worker_t *app;
 
@@ -504,10 +504,7 @@ session_enqueue_notify (stream_session_t * s, u8 lock)
   }));
   /* *INDENT-ON* */
 
-  if (lock)
-    return app_worker_lock_and_send_event (app, s, FIFO_EVENT_APP_RX);
-
-  return app_worker_send_event (app, s, FIFO_EVENT_APP_RX);
+  return app_worker_lock_and_send_event (app, s, FIFO_EVENT_APP_RX);
 }
 
 int
@@ -519,10 +516,7 @@ session_dequeue_notify (stream_session_t * s)
   if (PREDICT_FALSE (!app))
     return -1;
 
-  if (session_transport_service_type (s) == TRANSPORT_SERVICE_CL)
-    return app_worker_lock_and_send_event (app, s, FIFO_EVENT_APP_RX);
-
-  return app_worker_send_event (app, s, FIFO_EVENT_APP_TX);
+  return app_worker_lock_and_send_event (app, s, FIFO_EVENT_APP_TX);
 }
 
 /**
@@ -537,14 +531,11 @@ int
 session_manager_flush_enqueue_events (u8 transport_proto, u32 thread_index)
 {
   session_manager_main_t *smm = &session_manager_main;
-  transport_service_type_t tp_service;
-  int i, errors = 0, lock;
   stream_session_t *s;
+  int i, errors = 0;
   u32 *indices;
 
   indices = smm->session_to_enqueue[transport_proto][thread_index];
-  tp_service = transport_protocol_service_type (transport_proto);
-  lock = tp_service == TRANSPORT_SERVICE_CL;
 
   for (i = 0; i < vec_len (indices); i++)
     {
@@ -554,7 +545,7 @@ session_manager_flush_enqueue_events (u8 transport_proto, u32 thread_index)
 	  errors++;
 	  continue;
 	}
-      if (PREDICT_FALSE (session_enqueue_notify (s, lock)))
+      if (PREDICT_FALSE (session_enqueue_notify (s)))
 	errors++;
     }
 
@@ -838,7 +829,7 @@ stream_session_accept (transport_connection_t * tc, u32 listener_index,
 
   /* Find the server */
   listener = listen_session_get (listener_index);
-  app_wrk = app_worker_get (listener->app_wrk_index);
+  app_wrk = application_listener_select_worker (listener, 0);
 
   sm = app_worker_get_listen_segment_manager (app_wrk, listener);
   if ((rv = session_alloc_and_init (sm, tc, 1, &s)))
diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index 119cdd812ad..f5aed7490ea 100644
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -29,6 +29,7 @@ session_mq_accepted_reply_handler (void *data)
 {
   session_accepted_reply_msg_t *mp = (session_accepted_reply_msg_t *) data;
   vnet_disconnect_args_t _a = { 0 }, *a = &_a;
+  app_worker_t *app_wrk;
   local_session_t *ls;
   stream_session_t *s;
 
@@ -43,8 +44,6 @@ session_mq_accepted_reply_handler (void *data)
 
   if (session_handle_is_local (mp->handle))
     {
-      app_worker_t *app_wrk;
-      application_t *app;
       ls = application_get_local_session_from_handle (mp->handle);
       if (!ls)
 	{
@@ -52,8 +51,7 @@ session_mq_accepted_reply_handler (void *data)
 	  return;
 	}
       app_wrk = app_worker_get (ls->app_wrk_index);
-      app = application_get (app_wrk->app_index);
-      if (app->app_index != mp->context)
+      if (app_wrk->app_index != mp->context)
 	{
 	  clib_warning ("server %u doesn't own local handle 0x%lx",
 			mp->context, mp->handle);
@@ -71,18 +69,15 @@ session_mq_accepted_reply_handler (void *data)
 	  clib_warning ("session doesn't exist");
 	  return;
 	}
-      if (s->app_wrk_index != mp->context)
+      app_wrk = app_worker_get (s->app_wrk_index);
+      if (app_wrk->app_index != mp->context)
 	{
 	  clib_warning ("app doesn't own session");
 	  return;
 	}
       s->session_state = SESSION_STATE_READY;
       if (!svm_fifo_is_empty (s->server_rx_fifo))
-	{
-	  app_worker_t *app;
-	  app = app_worker_get (s->app_wrk_index);
-	  app_worker_send_event (app, s, FIFO_EVENT_APP_RX);
-	}
+	app_worker_lock_and_send_event (app_wrk, s, FIFO_EVENT_APP_RX);
     }
 }