svm: allow mq attachments at random offsets
[vpp.git] / src / vcl / vppcom.c
index 229a251..dbb2cd5 100644 (file)
 
 __thread uword __vcl_worker_index = ~0;
 
-static int
-vcl_segment_is_not_mounted (vcl_worker_t * wrk, u64 segment_handle)
-{
-  u32 segment_index;
-
-  if (segment_handle == VCL_INVALID_SEGMENT_HANDLE)
-    return 0;
-
-  segment_index = vcl_segment_table_lookup (segment_handle);
-  if (segment_index != VCL_INVALID_SEGMENT_INDEX)
-    return 0;
-
-  return 1;
-}
-
 static inline int
 vcl_mq_dequeue_batch (vcl_worker_t * wrk, svm_msg_q_t * mq, u32 n_max_msg)
 {
@@ -293,32 +278,33 @@ vcl_send_session_accepted_reply (svm_msg_q_t * mq, u32 context,
 }
 
 static void
-vcl_send_session_disconnected_reply (svm_msg_q_t * mq, u32 context,
-                                    session_handle_t handle, int retval)
+vcl_send_session_disconnected_reply (vcl_worker_t * wrk, vcl_session_t * s,
+                                    int retval)
 {
   app_session_evt_t _app_evt, *app_evt = &_app_evt;
   session_disconnected_reply_msg_t *rmp;
-  app_alloc_ctrl_evt_to_vpp (mq, app_evt,
+  app_alloc_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt,
                             SESSION_CTRL_EVT_DISCONNECTED_REPLY);
   rmp = (session_disconnected_reply_msg_t *) app_evt->evt->data;
-  rmp->handle = handle;
-  rmp->context = context;
+  rmp->handle = s->vpp_handle;
+  rmp->context = wrk->api_client_handle;
   rmp->retval = retval;
-  app_send_ctrl_evt_to_vpp (mq, app_evt);
+  app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 }
 
 static void
-vcl_send_session_reset_reply (svm_msg_q_t * mq, u32 context,
-                             session_handle_t handle, int retval)
+vcl_send_session_reset_reply (vcl_worker_t * wrk, vcl_session_t * s,
+                             int retval)
 {
   app_session_evt_t _app_evt, *app_evt = &_app_evt;
   session_reset_reply_msg_t *rmp;
-  app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_RESET_REPLY);
+  app_alloc_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt,
+                            SESSION_CTRL_EVT_RESET_REPLY);
   rmp = (session_reset_reply_msg_t *) app_evt->evt->data;
-  rmp->handle = handle;
-  rmp->context = context;
+  rmp->handle = s->vpp_handle;
+  rmp->context = wrk->api_client_handle;
   rmp->retval = retval;
-  app_send_ctrl_evt_to_vpp (mq, app_evt);
+  app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 }
 
 void
@@ -327,16 +313,15 @@ vcl_send_session_worker_update (vcl_worker_t * wrk, vcl_session_t * s,
 {
   app_session_evt_t _app_evt, *app_evt = &_app_evt;
   session_worker_update_msg_t *mp;
-  svm_msg_q_t *mq;
 
-  mq = vcl_session_vpp_evt_q (wrk, s);
-  app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_WORKER_UPDATE);
+  app_alloc_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt,
+                            SESSION_CTRL_EVT_WORKER_UPDATE);
   mp = (session_worker_update_msg_t *) app_evt->evt->data;
   mp->client_index = wrk->api_client_handle;
   mp->handle = s->vpp_handle;
   mp->req_wrk_index = wrk->vpp_wrk_index;
   mp->wrk_index = wrk_index;
-  app_send_ctrl_evt_to_vpp (mq, app_evt);
+  app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 }
 
 int
@@ -377,8 +362,6 @@ vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
                              u32 ls_index)
 {
   vcl_session_t *session, *listen_session;
-  svm_fifo_t *rx_fifo, *tx_fifo;
-  u32 vpp_wrk_index;
   svm_msg_q_t *evt_q;
 
   session = vcl_session_alloc (wrk);
@@ -391,30 +374,15 @@ vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
       goto error;
     }
 
-  if (vcl_segment_is_not_mounted (wrk, mp->segment_handle))
+  if (vcl_segment_attach_session (mp->segment_handle, mp->server_rx_fifo,
+                                 mp->server_tx_fifo,
+                                 mp->vpp_event_queue_address, 0, session))
     {
-      VDBG (0, "ERROR: segment for session %u is not mounted!",
-           session->session_index);
+      VDBG (0, "failed to attach fifos for %u", session->session_index);
       goto error;
     }
 
-  rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
-  tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
-  session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-                                        svm_msg_q_t *);
-  rx_fifo->client_session_index = session->session_index;
-  tx_fifo->client_session_index = session->session_index;
-  rx_fifo->client_thread_index = vcl_get_worker_index ();
-  tx_fifo->client_thread_index = vcl_get_worker_index ();
-  vpp_wrk_index = tx_fifo->master_thread_index;
-  vec_validate (wrk->vpp_event_queues, vpp_wrk_index);
-  wrk->vpp_event_queues[vpp_wrk_index] = session->vpp_evt_q;
-
   session->vpp_handle = mp->handle;
