vcl: add support for app socket api
[vpp.git] / src / vcl / vppcom.c
index 35e7804..dcf8ba2 100644 (file)
@@ -202,7 +202,7 @@ vcl_send_session_listen (vcl_worker_t * wrk, vcl_session_t * s)
   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_LISTEN);
   mp = (session_listen_msg_t *) app_evt->evt->data;
   memset (mp, 0, sizeof (*mp));
-  mp->client_index = wrk->my_client_index;
+  mp->client_index = wrk->bapi_client_index;
   mp->context = s->session_index;
   mp->wrk_index = wrk->vpp_wrk_index;
   mp->is_ip4 = s->transport.is_ip4;
@@ -225,7 +225,7 @@ vcl_send_session_connect (vcl_worker_t * wrk, vcl_session_t * s)
   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_CONNECT);
   mp = (session_connect_msg_t *) app_evt->evt->data;
   memset (mp, 0, sizeof (*mp));
-  mp->client_index = wrk->my_client_index;
+  mp->client_index = wrk->bapi_client_index;
   mp->context = s->session_index;
   mp->wrk_index = wrk->vpp_wrk_index;
   mp->is_ip4 = s->transport.is_ip4;
@@ -251,7 +251,7 @@ vcl_send_session_unlisten (vcl_worker_t * wrk, vcl_session_t * s)
   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_UNLISTEN);
   mp = (session_unlisten_msg_t *) app_evt->evt->data;
   memset (mp, 0, sizeof (*mp));
-  mp->client_index = wrk->my_client_index;
+  mp->client_index = wrk->bapi_client_index;
   mp->wrk_index = wrk->vpp_wrk_index;
   mp->handle = s->vpp_handle;
   mp->context = wrk->wrk_index;
@@ -270,7 +270,7 @@ vcl_send_session_disconnect (vcl_worker_t * wrk, vcl_session_t * s)
   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_DISCONNECT);
   mp = (session_disconnect_msg_t *) app_evt->evt->data;
   memset (mp, 0, sizeof (*mp));
-  mp->client_index = wrk->my_client_index;
+  mp->client_index = wrk->bapi_client_index;
   mp->handle = s->vpp_handle;
   app_send_ctrl_evt_to_vpp (mq, app_evt);
 }
@@ -286,7 +286,7 @@ vcl_send_app_detach (vcl_worker_t * wrk)
   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_APP_DETACH);
   mp = (session_app_detach_msg_t *) app_evt->evt->data;
   memset (mp, 0, sizeof (*mp));
-  mp->client_index = wrk->my_client_index;
+  mp->client_index = wrk->bapi_client_index;
   app_send_ctrl_evt_to_vpp (mq, app_evt);
 }
 
@@ -344,13 +344,46 @@ vcl_send_session_worker_update (vcl_worker_t * wrk, vcl_session_t * s,
   mq = vcl_session_vpp_evt_q (wrk, s);
   app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_WORKER_UPDATE);
   mp = (session_worker_update_msg_t *) app_evt->evt->data;
-  mp->client_index = wrk->my_client_index;
+  mp->client_index = wrk->bapi_client_index;
   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);
 }
 
+int
+vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len)
+{
+  app_session_evt_t _app_evt, *app_evt = &_app_evt;
+  session_app_wrk_rpc_msg_t *mp;
+  vcl_worker_t *dst_wrk, *wrk;
+  svm_msg_q_t *mq;
+  int ret = -1;
+
+  if (data_len > sizeof (mp->data))
+    goto done;
+
+  clib_spinlock_lock (&vcm->workers_lock);
+
+  dst_wrk = vcl_worker_get_if_valid (dst_wrk_index);
+  if (!dst_wrk)
+    goto done;
+
+  wrk = vcl_worker_get_current ();
+  mq = vcl_worker_ctrl_mq (wrk);
+  app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_APP_WRK_RPC);
+  mp = (session_app_wrk_rpc_msg_t *) app_evt->evt->data;
+  mp->client_index = wrk->bapi_client_index;
+  mp->wrk_index = dst_wrk->vpp_wrk_index;
+  clib_memcpy (mp->data, data, data_len);
+  app_send_ctrl_evt_to_vpp (mq, app_evt);
+  ret = 0;
+
+done:
+  clib_spinlock_unlock (&vcm->workers_lock);
+  return ret;
+}
+
 static u32
 vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
                              u32 ls_index)
