vcl: support for eventfd mq signaling

- support eventfd based mq signaling. Based on configuration, vcl
epoll/select can use either condvars or epoll on mq eventfds.
- add vcl support for memfd segments
- vpp explicitly registers cut-through segments with apps/vcl
- if using eventfd, make ldp allow one call to libc_epoll_create. Needed
for the message queue epfd
- update svm_queue_t to allow blocking calls with eventfd signaling.

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

9 files changed, 249 insertions, 69 deletions

diff --git a/src/vnet/session/application.c b/src/vnet/session/application.c
index 757e12e1b58..3e127df3bd2 100644
--- a/src/vnet/session/application.c
+++ b/src/vnet/session/application.c
@@ -312,6 +312,12 @@ application_init (application_t * app, u32 api_client_index, u8 * app_name,
     }
   else
     {
+      if (options[APP_OPTIONS_FLAGS] & APP_OPTIONS_FLAGS_EVT_MQ_USE_EVENTFD)
+	{
+	  clib_warning ("mq eventfds can only be used if socket transport is "
+			"used for api");
+	  return VNET_API_ERROR_APP_UNSUPPORTED_CFG;
+	}
       seg_type = SSVM_SEGMENT_PRIVATE;
     }
 
@@ -336,6 +342,8 @@ application_init (application_t * app, u32 api_client_index, u8 * app_name,
     props->tx_fifo_size = options[APP_OPTIONS_TX_FIFO_SIZE];
   if (options[APP_OPTIONS_EVT_QUEUE_SIZE])
     props->evt_q_size = options[APP_OPTIONS_EVT_QUEUE_SIZE];
+  if (options[APP_OPTIONS_FLAGS] & APP_OPTIONS_FLAGS_EVT_MQ_USE_EVENTFD)
+    props->use_mq_eventfd = 1;
   if (options[APP_OPTIONS_TLS_ENGINE])
     app->tls_engine = options[APP_OPTIONS_TLS_ENGINE];
   props->segment_type = seg_type;
@@ -970,6 +978,8 @@ application_free_local_session (application_t * app, local_session_t * s)
 local_session_t *
 application_get_local_session (application_t * app, u32 session_index)
 {
+  if (pool_is_free_index (app->local_sessions, session_index))
+    return 0;
   return pool_elt_at_index (app->local_sessions, session_index);
 }
 
@@ -1078,6 +1088,23 @@ application_stop_local_listen (application_t * server, session_handle_t lh)
   return 0;
 }
 
+static void
+application_local_session_fix_eventds (svm_msg_q_t * sq, svm_msg_q_t * cq)
+{
+  int fd;
+
+  /*
+   * segment manager initializes only the producer eventds, since vpp is
+   * typically the producer. But for local sessions, we also pass to the
+   * apps the mqs they listen on for events from peer apps, so they are also
+   * consumer fds.
+   */
+  fd = svm_msg_q_get_producer_eventfd (sq);
+  svm_msg_q_set_consumer_eventfd (sq, fd);
+  fd = svm_msg_q_get_producer_eventfd (cq);
+  svm_msg_q_set_consumer_eventfd (cq, fd);
+}
+
 int
 application_local_session_connect (u32 table_index, application_t * client,
 				   application_t * server,
@@ -1125,8 +1152,12 @@ application_local_session_connect (u32 table_index, application_t * client,
       return seg_index;
     }
   seg = segment_manager_get_segment_w_lock (sm, seg_index);
-  sq = segment_manager_alloc_queue (seg, props->evt_q_size);
-  cq = segment_manager_alloc_queue (seg, cprops->evt_q_size);
+  sq = segment_manager_alloc_queue (seg, props);
+  cq = segment_manager_alloc_queue (seg, cprops);
+
+  if (props->use_mq_eventfd)
+    application_local_session_fix_eventds (sq, cq);
+
   ls->server_evt_q = pointer_to_uword (sq);
   ls->client_evt_q = pointer_to_uword (cq);
   rv = segment_manager_try_alloc_fifos (seg, props->rx_fifo_size,
@@ -1273,7 +1304,7 @@ application_local_session_disconnect (u32 app_index, local_session_t * ls)
 
   if (app_index == ls->client_index)
     {
-      send_local_session_disconnect_callback (ls->app_index, ls);
+      mq_send_local_session_disconnected_cb (ls->app_index, ls);
     }
   else
     {
@@ -1292,7 +1323,7 @@ application_local_session_disconnect (u32 app_index, local_session_t * ls)
 	}
       else
 	{
-	  send_local_session_disconnect_callback (client->index, ls);
+	  mq_send_local_session_disconnected_cb (client->index, ls);
 	}
     }
 
diff --git a/src/vnet/session/application.h b/src/vnet/session/application.h
index f6c81275826..22df21e6140 100644
--- a/src/vnet/session/application.h
+++ b/src/vnet/session/application.h
@@ -267,8 +267,8 @@ application_local_session_listener_has_transport (local_session_t * ls)
   return (tp != TRANSPORT_PROTO_NONE);
 }
 
