session: free ctrl event data on connect rpc

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

diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index af7a6ed..8817b08 100644 (file)
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -213,6 +213,8 @@ session_mq_handle_connects_rpc (void *arg)
 
   /* Pending connects on linked list pertaining to first worker */
   fwrk = session_main_get_worker (1);
+  if (!fwrk->n_pending_connects)
+    goto update_state;
 
   vlib_worker_thread_barrier_sync (vm);
 
@@ -225,24 +227,32 @@ session_mq_handle_connects_rpc (void *arg)
       next = clib_llist_next (fwrk->event_elts, evt_list, elt);
       clib_llist_remove (fwrk->event_elts, evt_list, elt);
       session_mq_connect_one (session_evt_ctrl_data (fwrk, elt));
+      session_evt_ctrl_data_free (fwrk, elt);
       clib_llist_put (fwrk->event_elts, elt);
       elt = next;
       n_connects += 1;
     }
 
+  /* Decrement with worker barrier */
+  fwrk->n_pending_connects -= n_connects;
+
+  vlib_worker_thread_barrier_release (vm);
+
+update_state:
+
   /* Switch worker to poll mode if it was in interrupt mode and had work or
    * back to interrupt if threshold of loops without a connect is passed.
    * While in poll mode, reprogram connects rpc */
   wrk = session_main_get_worker (0);
   if (wrk->state != SESSION_WRK_POLLING)
     {
-      if (!n_connects)
-       goto done;
-
-      session_wrk_set_state (wrk, SESSION_WRK_POLLING);
-      vlib_node_set_state (vm, session_queue_node.index,
-                          VLIB_NODE_STATE_POLLING);
-      wrk->no_connect_loops = 0;
+      if (n_connects)
+       {
+         session_wrk_set_state (wrk, SESSION_WRK_POLLING);
+         vlib_node_set_state (vm, session_queue_node.index,
+                              VLIB_NODE_STATE_POLLING);
+         wrk->no_connect_loops = 0;
+       }
     }
   else
     {
@@ -253,20 +263,18 @@ session_mq_handle_connects_rpc (void *arg)
              session_wrk_set_state (wrk, SESSION_WRK_INTERRUPT);
              vlib_node_set_state (vm, session_queue_node.index,
                                   VLIB_NODE_STATE_INTERRUPT);
-             fwrk->pending_connects_ntf = 0;
-             goto done;
            }
        }
       else
        wrk->no_connect_loops = 0;
     }
 
-  elt = session_evt_alloc_ctrl (wrk);
-  elt->evt.event_type = SESSION_CTRL_EVT_RPC;
-  elt->evt.rpc_args.fp = session_mq_handle_connects_rpc;
-
-done:
-  vlib_worker_thread_barrier_release (vm);
+  if (wrk->state == SESSION_WRK_POLLING)
+    {
+      elt = session_evt_alloc_ctrl (wrk);
+      elt->evt.event_type = SESSION_CTRL_EVT_RPC;
+      elt->evt.rpc_args.fp = session_mq_handle_connects_rpc;
+    }
 }
 
 static void
@@ -292,12 +300,13 @@ session_mq_connect_handler (session_worker_t *wrk, session_evt_elt_t *elt)
   he = clib_llist_elt (wrk->event_elts, wrk->pending_connects);
   clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
 
-  if (!wrk->pending_connects_ntf)
+  /* Decremented with worker barrier */
+  wrk->n_pending_connects += 1;
+  if (wrk->n_pending_connects == 1)
     {
       vlib_node_set_interrupt_pending (vlib_get_main_by_index (0),
                                       session_queue_node.index);
       session_send_rpc_evt_to_thread (0, session_mq_handle_connects_rpc, 0);
-      wrk->pending_connects_ntf = 1;
     }
 }
 
@@ -379,15 +388,20 @@ app_mq_detach_handler (void *data)
 }
 
 static void
-session_mq_unlisten_handler (void *data)
+session_mq_unlisten_rpc (session_unlisten_msg_t *mp)
 {
-  session_unlisten_msg_t *mp = (session_unlisten_msg_t *) data;
+  vlib_main_t *vm = vlib_get_main ();
   vnet_unlisten_args_t _a, *a = &_a;
   app_worker_t *app_wrk;
+  session_handle_t sh;
   application_t *app;
+  u32 context;
   int rv;
 
-  app_check_thread_and_barrier (session_mq_unlisten_handler, mp);
+  sh = mp->handle;
+  context = mp->context;
+
+  vlib_worker_thread_barrier_sync (vm);
 
   app = application_lookup (mp->client_index);
   if (!app)
@@ -395,7 +409,7 @@ session_mq_unlisten_handler (void *data)
 
   clib_memset (a, 0, sizeof (*a));
   a->app_index = app->app_index;
-  a->handle = mp->handle;
+  a->handle = sh;
   a->wrk_map_index = mp->wrk_index;
   if ((rv = vnet_unlisten (a)))
     clib_warning ("unlisten returned: %d", rv);
@@ -404,7 +418,29 @@ session_mq_unlisten_handler (void *data)
   if (!app_wrk)
     return;
 
-  mq_send_unlisten_reply (app_wrk, mp->handle, mp->context, rv);
+  vlib_worker_thread_barrier_release (vm);
+
+  mq_send_unlisten_reply (app_wrk, sh, context, rv);
+  clib_mem_free (mp);
+}
+
+static void
+session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+  u32 thread_index = wrk - session_main.wrk;
+  session_unlisten_msg_t *mp, *arg;
+
+  mp = session_evt_ctrl_data (wrk, elt);
+  arg = clib_mem_alloc (sizeof (session_unlisten_msg_t));
+  clib_memcpy_fast (arg, mp, sizeof (*arg));
+
+  if (PREDICT_FALSE (!thread_index))
+    {
+      session_mq_unlisten_rpc (arg);
+      return;
+    }
+
+  session_send_rpc_evt_to_thread_force (0, session_mq_unlisten_rpc, arg);
 }
 
 static void
@@ -449,7 +485,7 @@ session_mq_accepted_reply_handler (void *data)
   if (!session_has_transport (s))
     {
       s->session_state = SESSION_STATE_READY;
-      if (ct_session_connect_notify (s))
+      if (ct_session_connect_notify (s, SESSION_E_NONE))
        return;
     }
   else
@@ -1460,7 +1496,7 @@ session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
       session_mq_listen_uri_handler (session_evt_ctrl_data (wrk, elt));
       break;
     case SESSION_CTRL_EVT_UNLISTEN:
-      session_mq_unlisten_handler (session_evt_ctrl_data (wrk, elt));
+      session_mq_unlisten_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_CONNECT:
       session_mq_connect_handler (wrk, elt);