@@ -497,7 +530,13 @@ vcl_session_connected_handler (vcl_worker_t * wrk,
   clib_memcpy_fast (&session->transport.lcl_ip, &mp->lcl.ip,
                    sizeof (session->transport.lcl_ip));
   session->transport.lcl_port = mp->lcl.port;
-  session->session_state = STATE_CONNECT;
+
+  /* Application closed session before connect reply */
+  if (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK)
+      && session->session_state == STATE_CLOSED)
+    vcl_send_session_disconnect (wrk, session);
+  else
+    session->session_state = STATE_CONNECT;
 
   /* Add it to lookup table */
   vcl_session_table_add_vpp_handle (wrk, mp->handle, session_index);
@@ -552,7 +591,8 @@ vcl_session_reset_handler (vcl_worker_t * wrk,
       return VCL_INVALID_SESSION_INDEX;
     }
 
-  session->session_state = STATE_DISCONNECT;
+  if (session->session_state != STATE_CLOSED)
+    session->session_state = STATE_DISCONNECT;
   VDBG (0, "reset session %u [0x%llx]", sid, reset_msg->handle);
   return sid;
 }
@@ -823,6 +863,16 @@ vcl_session_worker_update_reply_handler (vcl_worker_t * wrk, void *data)
        s->vpp_handle, wrk->wrk_index);
 }
 
+static int
+vcl_api_recv_fd (vcl_worker_t * wrk, int *fds, int n_fds)
+{
+
+  if (vcm->cfg.vpp_app_socket_api)
+    return vcl_sapi_recv_fds (wrk, fds, n_fds);
+
+  return vcl_bapi_recv_fds (wrk, fds, n_fds);
+}
+
 static void
 vcl_session_app_add_segment_handler (vcl_worker_t * wrk, void *data)
 {
@@ -835,7 +885,7 @@ vcl_session_app_add_segment_handler (vcl_worker_t * wrk, void *data)
 
   if (msg->fd_flags)
     {
-      vl_socket_client_recv_fd_msg2 (&wrk->bapi_sock_ctx, &fd, 1, 5);
+      vcl_api_recv_fd (wrk, &fd, 1);
       seg_type = SSVM_SEGMENT_MEMFD;
     }
 
@@ -865,6 +915,15 @@ vcl_session_app_del_segment_handler (vcl_worker_t * wrk, void *data)
   VDBG (1, "Unmapped segment: %d", msg->segment_handle);
 }
 
+static void
+vcl_worker_rpc_handler (vcl_worker_t * wrk, void *data)
+{
+  if (!vcm->wrk_rpc_fn)
+    return;
+
+  (vcm->wrk_rpc_fn) (((session_app_wrk_rpc_msg_t *) data)->data);
+}
+
 static int
 vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
 {
@@ -922,6 +981,9 @@ vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
       vcl_session_app_del_segment_handler (wrk, e->data);
       break;
+    case SESSION_CTRL_EVT_APP_WRK_RPC:
+      vcl_worker_rpc_handler (wrk, e->data);
+      break;
     default:
       clib_warning ("unhandled %u", e->event_type);
     }
@@ -1019,42 +1081,6 @@ vcl_flush_mq_events (void)
   vcl_handle_pending_wrk_updates (wrk);
 }
 