-void send_local_session_disconnect_callback (u32 app_index,
-					     local_session_t * ls);
+void mq_send_local_session_disconnected_cb (u32 app_index,
+					    local_session_t * ls);
 
 int application_connect (u32 client_index, u32 api_context,
 			 session_endpoint_t * sep);
diff --git a/src/vnet/session/application_interface.c b/src/vnet/session/application_interface.c
index c8fa37f0b9d..524e7b53663 100644
--- a/src/vnet/session/application_interface.c
+++ b/src/vnet/session/application_interface.c
@@ -475,7 +475,7 @@ vnet_application_attach (vnet_app_attach_args_t * a)
 			      a->session_cb_vft)))
     return clib_error_return_code (0, rv, 0, "app init: %d", rv);
 
-  a->app_event_queue_address = pointer_to_uword (app->event_queue);
+  a->app_evt_q = app->event_queue;
   sm = segment_manager_get (app->first_segment_manager);
   fs = segment_manager_get_segment_w_lock (sm, 0);
 
@@ -569,7 +569,18 @@ vnet_disconnect_session (vnet_disconnect_args_t * a)
   if (session_handle_is_local (a->handle))
     {
       local_session_t *ls;
-      ls = application_get_local_session_from_handle (a->handle);
+
+      /* Disconnect reply came to worker 1 not main thread */
+      if (vlib_get_thread_index () == 1)
+	{
+	  vlib_rpc_call_main_thread (vnet_disconnect_session, (u8 *) a,
+				     sizeof (*a));
+	  return 0;
+	}
+
+      if (!(ls = application_get_local_session_from_handle (a->handle)))
+	return 0;
+
       if (ls->app_index != a->app_index && ls->client_index != a->app_index)
 	{
 	  clib_warning ("app %u is neither client nor server for session %u",
diff --git a/src/vnet/session/application_interface.h b/src/vnet/session/application_interface.h
index daba169a297..ba6d2660d7f 100644
--- a/src/vnet/session/application_interface.h
+++ b/src/vnet/session/application_interface.h
@@ -42,7 +42,7 @@ typedef struct _vnet_app_attach_args_t
    * Results
    */
   ssvm_private_t *segment;
-  u64 app_event_queue_address;
+  svm_msg_q_t *app_evt_q;
   u32 app_index;
 } vnet_app_attach_args_t;
 
@@ -138,6 +138,7 @@ typedef enum
   _(USE_GLOBAL_SCOPE, "App can use global session scope")	\
   _(USE_LOCAL_SCOPE, "App can use local session scope")		\
   _(USE_MQ_FOR_CTRL_MSGS, "Use message queue for ctr msgs")	\
+  _(EVT_MQ_USE_EVENTFD, "Use eventfds for signaling")		\
 
 typedef enum _app_options
 {
@@ -153,6 +154,27 @@ typedef enum _app_options_flags
 #undef _
 } app_options_flags_t;
 
+#define foreach_fd_type						\
+  _(VPP_MQ_SEGMENT, "Fd for vpp's event mq segment")		\
+  _(MEMFD_SEGMENT, "Fd for memfd segment")			\
+  _(MQ_EVENTFD, "Event fd used by message queue")		\
+  _(VPP_MQ_EVENTFD, "Event fd used by vpp's message queue")	\
+
+typedef enum session_fd_type_
+{
+#define _(sym, str) SESSION_FD_##sym,
+  foreach_fd_type
+#undef _
+  SESSION_N_FD_TYPE
+} session_fd_type_t;
+
+typedef enum session_fd_flag_
+{
+#define _(sym, str) SESSION_FD_F_##sym = 1 << SESSION_FD_##sym,
+  foreach_fd_type
+#undef _
+} session_fd_flag_t;
+
 int vnet_bind_uri (vnet_bind_args_t *);
 int vnet_unbind_uri (vnet_unbind_args_t * a);
 clib_error_t *vnet_connect_uri (vnet_connect_args_t * a);
diff --git a/src/vnet/session/segment_manager.c b/src/vnet/session/segment_manager.c
index c7d2482d6bf..6f149ad9412 100644
--- a/src/vnet/session/segment_manager.c
+++ b/src/vnet/session/segment_manager.c
@@ -88,7 +88,7 @@ segment_manager_del_segment (segment_manager_t * sm,
 /**
  * Removes segment after acquiring writer lock
  */
-always_inline void
+static inline void
 segment_manager_lock_and_del_segment (segment_manager_t * sm, u32 fs_index)
 {
   svm_fifo_segment_private_t *fs;
@@ -290,8 +290,7 @@ segment_manager_init (segment_manager_t * sm, u32 first_seg_size,
 
 	  segment = segment_manager_get_segment (sm, seg_index);
 	  if (i == 0)
-	    sm->event_queue = segment_manager_alloc_queue (segment,
-							   props->evt_q_size);
+	    sm->event_queue = segment_manager_alloc_queue (segment, props);
 
 	  svm_fifo_segment_preallocate_fifo_pairs (segment,
 						   props->rx_fifo_size,
@@ -311,8 +310,7 @@ segment_manager_init (segment_manager_t * sm, u32 first_seg_size,
 	  return seg_index;
 	}
       segment = segment_manager_get_segment (sm, seg_index);
-      sm->event_queue = segment_manager_alloc_queue (segment,
-						     props->evt_q_size);
+      sm->event_queue = segment_manager_alloc_queue (segment, props);
     }
 
   return 0;
@@ -623,7 +621,7 @@ segment_manager_evt_q_expected_size (u32 q_len)
  */
 svm_msg_q_t *
 segment_manager_alloc_queue (svm_fifo_segment_private_t * segment,
-			     u32 queue_size)
+			     segment_manager_properties_t * props)
 {
   u32 fifo_evt_size, session_evt_size = 256, notif_q_size;
   svm_msg_q_cfg_t _cfg, *cfg = &_cfg;
@@ -631,21 +629,27 @@ segment_manager_alloc_queue (svm_fifo_segment_private_t * segment,
   void *oldheap;
 
   fifo_evt_size = sizeof (session_event_t);
-  notif_q_size = clib_max (16, queue_size >> 4);
+  notif_q_size = clib_max (16, props->evt_q_size >> 4);
   /* *INDENT-OFF* */
   svm_msg_q_ring_cfg_t rc[SESSION_MQ_N_RINGS] = {
-    {queue_size, fifo_evt_size, 0},
+    {props->evt_q_size, fifo_evt_size, 0},
     {notif_q_size, session_evt_size, 0}
   };
   /* *INDENT-ON* */
   cfg->consumer_pid = 0;
   cfg->n_rings = 2;
-  cfg->q_nitems = queue_size;
+  cfg->q_nitems = props->evt_q_size;
   cfg->ring_cfgs = rc;
 
   oldheap = ssvm_push_heap (segment->ssvm.sh);
   q = svm_msg_q_alloc (cfg);
   ssvm_pop_heap (oldheap);
+
+  if (props->use_mq_eventfd)
+    {
+      if (svm_msg_q_alloc_producer_eventfd (q))
+	clib_warning ("failed to alloc eventfd");
+    }
   return q;
 }
 
diff --git a/src/vnet/session/segment_manager.h b/src/vnet/session/segment_manager.h
index 73cb4827a8b..aae4f2e4e51 100644
--- a/src/vnet/session/segment_manager.h
+++ b/src/vnet/session/segment_manager.h
@@ -32,8 +32,10 @@ typedef struct _segment_manager_properties
   /** Configured additional segment size */
   u32 add_segment_size;
 
-  /** Flag that indicates if additional segments should be created */
-  u8 add_segment;
+  /** Flags */
+  u8 add_segment:1;		/**< can add new segments */
+  u8 use_mq_eventfd:1;		/**< use eventfds for mqs */
+  u8 reserved:6;
 
   /** Segment type: if set to SSVM_N_TYPES, private segments are used */
   ssvm_segment_type_t segment_type;
@@ -154,7 +156,8 @@ void segment_manager_dealloc_fifos (u32 segment_index, svm_fifo_t * rx_fifo,
 				    svm_fifo_t * tx_fifo);
 u32 segment_manager_evt_q_expected_size (u32 q_size);
 svm_msg_q_t *segment_manager_alloc_queue (svm_fifo_segment_private_t * fs,
-					  u32 queue_size);
+					  segment_manager_properties_t *
+					  props);
 void segment_manager_dealloc_queue (segment_manager_t * sm, svm_queue_t * q);
 void segment_manager_app_detach (segment_manager_t * sm);
 
diff --git a/src/vnet/session/session.api b/src/vnet/session/session.api
index 98748d8fbbb..58054ae4ba2 100644
--- a/src/vnet/session/session.api
+++ b/src/vnet/session/session.api
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-option version = "1.0.3";
+option version = "1.1.0";
 
 /** \brief client->vpp, attach application to session layer
     @param client_index - opaque cookie to identify the sender
@@ -38,6 +38,9 @@ option version = "1.0.3";
     @param retval - return code for the request
     @param app_event_queue_address - vpp event queue address or 0 if this 
                                  	 connection shouldn't send events
+    @param n_fds - number of fds exchanged
+    @param fd_flags - set of flags that indicate which fds are to be expected
+    				  over the socket (set only if socket transport available) 
     @param segment_size - size of first shm segment
     @param segment_name_length - length of segment name 
     @param segment_name - name of segment client needs to attach to
@@ -46,6 +49,8 @@ define application_attach_reply {
     u32 context;
     i32 retval;
     u64 app_event_queue_address;
+    u8 n_fds;
+    u8 fd_flags;
     u32 segment_size;
     u8 segment_name_length;
     u8 segment_name[128];
@@ -91,11 +96,16 @@ autoreply define application_detach {
 /** \brief vpp->client, please map an additional shared memory segment
     @param client_index - opaque cookie to identify the sender
     @param context - sender context, to match reply w/ request
-    @param segment_name - 
+   	@param fd_flags - set of flags that indicate which, if any, fds are 
+   					  to be expected over the socket. This is set only if 
+   					  socket transport available
+    @param segment_size - size of the segment to be mapped
+    @param segment_name - name of the segment to be mapped
 */
 autoreply define map_another_segment {
     u32 client_index;
     u32 context;
+    u8 fd_flags;
     u32 segment_size;
     u8 segment_name[128];
 };
@@ -391,6 +401,26 @@ define connect_session_reply {
   u16 lcl_port;
 };
 
+/** \brief ask app to add a new cut-through registration
+    @param client_index - opaque cookie to identify the sender
+                          client to vpp direction only
+    @param context - sender context, to match reply w/ request
+    @param evt_q_address - address of the mq in ssvm segment
+    @param peer_evt_q_address - address of peer's mq in ssvm segment
+    @param n_fds - number of fds exchanged
+    @param fd_flags - flag indicating the fds that will be exchanged over
+    				  api socket
+*/
+autoreply define app_cut_through_registration_add
+{
+  u32 client_index;
+  u32 context;
+  u64 evt_q_address;
+  u64 peer_evt_q_address;
+  u8 n_fds;
+  u8 fd_flags;
+};
+
 /** \brief enable/disable session layer
     @param client_index - opaque cookie to identify the sender
                           client to vpp direction only
diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 56f885b2aca..6378fe8e299 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -1245,6 +1245,11 @@ session_vpp_event_queues_allocate (session_manager_main_t * smm)
       cfg->q_nitems = evt_q_length;
       cfg->ring_cfgs = rc;
       smm->vpp_event_queues[i] = svm_msg_q_alloc (cfg);
+      if (smm->evt_qs_use_memfd_seg)
+	{
+	  if (svm_msg_q_alloc_consumer_eventfd (smm->vpp_event_queues[i]))
+	    clib_warning ("eventfd returned");
+	}
     }
 
   if (smm->evt_qs_use_memfd_seg)
diff --git a/src/vnet/session/session_api.c b/src/vnet/session/session_api.c
index 76c10210a50..d1cbc646e52 100755
--- a/src/vnet/session/session_api.c
+++ b/src/vnet/session/session_api.c
@@ -60,16 +60,15 @@ _(APPLICATION_TLS_CERT_ADD, application_tls_cert_add)			\
 _(APPLICATION_TLS_KEY_ADD, application_tls_key_add)			\
 
 static int
-session_send_memfd_fd (vl_api_registration_t * reg, const ssvm_private_t * sp)
+session_send_fds (vl_api_registration_t * reg, int fds[], int n_fds)
 {
   clib_error_t *error;
-  int fd = sp->fd;
   if (vl_api_registration_file_index (reg) == VL_API_INVALID_FI)
     {
       clib_warning ("can't send memfd fd");
       return -1;
     }
-  error = vl_api_send_fd_msg (reg, &fd, 1);
+  error = vl_api_send_fd_msg (reg, fds, n_fds);
   if (error)
     {
       clib_error_report (error);
@@ -81,8 +80,10 @@ session_send_memfd_fd (vl_api_registration_t * reg, const ssvm_private_t * sp)
 static int
 send_add_segment_callback (u32 api_client_index, const ssvm_private_t * sp)
 {
+  int fds[SESSION_N_FD_TYPE], n_fds = 0;
   vl_api_map_another_segment_t *mp;
   vl_api_registration_t *reg;
+  u8 fd_flags = 0;
 
   reg = vl_mem_api_client_index_to_registration (api_client_index);
   if (!reg)
@@ -91,24 +92,31 @@ send_add_segment_callback (u32 api_client_index, const ssvm_private_t * sp)
       return -1;
     }
 
-  if (ssvm_type (sp) == SSVM_SEGMENT_MEMFD
-      && vl_api_registration_file_index (reg) == VL_API_INVALID_FI)
+  if (ssvm_type (sp) == SSVM_SEGMENT_MEMFD)
     {
-      clib_warning ("can't send memfd fd");
-      return -1;
+      if (vl_api_registration_file_index (reg) == VL_API_INVALID_FI)
+	{
+	  clib_warning ("can't send memfd fd");
+	  return -1;
+	}
+
+      fd_flags |= SESSION_FD_F_MEMFD_SEGMENT;
+      fds[n_fds] = sp->fd;
+      n_fds += 1;
     }
 
-  mp = vl_msg_api_alloc_as_if_client (sizeof (*mp));
+  mp = vl_mem_api_alloc_as_if_client_w_reg (reg, sizeof (*mp));
   memset (mp, 0, sizeof (*mp));
   mp->_vl_msg_id = clib_host_to_net_u16 (VL_API_MAP_ANOTHER_SEGMENT);
   mp->segment_size = sp->ssvm_size;
+  mp->fd_flags = fd_flags;
   strncpy ((char *) mp->segment_name, (char *) sp->name,
 	   sizeof (mp->segment_name) - 1);
 
   vl_msg_api_send_shmem (reg->vl_input_queue, (u8 *) & mp);
 
-  if (ssvm_type (sp) == SSVM_SEGMENT_MEMFD)
-    return session_send_memfd_fd (reg, sp);
+  if (n_fds)
+    return session_send_fds (reg, fds, n_fds);
 
   return 0;
 }
@@ -126,23 +134,58 @@ send_del_segment_callback (u32 api_client_index, const ssvm_private_t * fs)
       return -1;
     }
 
-  if (ssvm_type (fs) == SSVM_SEGMENT_MEMFD
-      && vl_api_registration_file_index (reg) == VL_API_INVALID_FI)
+  mp = vl_mem_api_alloc_as_if_client_w_reg (reg, sizeof (*mp));
+  memset (mp, 0, sizeof (*mp));
+  mp->_vl_msg_id = clib_host_to_net_u16 (VL_API_UNMAP_SEGMENT);
+  strcpy ((char *) mp->segment_name, (char *) fs->name);
+
+  vl_msg_api_send_shmem (reg->vl_input_queue, (u8 *) & mp);
+
+  return 0;
+}
+
+static int
+send_app_cut_through_registration_add (u32 api_client_index, u64 mq_addr,
+				       u64 peer_mq_addr)
+{
+  vl_api_app_cut_through_registration_add_t *mp;
+  vl_api_registration_t *reg;
+  svm_msg_q_t *mq, *peer_mq;
+  int fds[2];
+
+  reg = vl_mem_api_client_index_to_registration (api_client_index);
+  if (!reg)
     {
-      clib_warning ("can't send memfd fd");
+      clib_warning ("no registration: %u", api_client_index);
       return -1;
     }
 
-  mp = vl_msg_api_alloc_as_if_client (sizeof (*mp));
+  mp = vl_mem_api_alloc_as_if_client_w_reg (reg, sizeof (*mp));
   memset (mp, 0, sizeof (*mp));
-  mp->_vl_msg_id = clib_host_to_net_u16 (VL_API_UNMAP_SEGMENT);
-  strncpy ((char *) mp->segment_name, (char *) fs->name,
-	   sizeof (mp->segment_name) - 1);
+  mp->_vl_msg_id =
+    clib_host_to_net_u16 (VL_API_APP_CUT_THROUGH_REGISTRATION_ADD);
+
+  mp->evt_q_address = mq_addr;
+  mp->peer_evt_q_address = peer_mq_addr;
+
+  mq = uword_to_pointer (mq_addr, svm_msg_q_t *);
+  peer_mq = uword_to_pointer (peer_mq_addr, svm_msg_q_t *);
+
+  if (svm_msg_q_get_producer_eventfd (mq) != -1)
+    {
+      mp->fd_flags |= SESSION_FD_F_MQ_EVENTFD;
+      mp->n_fds = 2;
+      /* app will overwrite exactly the fds we pass here. So
+       * when we swap mq with peer_mq (accept vs connect) the
+       * fds will still be valid */
+      fds[0] = svm_msg_q_get_consumer_eventfd (mq);
+      fds[1] = svm_msg_q_get_producer_eventfd (peer_mq);
+    }
 
   vl_msg_api_send_shmem (reg->vl_input_queue, (u8 *) & mp);
 
-  if (ssvm_type (fs) == SSVM_SEGMENT_MEMFD)
-    return session_send_memfd_fd (reg, fs);
+  if (mp->n_fds != 0)
+    session_send_fds (reg, fds, mp->n_fds);
 
   return 0;
 }
@@ -230,25 +273,25 @@ send_session_accept_callback (stream_session_t * s)
 }
 
 void
-send_local_session_disconnect_callback (u32 app_index, local_session_t * ls)
+mq_send_local_session_disconnected_cb (u32 app_index, local_session_t * ls)
 {
   application_t *app = application_get (app_index);
-  vl_api_disconnect_session_t *mp;
-  vl_api_registration_t *reg;
-
-  reg = vl_mem_api_client_index_to_registration (app->api_client_index);
-  if (!reg)
-    {
-      clib_warning ("no registration: %u", app->api_client_index);
-      return;
-    }
+  svm_msg_q_msg_t _msg, *msg = &_msg;
+  session_disconnected_msg_t *mp;
+  svm_msg_q_t *app_mq;
+  session_event_t *evt;
 
-  mp = vl_mem_api_alloc_as_if_client_w_reg (reg, sizeof (*mp));
-  memset (mp, 0, sizeof (*mp));
-  mp->_vl_msg_id = clib_host_to_net_u16 (VL_API_DISCONNECT_SESSION);
+  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);
+  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;
-  vl_msg_api_send_shmem (reg->vl_input_queue, (u8 *) & mp);
+  svm_msg_q_add (app_mq, msg, SVM_Q_WAIT);
 }
 
 static void
