1 files changed, 319 insertions, 53 deletions

diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index 1d662a20e3c..ad18637a952 100644
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -25,12 +25,202 @@
 #include <vnet/session/session_debug.h>
 #include <svm/queue.h>
 
-static void session_mq_accepted_reply_handler (void *data);
+#define app_check_thread_and_barrier(_fn, _arg)				\
+  if (!vlib_thread_is_main_w_barrier ())				\
+    {									\
+     vlib_rpc_call_main_thread (_fn, (u8 *) _arg, sizeof(*_arg));	\
+      return;								\
+   }
 
 static void
-accepted_notify_cb (void *data, u32 data_len)
+session_mq_listen_handler (void *data)
 {
-  session_mq_accepted_reply_handler (data);
+  session_listen_msg_t *mp = (session_listen_msg_t *) data;
+  vnet_listen_args_t _a, *a = &_a;
+  app_worker_t *app_wrk;
+  application_t *app;
+  int rv;
+
+  app_check_thread_and_barrier (session_mq_listen_handler, mp);
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  clib_memset (a, 0, sizeof (*a));
+  a->sep.is_ip4 = mp->is_ip4;
+  clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
+  a->sep.port = mp->port;
+  a->sep.fib_index = mp->vrf;
+  a->sep.sw_if_index = ENDPOINT_INVALID_INDEX;
+  a->sep.transport_proto = mp->proto;
+  a->app_index = app->app_index;
+  a->wrk_map_index = mp->wrk_index;
+
+  if ((rv = vnet_listen (a)))
+    clib_warning ("listen returned: %d", rv);
+
+  app_wrk = application_get_worker (app, mp->wrk_index);
+  mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
+  return;
+}
+
+static void
+session_mq_listen_uri_handler (void *data)
+{
+  session_listen_uri_msg_t *mp = (session_listen_uri_msg_t *) data;
+  vnet_listen_args_t _a, *a = &_a;
+  app_worker_t *app_wrk;
+  application_t *app;
+  int rv;
+
+  app_check_thread_and_barrier (session_mq_listen_uri_handler, mp);
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  clib_memset (a, 0, sizeof (*a));
+  a->uri = (char *) mp->uri;
+  a->app_index = app->app_index;
+  rv = vnet_bind_uri (a);
+
+  app_wrk = application_get_worker (app, 0);
+  mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
+}
+
+static void
+session_mq_connect_handler (void *data)
+{
+  session_connect_msg_t *mp = (session_connect_msg_t *) data;
+  vnet_connect_args_t _a, *a = &_a;
+  app_worker_t *app_wrk;
+  application_t *app;
+  int rv;
+
+  app_check_thread_and_barrier (session_mq_connect_handler, mp);
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  clib_memset (a, 0, sizeof (*a));
+  a->sep.is_ip4 = mp->is_ip4;
+  clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
+  a->sep.port = mp->port;
+  a->sep.transport_proto = mp->proto;
+  a->sep.peer.fib_index = mp->vrf;
+  a->sep.peer.sw_if_index = ENDPOINT_INVALID_INDEX;
+  a->sep_ext.parent_handle = mp->parent_handle;
+  if (mp->hostname_len)
+    {
+      vec_validate (a->sep_ext.hostname, mp->hostname_len - 1);
+      clib_memcpy_fast (a->sep_ext.hostname, mp->hostname, mp->hostname_len);
+    }
+  a->api_context = mp->context;
+  a->app_index = app->app_index;
+  a->wrk_map_index = mp->wrk_index;
+
+  if ((rv = vnet_connect (a)))
+    {
+      clib_warning ("connect returned: %U", format_vnet_api_errno, rv);
+      app_wrk = application_get_worker (app, mp->wrk_index);
+      mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0,
+				    /* is_fail */ 1);
+    }
+
+  vec_free (a->sep_ext.hostname);
+}
+
+static void
+session_mq_connect_uri_handler (void *data)
+{
+  session_connect_uri_msg_t *mp = (session_connect_uri_msg_t *) data;
+  vnet_connect_args_t _a, *a = &_a;
+  app_worker_t *app_wrk;
+  application_t *app;
+  int rv;
+
+  app_check_thread_and_barrier (session_mq_connect_uri_handler, mp);
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  clib_memset (a, 0, sizeof (*a));
+  a->uri = (char *) mp->uri;
+  a->api_context = mp->context;
+  a->app_index = app->app_index;
+  if ((rv = vnet_connect_uri (a)))
+    {
+      clib_warning ("connect_uri returned: %d", rv);
+      app_wrk = application_get_worker (app, 0 /* default wrk only */ );
+      mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0,
+				    /* is_fail */ 1);
+    }
+}
+
+static void
+session_mq_disconnect_handler (void *data)
+{
+  session_disconnect_msg_t *mp = (session_disconnect_msg_t *) data;
+  vnet_disconnect_args_t _a, *a = &_a;
+  application_t *app;
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  a->app_index = app->app_index;
+  a->handle = mp->handle;
+  vnet_disconnect_session (a);
+}
+
+static void
+app_mq_detach_handler (void *data)
+{
+  session_app_detach_msg_t *mp = (session_app_detach_msg_t *) data;
+  vnet_app_detach_args_t _a, *a = &_a;
+  application_t *app;
+
+  app_check_thread_and_barrier (app_mq_detach_handler, mp);
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  a->app_index = app->app_index;
+  a->api_client_index = mp->client_index;
+  vnet_application_detach (a);
+}
+
+static void
+session_mq_unlisten_handler (void *data)
+{
+  session_unlisten_msg_t *mp = (session_unlisten_msg_t *) data;
+  vnet_unlisten_args_t _a, *a = &_a;
+  app_worker_t *app_wrk;
+  application_t *app;
+  int rv;
+
+  app_check_thread_and_barrier (session_mq_unlisten_handler, mp);
+
+  app = application_lookup (mp->client_index);
+  if (!app)
+    return;
+
+  clib_memset (a, 0, sizeof (*a));
+  a->app_index = app->app_index;
+  a->handle = mp->handle;
+  a->wrk_map_index = mp->wrk_index;
+  if ((rv = vnet_unlisten (a)))
+    clib_warning ("unlisten returned: %d", rv);
+
+  app_wrk = application_get_worker (app, a->wrk_map_index);
+  if (!app_wrk)
+    return;
+
+  mq_send_unlisten_reply (app_wrk, mp->handle, mp->context, rv);
 }
 
 static void