-static int
-vppcom_app_session_enable (void)
-{
-  int rv;
-
-  if (vcm->app_state != STATE_APP_ENABLED)
-    {
-      vppcom_send_session_enable_disable (1 /* is_enabled == TRUE */ );
-      rv = vcl_wait_for_app_state_change (STATE_APP_ENABLED);
-      if (PREDICT_FALSE (rv))
-       {
-         VDBG (0, "application session enable timed out! returning %d (%s)",
-               rv, vppcom_retval_str (rv));
-         return rv;
-       }
-    }
-  return VPPCOM_OK;
-}
-
-static int
-vppcom_app_attach (void)
-{
-  int rv;
-
-  vppcom_app_send_attach ();
-  rv = vcl_wait_for_app_state_change (STATE_APP_ATTACHED);
-  if (PREDICT_FALSE (rv))
-    {
-      VDBG (0, "application attach timed out! returning %d (%s)", rv,
-           vppcom_retval_str (rv));
-      return rv;
-    }
-
-  return VPPCOM_OK;
-}
-
 static int
 vppcom_session_unbind (u32 session_handle)
 {
@@ -1119,13 +1145,18 @@ vppcom_session_disconnect (u32 session_handle)
   if (state & STATE_VPP_CLOSING)
     {
       vpp_evt_q = vcl_session_vpp_evt_q (wrk, session);
-      vcl_send_session_disconnected_reply (vpp_evt_q, wrk->my_client_index,
+      vcl_send_session_disconnected_reply (vpp_evt_q, wrk->bapi_client_index,
                                           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);
@@ -1157,11 +1188,31 @@ vppcom_app_exit (void)
   vcl_elog_stop (vcm);
 }
 
+static int
+vcl_api_attach (void)
+{
+  if (vcm->cfg.vpp_app_socket_api)
+    return vcl_sapi_attach ();
+
+  return vcl_bapi_attach ();
+}
+
+static void
+vcl_api_detach (vcl_worker_t * wrk)
+{
+  vcl_send_app_detach (wrk);
+
+  if (vcm->cfg.vpp_app_socket_api)
+    return vcl_sapi_detach (wrk);
+
+  return vcl_bapi_disconnect_from_vpp ();
+}
+
 /*
  * VPPCOM Public API functions
  */
 int
-vppcom_app_create (char *app_name)
+vppcom_app_create (const char *app_name)
 {
   vppcom_cfg_t *vcl_cfg = &vcm->cfg;
   int rv;
@@ -1179,44 +1230,22 @@ vppcom_app_create (char *app_name)
   vcm->main_cpu = pthread_self ();
   vcm->main_pid = getpid ();
   vcm->app_name = format (0, "%s", app_name);
-  vppcom_init_error_string_table ();
   fifo_segment_main_init (&vcm->segment_main, vcl_cfg->segment_baseva,
                          20 /* timeout in secs */ );
   pool_alloc (vcm->workers, vcl_cfg->max_workers);
   clib_spinlock_init (&vcm->workers_lock);
   clib_rwlock_init (&vcm->segment_table_lock);
   atexit (vppcom_app_exit);
+  vcl_elog_init (vcm);
 
   /* Allocate default worker */
   vcl_worker_alloc_and_init ();
 
-  /* API hookup and connect to VPP */
-  vcl_elog_init (vcm);
-  vcm->app_state = STATE_APP_START;
-  rv = vppcom_connect_to_vpp (app_name);
-  if (rv)
-    {
-      VERR ("couldn't connect to VPP!");
-      return rv;
-    }
-  VDBG (0, "sending session enable");
-  rv = vppcom_app_session_enable ();
-  if (rv)
-    {
-      VERR ("vppcom_app_session_enable() failed!");
-      return rv;
-    }
-
-  VDBG (0, "sending app attach");
-  rv = vppcom_app_attach ();
-  if (rv)
-    {
-      VERR ("vppcom_app_attach() failed!");
-      return rv;
-    }
+  if ((rv = vcl_api_attach ()))
+    return rv;
 
   VDBG (0, "app_name '%s', my_client_index %d (0x%x)", app_name,
-       vcm->workers[0].my_client_index, vcm->workers[0].my_client_index);
+       vcm->workers[0].bapi_client_index, vcm->workers[0].bapi_client_index);
 
   return VPPCOM_OK;
 }
@@ -1242,8 +1271,7 @@ vppcom_app_destroy (void)
   }));
   /* *INDENT-ON* */
 
-  vcl_send_app_detach (current_wrk);
-  vppcom_disconnect_from_vpp ();
+  vcl_api_detach (current_wrk);
   vcl_worker_cleanup (current_wrk, 0 /* notify vpp */ );
 
   vcl_elog_stop (vcm);
@@ -1353,7 +1381,7 @@ vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * session,
   else if (state == STATE_DISCONNECT)
     {
       svm_msg_q_t *mq = vcl_session_vpp_evt_q (wrk, session);
-      vcl_send_session_reset_reply (mq, wrk->my_client_index,
+      vcl_send_session_reset_reply (mq, wrk->bapi_client_index,
                                    session->vpp_handle, 0);
     }
   else if (state == STATE_DETACHED)
@@ -1479,52 +1507,6 @@ vppcom_session_listen (uint32_t listen_sh, uint32_t q_len)
   return VPPCOM_OK;
 }
 