@@ -414,6 +457,12 @@ mq_send_session_accepted_cb (stream_session_t * s)
     {
       local_session_t *ls = (local_session_t *) s;
       local_session_t *ll;
+      u8 main_thread = vlib_num_workers ()? 1 : 0;
+
+      send_app_cut_through_registration_add (app->api_client_index,
+					     ls->server_evt_q,
+					     ls->client_evt_q);
+
       if (application_local_session_listener_has_transport (ls))
 	{
 	  listener = listen_session_get (ls->listener_index);
@@ -436,7 +485,7 @@ mq_send_session_accepted_cb (stream_session_t * s)
 	}
       mp->handle = application_local_session_handle (ls);
       mp->port = ls->port;
-      vpp_queue = session_manager_get_vpp_event_queue (0);
+      vpp_queue = session_manager_get_vpp_event_queue (main_thread);
       mp->vpp_event_queue_address = pointer_to_uword (vpp_queue);
       mp->client_event_queue_address = ls->client_evt_q;
       mp->server_event_queue_address = ls->server_evt_q;
@@ -542,9 +591,15 @@ mq_send_session_connected_cb (u32 app_index, u32 api_context,
   else
     {
       local_session_t *ls = (local_session_t *) s;
+      u8 main_thread = vlib_num_workers ()? 1 : 0;
+
+      send_app_cut_through_registration_add (app->api_client_index,
+					     ls->client_evt_q,
+					     ls->server_evt_q);
+
       mp->handle = application_local_session_handle (ls);
       mp->lcl_port = ls->port;
-      vpp_mq = session_manager_get_vpp_event_queue (0);
+      vpp_mq = session_manager_get_vpp_event_queue (main_thread);
       mp->vpp_event_queue_address = pointer_to_uword (vpp_mq);
       mp->client_event_queue_address = ls->client_evt_q;
       mp->server_event_queue_address = ls->server_evt_q;
@@ -583,12 +638,13 @@ vl_api_session_enable_disable_t_handler (vl_api_session_enable_disable_t * mp)
 static void
 vl_api_application_attach_t_handler (vl_api_application_attach_t * mp)
 {
+  int rv = 0, fds[SESSION_N_FD_TYPE], n_fds = 0;
   vl_api_application_attach_reply_t *rmp;
   ssvm_private_t *segp, *evt_q_segment;
   vnet_app_attach_args_t _a, *a = &_a;
   vl_api_registration_t *reg;
   clib_error_t *error = 0;
-  int rv = 0;
+  u8 fd_flags = 0;
 
   reg = vl_api_client_index_to_registration (mp->client_index);
   if (!reg)
@@ -629,9 +685,32 @@ vl_api_application_attach_t_handler (vl_api_application_attach_t * mp)
     {
       rv = clib_error_get_code (error);
       clib_error_report (error);
+      vec_free (a->namespace_id);
+      goto done;
     }
   vec_free (a->namespace_id);
 
+  /* Send event queues segment */
+  if ((evt_q_segment = session_manager_get_evt_q_segment ()))
+    {
+      fd_flags |= SESSION_FD_F_VPP_MQ_SEGMENT;
+      fds[n_fds] = evt_q_segment->fd;
+      n_fds += 1;
+    }
+  /* Send fifo segment fd if needed */
+  if (ssvm_type (a->segment) == SSVM_SEGMENT_MEMFD)
+    {
+      fd_flags |= SESSION_FD_F_MEMFD_SEGMENT;
+      fds[n_fds] = a->segment->fd;
+      n_fds += 1;
+    }
+  if (a->options[APP_OPTIONS_FLAGS] & APP_OPTIONS_FLAGS_EVT_MQ_USE_EVENTFD)
+    {
+      fd_flags |= SESSION_FD_F_MQ_EVENTFD;
+      fds[n_fds] = svm_msg_q_get_producer_eventfd (a->app_evt_q);
+      n_fds += 1;
+    }
+
 done:
 
   /* *INDENT-OFF* */
@@ -646,20 +725,15 @@ done:
 	    memcpy (rmp->segment_name, segp->name, vec_len (segp->name));
 	    rmp->segment_name_length = vec_len (segp->name);
 	  }
-	rmp->app_event_queue_address = a->app_event_queue_address;
+	rmp->app_event_queue_address = pointer_to_uword (a->app_evt_q);
+	rmp->n_fds = n_fds;
+	rmp->fd_flags = fd_flags;
       }
   }));
   /* *INDENT-ON* */
 
-  if (rv)
-    return;
-
-  /* Send fifo segment fd if needed */
-  if (ssvm_type (a->segment) == SSVM_SEGMENT_MEMFD)
-    session_send_memfd_fd (reg, a->segment);
-  /* Send event queues segment */
-  if ((evt_q_segment = session_manager_get_evt_q_segment ()))
-    session_send_memfd_fd (reg, evt_q_segment);
+  if (n_fds)
+    session_send_fds (reg, fds, n_fds);
 }
 
 static void