-  session->vpp_thread_index = rx_fifo->master_thread_index;
-  session->rx_fifo = rx_fifo;
-  session->tx_fifo = tx_fifo;
-
   session->session_state = VCL_STATE_READY;
   session->transport.rmt_port = mp->rmt.port;
   session->transport.is_ip4 = mp->rmt.is_ip4;
@@ -442,7 +410,8 @@ vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
   return session->session_index;
 
 error:
-  evt_q = uword_to_pointer (mp->vpp_event_queue_address, svm_msg_q_t *);
+  vcl_segment_attach_mq (vcl_vpp_worker_segment_handle (0),
+                        mp->vpp_event_queue_address, mp->mq_index, &evt_q);
   vcl_send_session_accepted_reply (evt_q, mp->context, mp->handle,
                                   VNET_API_ERROR_INVALID_ARGUMENT);
   vcl_session_free (wrk, session);
@@ -453,9 +422,8 @@ static u32
 vcl_session_connected_handler (vcl_worker_t * wrk,
                               session_connected_msg_t * mp)
 {
-  u32 session_index, vpp_wrk_index;
-  svm_fifo_t *rx_fifo, *tx_fifo;
   vcl_session_t *session = 0;
+  u32 session_index;
 
   session_index = mp->context;
   session = vcl_session_get (wrk, session_index);
@@ -475,45 +443,29 @@ vcl_session_connected_handler (vcl_worker_t * wrk,
     }
 
   session->vpp_handle = mp->handle;
-  session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-                                        svm_msg_q_t *);
-  rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
-  tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
-  if (vcl_segment_is_not_mounted (wrk, mp->segment_handle))
+
+  if (vcl_segment_attach_session (mp->segment_handle, mp->server_rx_fifo,
+                                 mp->server_tx_fifo,
+                                 mp->vpp_event_queue_address, 0, session))
     {
-      VDBG (0, "segment for session %u is not mounted!",
-           session->session_index);
+      VDBG (0, "failed to attach fifos for %u", session->session_index);
       session->session_state = VCL_STATE_DETACHED;
       vcl_send_session_disconnect (wrk, session);
       return session_index;
     }
 
-  rx_fifo->client_session_index = session_index;
-  tx_fifo->client_session_index = session_index;
-  rx_fifo->client_thread_index = vcl_get_worker_index ();
-  tx_fifo->client_thread_index = vcl_get_worker_index ();
-
-  vpp_wrk_index = tx_fifo->master_thread_index;
-  vec_validate (wrk->vpp_event_queues, vpp_wrk_index);
-  wrk->vpp_event_queues[vpp_wrk_index] = session->vpp_evt_q;
-
   if (mp->ct_rx_fifo)
     {
-      session->ct_rx_fifo = uword_to_pointer (mp->ct_rx_fifo, svm_fifo_t *);
-      session->ct_tx_fifo = uword_to_pointer (mp->ct_tx_fifo, svm_fifo_t *);
-      if (vcl_segment_is_not_mounted (wrk, mp->ct_segment_handle))
+      if (vcl_segment_attach_session (mp->ct_segment_handle, mp->ct_rx_fifo,
+                                     mp->ct_tx_fifo, (uword) ~0, 1, session))
        {
-         VDBG (0, "ct segment for session %u is not mounted!",
-               session->session_index);
+         VDBG (0, "failed to attach ct fifos for %u", session->session_index);
          session->session_state = VCL_STATE_DETACHED;
          vcl_send_session_disconnect (wrk, session);
          return session_index;
        }
     }
 
-  session->rx_fifo = rx_fifo;
-  session->tx_fifo = tx_fifo;
-  session->vpp_thread_index = rx_fifo->master_thread_index;
   session->transport.is_ip4 = mp->lcl.is_ip4;
   clib_memcpy_fast (&session->transport.lcl_ip, &mp->lcl.ip,
                    sizeof (session->transport.lcl_ip));
@@ -618,20 +570,15 @@ vcl_session_bound_handler (vcl_worker_t * wrk, session_bound_msg_t * mp)
   vcl_session_table_add_listener (wrk, mp->handle, sid);
   session->session_state = VCL_STATE_LISTEN;
 
-  session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *);
-  vec_validate (wrk->vpp_event_queues, 0);
-  wrk->vpp_event_queues[0] = session->vpp_evt_q;
-
   if (vcl_session_is_cl (session))
     {
-      svm_fifo_t *rx_fifo, *tx_fifo;
-      session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *);
-      rx_fifo = uword_to_pointer (mp->rx_fifo, svm_fifo_t *);
-      rx_fifo->client_session_index = sid;
-      tx_fifo = uword_to_pointer (mp->tx_fifo, svm_fifo_t *);
-      tx_fifo->client_session_index = sid;
-      session->rx_fifo = rx_fifo;
-      session->tx_fifo = tx_fifo;
+      if (vcl_segment_attach_session (mp->segment_handle, mp->rx_fifo,
+                                     mp->tx_fifo, mp->vpp_evt_q, 0, session))
+       {
+         VDBG (0, "failed to attach fifos for %u", session->session_index);
+         session->session_state = VCL_STATE_DETACHED;
+         return VCL_INVALID_SESSION_INDEX;
+       }
     }
 
   VDBG (0, "session %u [0x%llx]: listen succeeded!", sid, mp->handle);