-int
-vppcom_session_tls_add_cert (uint32_t session_handle, char *cert,
-                            uint32_t cert_len)
-{
-
-  vcl_worker_t *wrk = vcl_worker_get_current ();
-  vcl_session_t *session = 0;
-
-  session = vcl_session_get_w_handle (wrk, session_handle);
-  if (!session)
-    return VPPCOM_EBADFD;
-
-  if (cert_len == 0 || cert_len == ~0)
-    return VPPCOM_EBADFD;
-
-  /*
-   * Send listen request to vpp and wait for reply
-   */
-  vppcom_send_application_tls_cert_add (session, cert, cert_len);
-  vcm->app_state = STATE_APP_ADDING_TLS_DATA;
-  vcl_wait_for_app_state_change (STATE_APP_READY);
-  return VPPCOM_OK;
-
-}
-
-int
-vppcom_session_tls_add_key (uint32_t session_handle, char *key,
-                           uint32_t key_len)
-{
-
-  vcl_worker_t *wrk = vcl_worker_get_current ();
-  vcl_session_t *session = 0;
-
-  session = vcl_session_get_w_handle (wrk, session_handle);
-  if (!session)
-    return VPPCOM_EBADFD;
-
-  if (key_len == 0 || key_len == ~0)
-    return VPPCOM_EBADFD;
-
-  vppcom_send_application_tls_key_add (session, key, key_len);
-  vcm->app_state = STATE_APP_ADDING_TLS_DATA;
-  vcl_wait_for_app_state_change (STATE_APP_READY);
-  return VPPCOM_OK;
-}
-
 static int
 validate_args_session_accept_ (vcl_worker_t * wrk, vcl_session_t * ls)
 {
@@ -1742,7 +1724,10 @@ vppcom_session_connect (uint32_t session_handle, vppcom_endpt_t * server_ep)
   vcl_send_session_connect (wrk, session);
 
   if (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK))
-    return VPPCOM_EINPROGRESS;
+    {
+      session->session_state = STATE_CONNECT;
+      return VPPCOM_EINPROGRESS;
+    }
 
   /*
    * Wait for reply from vpp if blocking
@@ -1832,7 +1817,7 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
                              u8 peek)
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
-  int n_read = 0, is_nonblocking;
+  int rv, n_read = 0, is_nonblocking;
   vcl_session_t *s = 0;
   svm_fifo_t *rx_fifo;
   svm_msg_q_msg_t msg;
@@ -1889,20 +1874,41 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
        }
     }
 
+read_again:
+
   if (s->is_dgram)
-    n_read = app_recv_dgram_raw (rx_fifo, buf, n, &s->transport, 0, peek);
+    rv = app_recv_dgram_raw (rx_fifo, buf, n, &s->transport, 0, peek);
   else
-    n_read = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
+    rv = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
+
+  ASSERT (rv >= 0);
+  n_read += rv;
 
   if (svm_fifo_is_empty_cons (rx_fifo))
-    svm_fifo_unset_event (s->rx_fifo);
+    {
+      svm_fifo_unset_event (s->rx_fifo);
+      if (!svm_fifo_is_empty_cons (rx_fifo)
+         && svm_fifo_set_event (s->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))
+    {
+      /* More data enqueued while reading. Try to drain it
+       * or fill the buffer */
+      buf += rv;
+      n -= rv;
+      goto read_again;
+    }
 
-  /* Cut-through sessions might request tx notifications on rx fifos */
-  if (PREDICT_FALSE (rx_fifo->want_deq_ntf))
+  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,
                              SESSION_IO_EVT_RX, SVM_Q_WAIT);