@@ -56,8 +246,8 @@ session_mq_accepted_reply_handler (void *data)
   if (vlib_num_workers () && vlib_get_thread_index () != 0
       && session_thread_from_handle (mp->handle) == 0)
     {
-      vl_api_rpc_call_main_thread (accepted_notify_cb, data,
-				   sizeof (session_accepted_reply_msg_t));
+      vlib_rpc_call_main_thread (session_mq_accepted_reply_handler,
+				 (u8 *) mp, sizeof (*mp));
       return;
     }
 
@@ -859,14 +1049,93 @@ session_event_get_session (session_event_t * e, u8 thread_index)
 }
 
 always_inline void
-session_event_dispatch (session_worker_t * wrk, vlib_node_runtime_t * node,
-			session_evt_elt_t * elt, u32 thread_index,
-			int *n_tx_packets)
+session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
+{
+  clib_llist_index_t ei;
+  void (*fp) (void *);
+  session_event_t *e;
+  session_t *s;
+
+  ei = clib_llist_entry_index (wrk->event_elts, elt);
+  e = &elt->evt;
+
+  switch (e->event_type)
+    {
+    case SESSION_CTRL_EVT_RPC:
+      fp = e->rpc_args.fp;
+      (*fp) (e->rpc_args.arg);
+      break;
+    case SESSION_CTRL_EVT_CLOSE:
+      s = session_get_from_handle_if_valid (e->session_handle);
+      if (PREDICT_FALSE (!s))
+	break;
+      session_transport_close (s);
+      break;
+    case SESSION_CTRL_EVT_RESET:
+      s = session_get_from_handle_if_valid (e->session_handle);
+      if (PREDICT_FALSE (!s))
+	break;
+      session_transport_reset (s);
+      break;
+    case SESSION_CTRL_EVT_LISTEN:
+      session_mq_listen_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_LISTEN_URI:
+      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));
+      break;
+    case SESSION_CTRL_EVT_CONNECT:
+      session_mq_connect_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_CONNECT_URI:
+      session_mq_connect_uri_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_DISCONNECT:
+      session_mq_disconnect_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_DISCONNECTED:
+      session_mq_disconnected_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_ACCEPTED_REPLY:
+      session_mq_accepted_reply_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_DISCONNECTED_REPLY:
+      session_mq_disconnected_reply_handler (session_evt_ctrl_data (wrk,
+								    elt));
+      break;
+    case SESSION_CTRL_EVT_RESET_REPLY:
+      session_mq_reset_reply_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_WORKER_UPDATE:
+      session_mq_worker_update_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    case SESSION_CTRL_EVT_APP_DETACH:
+      app_mq_detach_handler (session_evt_ctrl_data (wrk, elt));
+      break;
+    default:
+      clib_warning ("unhandled event type %d", e->event_type);
+    }
+
+  /* Regrab elements in case pool moved */
+  elt = pool_elt_at_index (wrk->event_elts, ei);
+  if (!clib_llist_elt_is_linked (elt, evt_list))
+    {
+      if (e->event_type >= SESSION_CTRL_EVT_BOUND)
+	session_evt_ctrl_data_free (wrk, elt);
+      session_evt_elt_free (wrk, elt);
+    }
+}
+
+always_inline void
+session_event_dispatch_io (session_worker_t * wrk, vlib_node_runtime_t * node,
+			   session_evt_elt_t * elt, u32 thread_index,
+			   int *n_tx_packets)
 {
   session_main_t *smm = &session_main;
   app_worker_t *app_wrk;
   clib_llist_index_t ei;
-  void (*fp) (void *);
   session_event_t *e;
   session_t *s;
 
@@ -896,18 +1165,6 @@ session_event_dispatch (session_worker_t * wrk, vlib_node_runtime_t * node,
       transport_app_rx_evt (session_get_transport_proto (s),
 			    s->connection_index, s->thread_index);
       break;
-    case SESSION_CTRL_EVT_CLOSE:
-      s = session_get_from_handle_if_valid (e->session_handle);
-      if (PREDICT_FALSE (!s))
-	break;
-      session_transport_close (s);
-      break;
-    case SESSION_CTRL_EVT_RESET:
-      s = session_get_from_handle_if_valid (e->session_handle);
-      if (PREDICT_FALSE (!s))
-	break;
-      session_transport_reset (s);
-      break;
     case SESSION_IO_EVT_BUILTIN_RX:
       s = session_event_get_session (e, thread_index);
       if (PREDICT_FALSE (!s || s->session_state >= SESSION_STATE_CLOSING))
@@ -922,27 +1179,6 @@ session_event_dispatch (session_worker_t * wrk, vlib_node_runtime_t * node,
       if (PREDICT_TRUE (s != 0))
 	session_tx_fifo_dequeue_internal (wrk, node, elt, n_tx_packets);
       break;
-    case SESSION_CTRL_EVT_RPC:
-      fp = e->rpc_args.fp;
-      (*fp) (e->rpc_args.arg);
-      break;
-    case SESSION_CTRL_EVT_DISCONNECTED:
-      session_mq_disconnected_handler (e->data);
-      break;
-    case SESSION_CTRL_EVT_ACCEPTED_REPLY:
-      session_mq_accepted_reply_handler (e->data);
-      break;
-    case SESSION_CTRL_EVT_CONNECTED_REPLY:
-      break;
-    case SESSION_CTRL_EVT_DISCONNECTED_REPLY:
-      session_mq_disconnected_reply_handler (e->data);
-      break;
-    case SESSION_CTRL_EVT_RESET_REPLY:
-      session_mq_reset_reply_handler (e->data);
-      break;
-    case SESSION_CTRL_EVT_WORKER_UPDATE:
-      session_mq_worker_update_handler (e->data);
-      break;
     default:
       clib_warning ("unhandled event type %d", e->event_type);
     }
@@ -953,6 +1189,43 @@ session_event_dispatch (session_worker_t * wrk, vlib_node_runtime_t * node,
     session_evt_elt_free (wrk, elt);
 }
 
+/* *INDENT-OFF* */
+static const u32 session_evt_msg_sizes[] = {
+#define _(symc, sym) 							\
+  [SESSION_CTRL_EVT_ ## symc] = sizeof (session_ ## sym ##_msg_t),
+  foreach_session_ctrl_evt
+#undef _
+};
+/* *INDENT-ON* */
+
+always_inline void
+session_evt_add_to_list (session_worker_t * wrk, session_event_t * evt)
+{
+  session_evt_elt_t *elt;
+
+  if (evt->event_type >= SESSION_CTRL_EVT_RPC)
+    {
+      elt = session_evt_alloc_ctrl (wrk);
+      if (evt->event_type >= SESSION_CTRL_EVT_BOUND)
+	{
+	  elt->evt.ctrl_data_index = session_evt_ctrl_data_alloc (wrk);
+	  elt->evt.event_type = evt->event_type;
+	  clib_memcpy_fast (session_evt_ctrl_data (wrk, elt), evt->data,
+			    session_evt_msg_sizes[evt->event_type]);
+	}
+      else
+	{
+	  /* Internal control events fit into io events footprint */
+	  clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
+	}
+    }
+  else
+    {
+      elt = session_evt_alloc_new (wrk);
+      clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
+    }
+}
+
 static uword
 session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 		       vlib_frame_t * frame)
@@ -990,14 +1263,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 	{
 	  svm_msg_q_sub_w_lock (mq, msg);
 	  evt = svm_msg_q_msg_data (mq, msg);
-	  if (evt->event_type > SESSION_IO_EVT_BUILTIN_TX)
-	    elt = session_evt_alloc_ctrl (wrk);
-	  else
-	    elt = session_evt_alloc_new (wrk);
-	  /* Works because reply messages are smaller than a session evt.
-	   * If we ever need to support bigger messages this needs to be
-	   * fixed */
-	  clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
+	  session_evt_add_to_list (wrk, evt);
 	  svm_msg_q_free_msg (mq, msg);
 	}
       svm_msg_q_unlock (mq);
@@ -1012,7 +1278,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   /* *INDENT-OFF* */
   clib_llist_foreach_safe (wrk->event_elts, evt_list, ctrl_he, elt, ({
     clib_llist_remove (wrk->event_elts, evt_list, elt);
-    session_event_dispatch (wrk, node, elt, thread_index, &n_tx_packets);
+    session_event_dispatch_ctrl (wrk, elt);
   }));
   /* *INDENT-ON* */
 
@@ -1037,7 +1303,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 	continue;
       }
 
-    session_event_dispatch (wrk, node, elt, thread_index, &n_tx_packets);
+    session_event_dispatch_io (wrk, node, elt, thread_index, &n_tx_packets);
   }));
   /* *INDENT-ON* */
 
@@ -1054,7 +1320,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 
       clib_llist_pop_first (wrk->event_elts, evt_list, elt, old_he);
       ei = clib_llist_entry_index (wrk->event_elts, elt);
-      session_event_dispatch (wrk, node, elt, thread_index, &n_tx_packets);
+      session_event_dispatch_io (wrk, node, elt, thread_index, &n_tx_packets);
 
       old_he = pool_elt_at_index (wrk->event_elts, wrk->old_head);
       if (n_tx_packets >= VLIB_FRAME_SIZE || ei == old_ti)