@@ -677,6 +624,7 @@ vcl_session_migrated_handler (vcl_worker_t * wrk, void *data)
 {
   session_migrated_msg_t *mp = (session_migrated_msg_t *) data;
   vcl_session_t *s;
+  u32 fs_index;
 
   s = vcl_session_get_w_vpp_handle (wrk, mp->handle);
   if (!s)
@@ -685,23 +633,30 @@ vcl_session_migrated_handler (vcl_worker_t * wrk, void *data)
       return;
     }
 
-  s->vpp_thread_index = mp->vpp_thread_index;
+  fs_index = vcl_segment_table_lookup (mp->segment_handle);
+  if (fs_index == VCL_INVALID_SEGMENT_INDEX)
+    {
+      VDBG (0, "segment for session %u is not mounted!", s->session_index);
+      s->session_state = VCL_STATE_DETACHED;
+      return;
+    }
+
   s->vpp_handle = mp->new_handle;
-  s->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *);
 
-  vec_validate (wrk->vpp_event_queues, s->vpp_thread_index);
-  wrk->vpp_event_queues[s->vpp_thread_index] = s->vpp_evt_q;
+  vcl_segment_attach_mq (vcl_vpp_worker_segment_handle (0), mp->vpp_evt_q,
+                        mp->vpp_thread_index, &s->vpp_evt_q);
 
   vcl_session_table_del_vpp_handle (wrk, mp->handle);
   vcl_session_table_add_vpp_handle (wrk, mp->new_handle, s->session_index);
 
   /* Generate new tx event if we have outstanding data */
   if (svm_fifo_has_event (s->tx_fifo))
-    app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo->master_session_index,
+    app_send_io_evt_to_vpp (s->vpp_evt_q,
+                           s->tx_fifo->shr->master_session_index,
                            SESSION_IO_EVT_TX, SVM_Q_WAIT);
 
   VDBG (0, "Migrated 0x%lx to thread %u 0x%lx", mp->handle,
-       s->vpp_thread_index, mp->new_handle);
+       mp->vpp_thread_index, mp->new_handle);
 }
 
 static vcl_session_t *
@@ -765,6 +720,57 @@ vcl_session_disconnected_handler (vcl_worker_t * wrk,
   return session;
 }
 