-      svm_fifo_reset_has_deq_ntf (s->rx_fifo);
     }
 
   VDBG (2, "session %u[0x%llx]: read %d bytes from (%p)", s->session_index,
@@ -2103,7 +2109,9 @@ vppcom_session_write_inline (vcl_worker_t * wrk, vcl_session_t * s, void *buf,
     app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo->master_session_index,
                            et, SVM_Q_WAIT);
 
-  ASSERT (n_write > 0);
+  /* The underlying fifo segment can run out of memory */
+  if (PREDICT_FALSE (n_write < 0))
+    return VPPCOM_EAGAIN;
 
   VDBG (2, "session %u [0x%llx]: wrote %d bytes", s->session_index,
        s->vpp_handle, n_write);
@@ -2152,7 +2160,7 @@ if (PREDICT_FALSE (svm_fifo_is_empty (_s->rx_fifo)))                      \
       }                                                                        \
     else if (svm_fifo_is_empty (_s->ct_rx_fifo))                       \
       {                                                                        \
-       svm_fifo_unset_event (_s->ct_rx_fifo);                          \
+       svm_fifo_unset_event (_s->rx_fifo); /* rx evts on actual fifo*/ \
        if (svm_fifo_is_empty (_s->ct_rx_fifo))                         \
          break;                                                        \
       }                                                                        \
@@ -2258,6 +2266,9 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
       vcl_session_app_del_segment_handler (wrk, e->data);
       break;
+    case SESSION_CTRL_EVT_APP_WRK_RPC:
+      vcl_worker_rpc_handler (wrk, e->data);
+      break;
     default:
       clib_warning ("unhandled: %u", e->event_type);
       break;
@@ -2564,8 +2575,9 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
   vcl_session_t *vep_session;
-  vcl_session_t *session;
   int rv = VPPCOM_OK;
+  vcl_session_t *s;
+  svm_fifo_t *txf;
 
   if (vep_handle == session_handle)
     {
@@ -2588,13 +2600,13 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
   ASSERT (vep_session->vep.vep_sh == ~0);
   ASSERT (vep_session->vep.prev_sh == ~0);
 
-  session = vcl_session_get_w_handle (wrk, session_handle);
-  if (PREDICT_FALSE (!session))
+  s = vcl_session_get_w_handle (wrk, session_handle);
+  if (PREDICT_FALSE (!s))
     {
       VDBG (0, "Invalid session_handle (%u)!", session_handle);
       return VPPCOM_EBADFD;
     }
-  if (PREDICT_FALSE (session->is_vep))
+  if (PREDICT_FALSE (s->is_vep))
     {
       VDBG (0, "session_handle (%u) is a vep!", vep_handle);
       return VPPCOM_EINVAL;
@@ -2622,39 +2634,38 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
          ASSERT (next_session->vep.prev_sh == vep_handle);
          next_session->vep.prev_sh = session_handle;
        }
-      session->vep.next_sh = vep_session->vep.next_sh;
-      session->vep.prev_sh = vep_handle;
-      session->vep.vep_sh = vep_handle;
-      session->vep.et_mask = VEP_DEFAULT_ET_MASK;
-      session->vep.ev = *event;
-      session->is_vep = 0;
-      session->is_vep_session = 1;
+      s->vep.next_sh = vep_session->vep.next_sh;
+      s->vep.prev_sh = vep_handle;
+      s->vep.vep_sh = vep_handle;
+      s->vep.et_mask = VEP_DEFAULT_ET_MASK;
+      s->vep.ev = *event;
+      s->is_vep = 0;
+      s->is_vep_session = 1;
       vep_session->vep.next_sh = session_handle;
 
-      if (session->tx_fifo)
-       svm_fifo_add_want_deq_ntf (session->tx_fifo,
-                                  SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
+      txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
+      if (txf && (event->events & EPOLLOUT))
+       svm_fifo_add_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
 
       /* Generate EPOLLOUT if tx fifo not full */
-      if ((event->events & EPOLLOUT) &&
-         (vcl_session_write_ready (session) > 0))
+      if ((event->events & EPOLLOUT) && (vcl_session_write_ready (s) > 0))
        {
          session_event_t e = { 0 };
          e.event_type = SESSION_IO_EVT_TX;
-         e.session_index = session->session_index;
+         e.session_index = s->session_index;
          vec_add1 (wrk->unhandled_evts_vector, e);
        }
       /* Generate EPOLLIN if rx fifo has data */
-      if ((event->events & EPOLLIN) && (vcl_session_read_ready (session) > 0))
+      if ((event->events & EPOLLIN) && (vcl_session_read_ready (s) > 0))
        {
          session_event_t e = { 0 };
          e.event_type = SESSION_IO_EVT_RX;
-         e.session_index = session->session_index;
+         e.session_index = s->session_index;
          vec_add1 (wrk->unhandled_evts_vector, e);
        }
       VDBG (1, "EPOLL_CTL_ADD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
            vep_handle, session_handle, event->events, event->data.u64);
-      vcl_evt (VCL_EVT_EPOLL_CTLADD, session, event->events, event->data.u64);
+      vcl_evt (VCL_EVT_EPOLL_CTLADD, s, event->events, event->data.u64);
       break;
 
     case EPOLL_CTL_MOD:
@@ -2664,92 +2675,97 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
          rv = VPPCOM_EINVAL;
          goto done;
        }
-      else if (PREDICT_FALSE (!session->is_vep_session))
+      else if (PREDICT_FALSE (!s->is_vep_session))
        {
          VDBG (0, "sh %u EPOLL_CTL_MOD: not a vep session!", session_handle);
          rv = VPPCOM_EINVAL;
          goto done;
        }
-      else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
+      else if (PREDICT_FALSE (s->vep.vep_sh != vep_handle))
        {
          VDBG (0, "EPOLL_CTL_MOD: sh %u vep_sh (%u) != vep_sh (%u)!",
-               session_handle, session->vep.vep_sh, vep_handle);
+               session_handle, s->vep.vep_sh, vep_handle);
          rv = VPPCOM_EINVAL;
          goto done;
        }
 
       /* Generate EPOLLOUT when tx_fifo/ct_tx_fifo not full */
       if ((event->events & EPOLLOUT) &&
-         !(session->vep.ev.events & EPOLLOUT) &&
-         (vcl_session_write_ready (session) > 0))
+         !(s->vep.ev.events & EPOLLOUT) && (vcl_session_write_ready (s) > 0))
        {
          session_event_t e = { 0 };
          e.event_type = SESSION_IO_EVT_TX;
-         e.session_index = session->session_index;
+         e.session_index = s->session_index;
          vec_add1 (wrk->unhandled_evts_vector, e);
        }
-      session->vep.et_mask = VEP_DEFAULT_ET_MASK;
-      session->vep.ev = *event;
+      s->vep.et_mask = VEP_DEFAULT_ET_MASK;
+      s->vep.ev = *event;
+      txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
+      if (event->events & EPOLLOUT)
+       svm_fifo_add_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
+      else
+       svm_fifo_del_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
       VDBG (1, "EPOLL_CTL_MOD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
            vep_handle, session_handle, event->events, event->data.u64);
       break;
 
     case EPOLL_CTL_DEL:
-      if (PREDICT_FALSE (!session->is_vep_session))
+      if (PREDICT_FALSE (!s->is_vep_session))
        {
          VDBG (0, "EPOLL_CTL_DEL: %u not a vep session!", session_handle);
          rv = VPPCOM_EINVAL;
          goto done;
        }
-      else if (PREDICT_FALSE (session->vep.vep_sh != vep_handle))
+      else if (PREDICT_FALSE (s->vep.vep_sh != vep_handle))
        {
          VDBG (0, "EPOLL_CTL_DEL: sh %u vep_sh (%u) != vep_sh (%u)!",
-               session_handle, session->vep.vep_sh, vep_handle);
+               session_handle, s->vep.vep_sh, vep_handle);
          rv = VPPCOM_EINVAL;
          goto done;
        }
 
-      if (session->vep.prev_sh == vep_handle)
-       vep_session->vep.next_sh = session->vep.next_sh;
+      if (s->vep.prev_sh == vep_handle)
+       vep_session->vep.next_sh = s->vep.next_sh;
       else
        {
          vcl_session_t *prev_session;
-         prev_session = vcl_session_get_w_handle (wrk, session->vep.prev_sh);
+         prev_session = vcl_session_get_w_handle (wrk, s->vep.prev_sh);
          if (PREDICT_FALSE (!prev_session))
            {
              VDBG (0, "EPOLL_CTL_DEL: Invalid prev_sh (%u) on sh (%u)!",
-                   session->vep.prev_sh, session_handle);
+                   s->vep.prev_sh, session_handle);
              return VPPCOM_EBADFD;
            }
          ASSERT (prev_session->vep.next_sh == session_handle);