+static int
+vppcom_session_disconnect (u32 session_handle)
+{
+  vcl_worker_t *wrk = vcl_worker_get_current ();
+  vcl_session_t *session, *listen_session;
+  vcl_session_state_t state;
+  u64 vpp_handle;
+
+  session = vcl_session_get_w_handle (wrk, session_handle);
+  if (!session)
+    return VPPCOM_EBADFD;
+
+  vpp_handle = session->vpp_handle;
+  state = session->session_state;
+
+  VDBG (1, "session %u [0x%llx] state 0x%x (%s)", session->session_index,
+       vpp_handle, state, vppcom_session_state_str (state));
+
+  if (PREDICT_FALSE (state == VCL_STATE_LISTEN))
+    {
+      VDBG (0, "ERROR: Cannot disconnect a listen socket!");
+      return VPPCOM_EBADFD;
+    }
+
+  if (state == VCL_STATE_VPP_CLOSING)
+    {
+      vcl_send_session_disconnected_reply (wrk, session, 0);
+      VDBG (1, "session %u [0x%llx]: sending disconnect REPLY...",
+           session->session_index, vpp_handle);
+    }
+  else
+    {
+      /* Session doesn't have an event queue yet. Probably a non-blocking
+       * connect. Wait for the reply */
+      if (PREDICT_FALSE (!session->vpp_evt_q))
+       return VPPCOM_OK;
+
+      VDBG (1, "session %u [0x%llx]: sending disconnect...",
+           session->session_index, vpp_handle);
+      vcl_send_session_disconnect (wrk, session);
+    }
+
+  if (session->listener_index != VCL_INVALID_SESSION_INDEX)
+    {
+      listen_session = vcl_session_get (wrk, session->listener_index);
+      listen_session->n_accepted_sessions--;
+    }
+
+  return VPPCOM_OK;
+}
+
 static void
 vcl_session_cleanup_handler (vcl_worker_t * wrk, void *data)
 {
@@ -783,21 +789,24 @@ vcl_session_cleanup_handler (vcl_worker_t * wrk, void *data)
     {
       /* Transport was cleaned up before we confirmed close. Probably the
        * app is still waiting for some data that cannot be delivered.
-       * Confirm close to make sure everything is cleaned up */
-      if (session->session_state == VCL_STATE_VPP_CLOSING
-         || session->session_state == VCL_STATE_DISCONNECT)
-       {
-         vcl_session_cleanup (wrk, session, vcl_session_handle (session),
-                              1 /* do_disconnect */ );
-         /* Move to undetermined state to ensure that the session is not
-          * removed before both vpp and the app cleanup.
-          * - If the app closes first, the session is moved to CLOSED state
-          *   and the session cleanup notification from vpp removes the
-          *   session.
-          * - If vpp cleans up the session first, the session is moved to
-          *   DETACHED state lower and subsequently the close from the app
-          *   frees the session
-          */
+       * Confirm close to make sure everything is cleaned up.
+       * Move to undetermined state to ensure that the session is not
+       * removed before both vpp and the app cleanup.
+       * - If the app closes first, the session is moved to CLOSED state
+       *   and the session cleanup notification from vpp removes the
+       *   session.
+       * - If vpp cleans up the session first, the session is moved to
+       *   DETACHED state lower and subsequently the close from the app
+       *   frees the session
+       */
+      if (session->session_state == VCL_STATE_VPP_CLOSING)
+       {
+         vppcom_session_disconnect (vcl_session_handle (session));
+         session->session_state = VCL_STATE_UPDATED;
+       }
+      else if (session->session_state == VCL_STATE_DISCONNECT)
+       {
+         vcl_send_session_reset_reply (wrk, session, 0);
          session->session_state = VCL_STATE_UPDATED;
        }
       return;
@@ -842,21 +851,15 @@ vcl_session_worker_update_reply_handler (vcl_worker_t * wrk, void *data)
       VDBG (0, "unknown handle 0x%llx", msg->handle);
       return;
     }
-  if (vcl_segment_is_not_mounted (wrk, msg->segment_handle))
-    {
-      clib_warning ("segment for session %u is not mounted!",
-                   s->session_index);
-      return;
-    }
 
   if (s->rx_fifo)
     {
-      s->rx_fifo = uword_to_pointer (msg->rx_fifo, svm_fifo_t *);
-      s->tx_fifo = uword_to_pointer (msg->tx_fifo, svm_fifo_t *);
-      s->rx_fifo->client_session_index = s->session_index;
-      s->tx_fifo->client_session_index = s->session_index;
-      s->rx_fifo->client_thread_index = wrk->wrk_index;
-      s->tx_fifo->client_thread_index = wrk->wrk_index;
+      if (vcl_segment_attach_session (msg->segment_handle, msg->rx_fifo,
+                                     msg->tx_fifo, (uword) ~0, 0, s))
+       {
+         VDBG (0, "failed to attach fifos for %u", s->session_index);
+         return;
+       }
     }
   s->session_state = VCL_STATE_UPDATED;
 
@@ -929,7 +932,11 @@ static int
 vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
 {
   session_disconnected_msg_t *disconnected_msg;
+  session_connected_msg_t *connected_msg;
+  session_reset_msg_t *reset_msg;
+  session_event_t *ecpy;
   vcl_session_t *s;
+  u32 sid;
 
   switch (e->event_type)
     {
@@ -940,27 +947,58 @@ vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
        break;
       vec_add1 (wrk->unhandled_evts_vector, *e);
       break;
+    case SESSION_CTRL_EVT_BOUND:
+      /* We can only wait for only one listen so not postponed */
+      vcl_session_bound_handler (wrk, (session_bound_msg_t *) e->data);
+      break;
     case SESSION_CTRL_EVT_ACCEPTED:
-      vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
+      s = vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
+      if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
+       {
+         vec_add2 (wrk->unhandled_evts_vector, ecpy, 1);
+         *ecpy = *e;
+         ecpy->postponed = 1;
+         ecpy->session_index = s->session_index;
+       }
       break;
     case SESSION_CTRL_EVT_CONNECTED:
-      vcl_session_connected_handler (wrk,
-                                    (session_connected_msg_t *) e->data);
+      connected_msg = (session_connected_msg_t *) e->data;
+      sid = vcl_session_connected_handler (wrk, connected_msg);
+      if (!(s = vcl_session_get (wrk, sid)))
+       break;
+      if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
+       {
+         vec_add2 (wrk->unhandled_evts_vector, ecpy, 1);
+         *ecpy = *e;
+         ecpy->postponed = 1;
+         ecpy->session_index = s->session_index;
+       }
       break;
     case SESSION_CTRL_EVT_DISCONNECTED:
       disconnected_msg = (session_disconnected_msg_t *) e->data;
-      s = vcl_session_disconnected_handler (wrk, disconnected_msg);
-      if (!s)
+      if (!(s = vcl_session_get_w_vpp_handle (wrk, disconnected_msg->handle)))
+       break;
+      if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
+       {
+         vec_add1 (wrk->unhandled_evts_vector, *e);
+         break;
+       }
+      if (!(s = vcl_session_disconnected_handler (wrk, disconnected_msg)))
        break;
       VDBG (0, "disconnected session %u [0x%llx]", s->session_index,
            s->vpp_handle);
       break;
     case SESSION_CTRL_EVT_RESET:
+      reset_msg = (session_reset_msg_t *) e->data;
+      if (!(s = vcl_session_get_w_vpp_handle (wrk, reset_msg->handle)))
+       break;
+      if (vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
+       {
+         vec_add1 (wrk->unhandled_evts_vector, *e);
+         break;
+       }
       vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
       break;
-    case SESSION_CTRL_EVT_BOUND:
-      vcl_session_bound_handler (wrk, (session_bound_msg_t *) e->data);
-      break;
     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
       vcl_session_unlisten_reply_handler (wrk, e->data);
       break;
@@ -1103,7 +1141,7 @@ vppcom_session_unbind (u32 session_handle)
       vcl_session_table_del_vpp_handle (wrk, accepted_msg->handle);
       vcl_send_session_accepted_reply (session->vpp_evt_q,
                                       accepted_msg->context,
-                                      session->vpp_handle, -1);
+                                      accepted_msg->handle, -1);
     }
   clib_fifo_free (session->accept_evts_fifo);
 
@@ -1119,60 +1157,6 @@ vppcom_session_unbind (u32 session_handle)
   return VPPCOM_OK;
 }
 
-static int
-vppcom_session_disconnect (u32 session_handle)
-{
-  vcl_worker_t *wrk = vcl_worker_get_current ();
-  svm_msg_q_t *vpp_evt_q;
-  vcl_session_t *session, *listen_session;
-  vcl_session_state_t state;
-  u64 vpp_handle;
-
-  session = vcl_session_get_w_handle (wrk, session_handle);
-  if (!session)
-    return VPPCOM_EBADFD;
-
-  vpp_handle = session->vpp_handle;
-  state = session->session_state;
-
-  VDBG (1, "session %u [0x%llx] state 0x%x (%s)", session->session_index,
-       vpp_handle, state, vppcom_session_state_str (state));
-
-  if (PREDICT_FALSE (state == VCL_STATE_LISTEN))
-    {
-      VDBG (0, "ERROR: Cannot disconnect a listen socket!");
-      return VPPCOM_EBADFD;
-    }
-
-  if (state == VCL_STATE_VPP_CLOSING)
-    {
-      vpp_evt_q = vcl_session_vpp_evt_q (wrk, session);
-      vcl_send_session_disconnected_reply (vpp_evt_q, wrk->api_client_handle,
-                                          vpp_handle, 0);
-      VDBG (1, "session %u [0x%llx]: sending disconnect REPLY...",
-           session->session_index, vpp_handle);
-    }
-  else
-    {
-      /* Session doesn't have an event queue yet. Probably a non-blocking
-       * connect. Wait for the reply */
-      if (PREDICT_FALSE (!session->vpp_evt_q))
-       return VPPCOM_OK;
-
-      VDBG (1, "session %u [0x%llx]: sending disconnect...",
-           session->session_index, vpp_handle);
-      vcl_send_session_disconnect (wrk, session);
-    }
-
-  if (session->listener_index != VCL_INVALID_SESSION_INDEX)
-    {
-      listen_session = vcl_session_get (wrk, session->listener_index);
-      listen_session->n_accepted_sessions--;
-    }
-
-  return VPPCOM_OK;
-}
-
 /**
  * Handle app exit
  *
@@ -1266,10 +1250,10 @@ vppcom_app_destroy (void)
   current_wrk = vcl_worker_get_current ();
 
   /* *INDENT-OFF* */
-  pool_foreach (wrk, vcm->workers, ({
+  pool_foreach (wrk, vcm->workers)  {
     if (current_wrk != wrk)
       vcl_worker_cleanup (wrk, 0 /* notify vpp */ );
-  }));
+  }
   /* *INDENT-ON* */
 
   vcl_api_detach (current_wrk);
@@ -1331,7 +1315,7 @@ vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * s,
                  s->vep.vep_sh, rv, vppcom_retval_str (rv));
          next_sh = s->vep.next_sh;
        }
-      goto cleanup;
+      goto free_session;
     }
 
   if (s->flags & VCL_SESSION_F_IS_VEP_SESSION)
@@ -1370,9 +1354,7 @@ vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * s,
     }
   else if (s->session_state == VCL_STATE_DISCONNECT)
     {
-      svm_msg_q_t *mq = vcl_session_vpp_evt_q (wrk, s);
-      vcl_send_session_reset_reply (mq, wrk->api_client_handle,
-                                   s->vpp_handle, 0);
+      vcl_send_session_reset_reply (wrk, s, 0);
     }
   else if (s->session_state == VCL_STATE_DETACHED)
     {
@@ -1844,7 +1826,9 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
        {
          if (vcl_session_is_closing (s))
            return vcl_session_closing_error (s);
-         svm_fifo_unset_event (s->rx_fifo);
+         if (is_ct)
+           svm_fifo_unset_event (s->rx_fifo);
+         svm_fifo_unset_event (rx_fifo);
          return VPPCOM_EWOULDBLOCK;
        }
       while (svm_fifo_is_empty_cons (rx_fifo))
@@ -1852,7 +1836,9 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
          if (vcl_session_is_closing (s))
            return vcl_session_closing_error (s);
 
-         svm_fifo_unset_event (s->rx_fifo);
+         if (is_ct)
+           svm_fifo_unset_event (s->rx_fifo);
+         svm_fifo_unset_event (rx_fifo);
          svm_msg_q_lock (mq);
          if (svm_msg_q_is_empty (mq))
            svm_msg_q_wait (mq);
@@ -1878,19 +1864,21 @@ read_again:
 
   if (svm_fifo_is_empty_cons (rx_fifo))
     {
-      svm_fifo_unset_event (s->rx_fifo);
+      if (is_ct)
+       svm_fifo_unset_event (s->rx_fifo);
+      svm_fifo_unset_event (rx_fifo);
       if (!svm_fifo_is_empty_cons (rx_fifo)
-         && svm_fifo_set_event (s->rx_fifo) && is_nonblocking)
+         && svm_fifo_set_event (rx_fifo) && is_nonblocking)
        {
          vec_add2 (wrk->unhandled_evts_vector, e, 1);
          e->event_type = SESSION_IO_EVT_RX;
          e->session_index = s->session_index;
        }
     }