-         prev_session->vep.next_sh = session->vep.next_sh;
+         prev_session->vep.next_sh = s->vep.next_sh;
        }
-      if (session->vep.next_sh != ~0)
+      if (s->vep.next_sh != ~0)
        {
          vcl_session_t *next_session;
-         next_session = vcl_session_get_w_handle (wrk, session->vep.next_sh);
+         next_session = vcl_session_get_w_handle (wrk, s->vep.next_sh);
          if (PREDICT_FALSE (!next_session))
            {
              VDBG (0, "EPOLL_CTL_DEL: Invalid next_sh (%u) on sh (%u)!",
-                   session->vep.next_sh, session_handle);
+                   s->vep.next_sh, session_handle);
              return VPPCOM_EBADFD;
            }
          ASSERT (next_session->vep.prev_sh == session_handle);
-         next_session->vep.prev_sh = session->vep.prev_sh;
+         next_session->vep.prev_sh = s->vep.prev_sh;
        }
 
-      memset (&session->vep, 0, sizeof (session->vep));
-      session->vep.next_sh = ~0;
-      session->vep.prev_sh = ~0;
-      session->vep.vep_sh = ~0;
-      session->is_vep_session = 0;
+      memset (&s->vep, 0, sizeof (s->vep));
+      s->vep.next_sh = ~0;
+      s->vep.prev_sh = ~0;
+      s->vep.vep_sh = ~0;
+      s->is_vep_session = 0;
 
-      if (session->tx_fifo)
-       svm_fifo_del_want_deq_ntf (session->tx_fifo, SVM_FIFO_NO_DEQ_NOTIF);
+      txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
+      if (txf)
+       svm_fifo_del_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
 
       VDBG (1, "EPOLL_CTL_DEL: vep_idx %u, sh %u!", vep_handle,
            session_handle);
-      vcl_evt (VCL_EVT_EPOLL_CTLDEL, session, vep_sh);
+      vcl_evt (VCL_EVT_EPOLL_CTLDEL, s, vep_sh);
       break;
 
     default:
@@ -2778,7 +2794,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     {
     case SESSION_IO_EVT_RX:
       sid = e->session_index;
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       vcl_fifo_rx_evt_valid_or_break (session);
       session_events = session->vep.ev.events;
@@ -2791,7 +2808,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       break;
     case SESSION_IO_EVT_TX:
       sid = e->session_index;
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       if (!(EPOLLOUT & session_events))
@@ -2799,7 +2817,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       add_event = 1;
       events[*num_ev].events |= EPOLLOUT;
       session_evt_data = session->vep.ev.data.u64;
-      svm_fifo_reset_has_deq_ntf (session->tx_fifo);
+      svm_fifo_reset_has_deq_ntf (vcl_session_is_ct (session) ?
+                                 session->ct_tx_fifo : session->tx_fifo);
       break;
     case SESSION_CTRL_EVT_ACCEPTED:
       session = vcl_session_accepted (wrk,
@@ -2819,7 +2838,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       connected_msg = (session_connected_msg_t *) e->data;
       sid = vcl_session_connected_handler (wrk, connected_msg);
       /* Generate EPOLLOUT because there's no connected event */
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       if (!(EPOLLOUT & session_events))
@@ -2833,7 +2853,7 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     case SESSION_CTRL_EVT_DISCONNECTED:
       disconnected_msg = (session_disconnected_msg_t *) e->data;
       session = vcl_session_disconnected_handler (wrk, disconnected_msg);
-      if (!session)
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       add_event = 1;
@@ -2842,7 +2862,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       break;
     case SESSION_CTRL_EVT_RESET:
       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       add_event = 1;
@@ -2870,6 +2891,9 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
       vcl_session_app_del_segment_handler (wrk, e->data);
       break;
+    case SESSION_CTRL_EVT_APP_WRK_RPC:
+      vcl_worker_rpc_handler (wrk, e->data);
+      break;
     default:
       VDBG (0, "unhandled: %u", e->event_type);
       break;
@@ -2962,7 +2986,7 @@ vppcom_epoll_wait_condvar (vcl_worker_t * wrk, struct epoll_event *events,
        continue;
 
       now = clib_time_now (&wrk->clib_time);
-      wait -= now - start;
+      wait -= (now - start) * 1e3;
       start = now;
     }
   while (wait > 0);
@@ -3076,8 +3100,10 @@ vppcom_session_attr (uint32_t session_handle, uint32_t op,
     case VPPCOM_ATTR_GET_FLAGS:
       if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
        {
-         *flags = O_RDWR | (VCL_SESS_ATTR_TEST (session->attr,
-                                                VCL_SESS_ATTR_NONBLOCK));
+         *flags =
+           O_RDWR |
+           (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK) ?
+            O_NONBLOCK : 0);
          *buflen = sizeof (*flags);
          VDBG (2, "VPPCOM_ATTR_GET_FLAGS: sh %u, flags = 0x%08x, "
                "is_nonblocking = %u", session_handle, *flags,
@@ -3808,22 +3834,9 @@ vppcom_session_worker (vcl_session_handle_t session_handle)
 int
 vppcom_worker_register (void)
 {
-  vcl_worker_t *wrk;
-  u8 *wrk_name = 0;
-  int rv;
-
   if (!vcl_worker_alloc_and_init ())
     return VPPCOM_EEXIST;
 
-  wrk = vcl_worker_get_current ();
-  wrk_name = format (0, "%s-wrk-%u", vcm->app_name, wrk->wrk_index);
-
-  rv = vppcom_connect_to_vpp ((char *) wrk_name);
-  vec_free (wrk_name);
-
-  if (rv)
-    return VPPCOM_EFAULT;
-
   if (vcl_worker_register_with_vpp ())
     return VPPCOM_EEXIST;
 
@@ -3895,6 +3908,95 @@ vppcom_session_n_accepted (uint32_t session_handle)
   return session->n_accepted_sessions;
 }
 
+const char *
+vppcom_proto_str (vppcom_proto_t proto)
+{
+  char const *proto_str;
+
+  switch (proto)
+    {
+    case VPPCOM_PROTO_TCP:
+      proto_str = "TCP";
+      break;
+    case VPPCOM_PROTO_UDP:
+      proto_str = "UDP";
+      break;
+    case VPPCOM_PROTO_TLS:
+      proto_str = "TLS";
+      break;
+    case VPPCOM_PROTO_QUIC:
+      proto_str = "QUIC";
+      break;
+    default:
+      proto_str = "UNKNOWN";
+      break;
+    }
+  return proto_str;
+}
+
+const char *
+vppcom_retval_str (int retval)
+{
+  char const *st;
+
+  switch (retval)
+    {
+    case VPPCOM_OK:
+      st = "VPPCOM_OK";
+      break;
+
+    case VPPCOM_EAGAIN:
+      st = "VPPCOM_EAGAIN";
+      break;
+
+    case VPPCOM_EFAULT:
+      st = "VPPCOM_EFAULT";
+      break;
+
+    case VPPCOM_ENOMEM:
+      st = "VPPCOM_ENOMEM";
+      break;
+
+    case VPPCOM_EINVAL:
+      st = "VPPCOM_EINVAL";
+      break;
+
+    case VPPCOM_EBADFD:
+      st = "VPPCOM_EBADFD";
+      break;
+
+    case VPPCOM_EAFNOSUPPORT:
+      st = "VPPCOM_EAFNOSUPPORT";
+      break;
+
+    case VPPCOM_ECONNABORTED:
+      st = "VPPCOM_ECONNABORTED";
+      break;
+
+    case VPPCOM_ECONNRESET:
+      st = "VPPCOM_ECONNRESET";
+      break;
+
+    case VPPCOM_ENOTCONN:
+      st = "VPPCOM_ENOTCONN";
+      break;
+
+    case VPPCOM_ECONNREFUSED:
+      st = "VPPCOM_ECONNREFUSED";
+      break;
+
+    case VPPCOM_ETIMEDOUT:
+      st = "VPPCOM_ETIMEDOUT";
+      break;
+
+    default:
+      st = "UNKNOWN_STATE";
+      break;
+    }
+
+  return st;
+}
+
 /*
  * fd.io coding-style-patch-verification: ON
  *