-  else if (PREDICT_FALSE (rv < n))
+  else if (PREDICT_FALSE (rv < n && !s->is_dgram))
     {
       /* More data enqueued while reading. Try to drain it
-       * or fill the buffer */
+       * or fill the buffer. Avoid doing that for dgrams */
       buf += rv;
       n -= rv;
       goto read_again;
@@ -1899,7 +1887,8 @@ read_again:
   if (PREDICT_FALSE (svm_fifo_needs_deq_ntf (rx_fifo, n_read)))
     {
       svm_fifo_clear_deq_ntf (rx_fifo);
-      app_send_io_evt_to_vpp (s->vpp_evt_q, s->rx_fifo->master_session_index,
+      app_send_io_evt_to_vpp (s->vpp_evt_q,
+                             s->rx_fifo->shr->master_session_index,
                              SESSION_IO_EVT_RX, SVM_Q_WAIT);
     }
 
@@ -1945,16 +1934,15 @@ vppcom_session_read_segments (uint32_t session_handle,
   is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
   is_ct = vcl_session_is_ct (s);
   mq = wrk->app_event_queue;
-  rx_fifo = s->rx_fifo;
+  rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
   s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
 
-  if (is_ct)
-    svm_fifo_unset_event (s->rx_fifo);
-
   if (svm_fifo_is_empty_cons (rx_fifo))
     {
       if (is_nonblocking)
        {
+         if (is_ct)
+           svm_fifo_unset_event (s->rx_fifo);
          svm_fifo_unset_event (rx_fifo);
          return VPPCOM_EWOULDBLOCK;
        }
@@ -1963,6 +1951,8 @@ vppcom_session_read_segments (uint32_t session_handle,
          if (vcl_session_is_closing (s))
            return vcl_session_closing_error (s);
 
+         if (is_ct)
+           svm_fifo_unset_event (s->rx_fifo);
          svm_fifo_unset_event (rx_fifo);
          svm_msg_q_lock (mq);
          if (svm_msg_q_is_empty (mq))
@@ -1984,9 +1974,11 @@ vppcom_session_read_segments (uint32_t session_handle,
 
   if (svm_fifo_max_dequeue_cons (rx_fifo) == n_read)
     {
-      svm_fifo_unset_event (s->rx_fifo);
+      if (is_ct)
+       svm_fifo_unset_event (s->rx_fifo);
+      svm_fifo_unset_event (rx_fifo);
       if (svm_fifo_max_dequeue_cons (rx_fifo) != n_read
-         && svm_fifo_set_event (s->rx_fifo)
+         && svm_fifo_set_event (rx_fifo)
          && vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK))
        {
          session_event_t *e;
@@ -2005,12 +1997,14 @@ vppcom_session_free_segments (uint32_t session_handle, uint32_t n_bytes)
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
   vcl_session_t *s;
+  u8 is_ct;
 
   s = vcl_session_get_w_handle (wrk, session_handle);
   if (PREDICT_FALSE (!s || (s->flags & VCL_SESSION_F_IS_VEP)))
     return;
 
-  svm_fifo_dequeue_drop (s->rx_fifo, n_bytes);
+  is_ct = vcl_session_is_ct (s);
+  svm_fifo_dequeue_drop (is_ct ? s->ct_rx_fifo : s->rx_fifo, n_bytes);
 
   ASSERT (s->rx_bytes_pending < n_bytes);
   s->rx_bytes_pending -= n_bytes;
@@ -2105,8 +2099,8 @@ vppcom_session_write_inline (vcl_worker_t * wrk, vcl_session_t * s, void *buf,
                                   0 /* do_evt */ , SVM_Q_WAIT);
 
   if (svm_fifo_set_event (s->tx_fifo))
-    app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo->master_session_index,
-                           et, SVM_Q_WAIT);
+    app_send_io_evt_to_vpp (
+      s->vpp_evt_q, s->tx_fifo->shr->master_session_index, et, SVM_Q_WAIT);
 
   /* The underlying fifo segment can run out of memory */
   if (PREDICT_FALSE (n_write < 0))
@@ -2173,17 +2167,17 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
 {
   session_disconnected_msg_t *disconnected_msg;
   session_connected_msg_t *connected_msg;
-  vcl_session_t *session;
+  vcl_session_t *s;
   u32 sid;
 
   switch (e->event_type)
     {
     case SESSION_IO_EVT_RX:
       sid = e->session_index;
-      session = vcl_session_get (wrk, sid);
-      if (!session || !vcl_session_is_open (session))
+      s = vcl_session_get (wrk, sid);
+      if (!s || !vcl_session_is_open (s))
        break;
-      vcl_fifo_rx_evt_valid_or_break (session);
+      vcl_fifo_rx_evt_valid_or_break (s);
       if (sid < n_bits && read_map)
        {
          clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
@@ -2192,8 +2186,8 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       break;
     case SESSION_IO_EVT_TX:
       sid = e->session_index;
-      session = vcl_session_get (wrk, sid);
-      if (!session || !vcl_session_is_open (session))
+      s = vcl_session_get (wrk, sid);
+      if (!s || !vcl_session_is_open (s))
        break;
       if (sid < n_bits && write_map)
        {
@@ -2202,11 +2196,13 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
        }
       break;
     case SESSION_CTRL_EVT_ACCEPTED:
-      session = vcl_session_accepted (wrk,
-                                     (session_accepted_msg_t *) e->data);
-      if (!session)
+      if (!e->postponed)
+       s = vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
+      else
+       s = vcl_session_get (wrk, e->session_index);
+      if (!s)
        break;
-      sid = session->session_index;
+      sid = s->session_index;
       if (sid < n_bits && read_map)
        {
          clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
@@ -2214,8 +2210,13 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
        }
       break;
     case SESSION_CTRL_EVT_CONNECTED:
-      connected_msg = (session_connected_msg_t *) e->data;
-      sid = vcl_session_connected_handler (wrk, connected_msg);
+      if (!e->postponed)
+       {
+         connected_msg = (session_connected_msg_t *) e->data;
+         sid = vcl_session_connected_handler (wrk, connected_msg);
+       }
+      else
+       sid = e->session_index;
       if (sid == VCL_INVALID_SESSION_INDEX)
        break;
       if (sid < n_bits && write_map)
@@ -2226,10 +2227,10 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       break;
     case SESSION_CTRL_EVT_DISCONNECTED:
       disconnected_msg = (session_disconnected_msg_t *) e->data;
-      session = vcl_session_disconnected_handler (wrk, disconnected_msg);
-      if (!session)
+      s = vcl_session_disconnected_handler (wrk, disconnected_msg);
+      if (!s)
        break;
-      sid = session->session_index;
+      sid = s->session_index;
       if (sid < n_bits && except_map)
        {
          clib_bitmap_set_no_check ((uword *) except_map, sid, 1);
@@ -2420,7 +2421,7 @@ vppcom_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
     goto check_rd;
 
   /* *INDENT-OFF* */
-  clib_bitmap_foreach (sid, wrk->wr_bitmap, ({
+  clib_bitmap_foreach (sid, wrk->wr_bitmap)  {
     if (!(session = vcl_session_get (wrk, sid)))
       {
        clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
@@ -2435,13 +2436,13 @@ vppcom_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
       }
     else
       svm_fifo_add_want_deq_ntf (session->tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
-  }));
+  }
 
 check_rd:
   if (!read_map)
     goto check_mq;
 
-  clib_bitmap_foreach (sid, wrk->rd_bitmap, ({
+  clib_bitmap_foreach (sid, wrk->rd_bitmap)  {
     if (!(session = vcl_session_get (wrk, sid)))
       {
        clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
@@ -2454,7 +2455,7 @@ check_rd:
         clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
         bits_set++;
       }
-  }));
+  }
   /* *INDENT-ON* */
 
 check_mq:
@@ -2786,89 +2787,96 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
   session_connected_msg_t *connected_msg;
   u32 sid = ~0, session_events;
   u64 session_evt_data = ~0;
-  vcl_session_t *session;
+  vcl_session_t *s;
   u8 add_event = 0;
 
   switch (e->event_type)
     {
     case SESSION_IO_EVT_RX:
       sid = e->session_index;
-      session = vcl_session_get (wrk, sid);
-      if (vcl_session_is_closed (session))
+      s = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (s))
        break;
-      vcl_fifo_rx_evt_valid_or_break (session);
-      session_events = session->vep.ev.events;
-      if (!(EPOLLIN & session->vep.ev.events)
-         || (session->flags & VCL_SESSION_F_HAS_RX_EVT))
+      vcl_fifo_rx_evt_valid_or_break (s);
+      session_events = s->vep.ev.events;
+      if (!(EPOLLIN & s->vep.ev.events)
+         || (s->flags & VCL_SESSION_F_HAS_RX_EVT))
        break;
       add_event = 1;
       events[*num_ev].events |= EPOLLIN;
-      session_evt_data = session->vep.ev.data.u64;
-      session->flags |= VCL_SESSION_F_HAS_RX_EVT;
+      session_evt_data = s->vep.ev.data.u64;
+      s->flags |= VCL_SESSION_F_HAS_RX_EVT;
       break;
     case SESSION_IO_EVT_TX:
       sid = e->session_index;
-      session = vcl_session_get (wrk, sid);
-      if (vcl_session_is_closed (session))
+      s = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (s))
        break;
-      session_events = session->vep.ev.events;
+      session_events = s->vep.ev.events;
       if (!(EPOLLOUT & session_events))
        break;
       add_event = 1;
       events[*num_ev].events |= EPOLLOUT;
-      session_evt_data = session->vep.ev.data.u64;
-      svm_fifo_reset_has_deq_ntf (vcl_session_is_ct (session) ?
-                                 session->ct_tx_fifo : session->tx_fifo);
+      session_evt_data = s->vep.ev.data.u64;
+      svm_fifo_reset_has_deq_ntf (vcl_session_is_ct (s) ?
+                                 s->ct_tx_fifo : s->tx_fifo);
       break;
     case SESSION_CTRL_EVT_ACCEPTED:
-      session = vcl_session_accepted (wrk,
-                                     (session_accepted_msg_t *) e->data);
-      if (!session)
+      if (!e->postponed)
+       s = vcl_session_accepted (wrk, (session_accepted_msg_t *) e->data);
+      else
+       s = vcl_session_get (wrk, e->session_index);
+      if (!s)
        break;
-
-      session_events = session->vep.ev.events;
+      session_events = s->vep.ev.events;
+      sid = s->session_index;
       if (!(EPOLLIN & session_events))
        break;
-
       add_event = 1;
       events[*num_ev].events |= EPOLLIN;
-      session_evt_data = session->vep.ev.data.u64;
+      session_evt_data = s->vep.ev.data.u64;
       break;
     case SESSION_CTRL_EVT_CONNECTED:
-      connected_msg = (session_connected_msg_t *) e->data;
-      sid = vcl_session_connected_handler (wrk, connected_msg);
-      /* Generate EPOLLOUT because there's no connected event */
-      session = vcl_session_get (wrk, sid);
-      if (vcl_session_is_closed (session))
+      if (!e->postponed)
+       {
+         connected_msg = (session_connected_msg_t *) e->data;
+         sid = vcl_session_connected_handler (wrk, connected_msg);
+       }
+      else
+       sid = e->session_index;
+      s = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (s))
        break;
-      session_events = session->vep.ev.events;
+      session_events = s->vep.ev.events;
+      /* Generate EPOLLOUT because there's no connected event */
       if (!(EPOLLOUT & session_events))
        break;
       add_event = 1;
       events[*num_ev].events |= EPOLLOUT;
-      session_evt_data = session->vep.ev.data.u64;
-      if (session->session_state == VCL_STATE_DETACHED)
+      session_evt_data = s->vep.ev.data.u64;
+      if (s->session_state == VCL_STATE_DETACHED)
        events[*num_ev].events |= EPOLLHUP;
       break;
     case SESSION_CTRL_EVT_DISCONNECTED:
       disconnected_msg = (session_disconnected_msg_t *) e->data;
-      session = vcl_session_disconnected_handler (wrk, disconnected_msg);
-      if (vcl_session_is_closed (session))
+      s = vcl_session_disconnected_handler (wrk, disconnected_msg);
+      if (vcl_session_is_closed (s))
        break;
-      session_events = session->vep.ev.events;
+      sid = s->session_index;
+      session_events = s->vep.ev.events;
       add_event = 1;
       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
-      session_evt_data = session->vep.ev.data.u64;
+      session_evt_data = s->vep.ev.data.u64;
       break;
     case SESSION_CTRL_EVT_RESET:
       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
-      session = vcl_session_get (wrk, sid);
-      if (vcl_session_is_closed (session))
+      s = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (s))
        break;
-      session_events = session->vep.ev.events;
+      session_events = s->vep.ev.events;
       add_event = 1;
       events[*num_ev].events |= EPOLLHUP | EPOLLRDHUP;
-      session_evt_data = session->vep.ev.data.u64;
+      session_evt_data = s->vep.ev.data.u64;
       break;
     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
       vcl_session_unlisten_reply_handler (wrk, e->data);
@@ -2904,8 +2912,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       events[*num_ev].data.u64 = session_evt_data;
       if (EPOLLONESHOT & session_events)
        {
-         session = vcl_session_get (wrk, sid);
-         session->vep.ev.events = 0;
+         s = vcl_session_get (wrk, sid);
+         s->vep.ev.events = 0;
        }
       *num_ev += 1;
     }