session: refactor local connects

- Switches local connects to cut-thru transport
- Removes local sessions as a separate session type

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

1 files changed, 271 insertions, 338 deletions

diff --git a/src/vnet/session/application_local.c b/src/vnet/session/application_local.c
index 9e874ce536e..b5cac56130c 100644
--- a/src/vnet/session/application_local.c
+++ b/src/vnet/session/application_local.c
@@ -16,15 +16,6 @@
 #include <vnet/session/application_local.h>
 #include <vnet/session/session.h>
 
-typedef struct ct_connection_
-{
-  transport_connection_t connection;
-  u32 client_wrk;
-  u32 server_wrk;
-  u32 transport_listener_index;
-  transport_proto_t actual_tp;
-} ct_connection_t;
-
 ct_connection_t *connections;
 
 ct_connection_t *
@@ -56,9 +47,17 @@ ct_connection_free (ct_connection_t * ct)
   pool_put (connections, ct);
 }
 
+session_t *
+ct_session_get_peer (session_t * s)
+{
+  ct_connection_t *ct, *peer_ct;
+  ct = ct_connection_get (s->connection_index);
+  peer_ct = ct_connection_get (ct->peer_index);
+  return session_get (peer_ct->c_s_index, 0);
+}
+
 void
-application_local_listener_session_endpoint (session_t * ll,
-					     session_endpoint_t * sep)
+ct_session_endpoint (session_t * ll, session_endpoint_t * sep)
 {
   ct_connection_t *ct;
   ct = (ct_connection_t *) session_get_transport (ll);
@@ -67,184 +66,32 @@ application_local_listener_session_endpoint (session_t * ll,
   sep->is_ip4 = ct->c_is_ip4;
 }
 
-local_session_t *
-app_worker_local_session_alloc (app_worker_t * app_wrk)
-{
-  local_session_t *s;
-  pool_get (app_wrk->local_sessions, s);
-  clib_memset (s, 0, sizeof (*s));
-  s->app_wrk_index = app_wrk->wrk_index;
-  s->session_index = s - app_wrk->local_sessions;
-  s->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE, 0);
-  return s;
-}
-
-void
-app_worker_local_session_free (app_worker_t * app_wrk, local_session_t * s)
-{
-  pool_put (app_wrk->local_sessions, s);
-  if (CLIB_DEBUG)
-    clib_memset (s, 0xfc, sizeof (*s));
-}
-
-local_session_t *
-app_worker_get_local_session (app_worker_t * app_wrk, u32 session_index)
-{
-  if (pool_is_free_index (app_wrk->local_sessions, session_index))
-    return 0;
-  return pool_elt_at_index (app_wrk->local_sessions, session_index);
-}
-
-local_session_t *
-app_worker_get_local_session_from_handle (session_handle_t handle)
-{
-  app_worker_t *server_wrk;
-  u32 session_index, server_wrk_index;
-  local_session_parse_handle (handle, &server_wrk_index, &session_index);
-  server_wrk = app_worker_get_if_valid (server_wrk_index);
-  if (!server_wrk)
-    return 0;
-  return app_worker_get_local_session (server_wrk, session_index);
-}
-
-static inline u64
-application_client_local_connect_key (local_session_t * ls)
-{
-  return (((u64) ls->app_wrk_index) << 32 | (u64) ls->session_index);
-}
-
-static inline void
-application_client_local_connect_key_parse (u64 key, u32 * app_wrk_index,
-					    u32 * session_index)
-{
-  *app_wrk_index = key >> 32;
-  *session_index = key & 0xFFFFFFFF;
-}
-
-void
-app_worker_local_sessions_free (app_worker_t * app_wrk)
-{
-  u32 index, server_wrk_index, session_index;
-  u64 handle, *handles = 0;
-  app_worker_t *server_wrk;
-  segment_manager_t *sm;
-  local_session_t *ls;
-  int i;
-
-  /*
-   * Local sessions
-   */
-  if (app_wrk->local_sessions)
-    {
-      /* *INDENT-OFF* */
-      pool_foreach (ls, app_wrk->local_sessions, ({
-	app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
-      }));
-      /* *INDENT-ON* */
-    }
-
-  /*
-   * Local connects
-   */
-  vec_reset_length (handles);
-  /* *INDENT-OFF* */
-  hash_foreach (handle, index, app_wrk->local_connects, ({
-    vec_add1 (handles, handle);
-  }));
-  /* *INDENT-ON* */
-
-  for (i = 0; i < vec_len (handles); i++)
-    {
-      application_client_local_connect_key_parse (handles[i],
-						  &server_wrk_index,
-						  &session_index);
-      server_wrk = app_worker_get_if_valid (server_wrk_index);
-      if (server_wrk)
-	{
-	  ls = app_worker_get_local_session (server_wrk, session_index);
-	  app_worker_local_session_disconnect (app_wrk->wrk_index, ls);
-	}
-    }
-
-  sm = segment_manager_get (app_wrk->local_segment_manager);
-  sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
-  segment_manager_del (sm);
-}
-
 int
-app_worker_local_session_cleanup (app_worker_t * client_wrk,
-				  app_worker_t * server_wrk,
-				  local_session_t * ls)
+ct_session_connect_notify (session_t * ss)
 {
   svm_fifo_segment_private_t *seg;
-  session_t *listener;
-  segment_manager_t *sm;
-  u64 client_key;
-  u8 has_transport;
-
-  /* Retrieve listener transport type as it is the one that decides where
-   * the fifos are allocated */
-  has_transport = application_local_session_listener_has_transport (ls);
-  if (!has_transport)
-    sm = app_worker_get_local_segment_manager_w_session (server_wrk, ls);
-  else
-    {
-      listener = listen_session_get (ls->listener_index);
-      sm = app_worker_get_listen_segment_manager (server_wrk, listener);
-    }
-
-  seg = segment_manager_get_segment (sm, ls->svm_segment_index);
-  if (client_wrk)
-    {
-      client_key = application_client_local_connect_key (ls);
-      hash_unset (client_wrk->local_connects, client_key);
-    }
-
-  if (!has_transport)
-    {
-      application_t *server = application_get (server_wrk->app_index);
-      u64 segment_handle = segment_manager_segment_handle (sm, seg);
-      server->cb_fns.del_segment_callback (server_wrk->api_client_index,
-					   segment_handle);
-      if (client_wrk)
-	{
-	  application_t *client = application_get (client_wrk->app_index);
-	  client->cb_fns.del_segment_callback (client_wrk->api_client_index,
-					       segment_handle);
-	}
-      segment_manager_del_segment (sm, seg);
-    }
-
-  app_worker_local_session_free (server_wrk, ls);
-
-  return 0;
-}
-
-int
-app_worker_local_session_connect_notify (local_session_t * ls)
-{
-  svm_fifo_segment_private_t *seg;
-  app_worker_t *client_wrk, *server_wrk;
+  ct_connection_t *sct, *cct;
+  app_worker_t *client_wrk;
   segment_manager_t *sm;
-  application_t *client;
-  int rv, is_fail = 0;
   u64 segment_handle;
-  u64 client_key;
+  int is_fail = 0;
+  session_t *cs;
+  u32 ss_index;
 
-  client_wrk = app_worker_get (ls->client_wrk_index);
-  server_wrk = app_worker_get (ls->app_wrk_index);
-  client = application_get (client_wrk->app_index);
+  ss_index = ss->session_index;
+  sct = (ct_connection_t *) session_get_transport (ss);
+  client_wrk = app_worker_get (sct->client_wrk);
 
-  sm = app_worker_get_local_segment_manager_w_session (server_wrk, ls);
-  seg = segment_manager_get_segment_w_lock (sm, ls->svm_segment_index);
+  sm = segment_manager_get (ss->rx_fifo->segment_manager);
+  seg = segment_manager_get_segment_w_lock (sm, ss->rx_fifo->segment_index);
   segment_handle = segment_manager_segment_handle (sm, seg);
-  if ((rv = client->cb_fns.add_segment_callback (client_wrk->api_client_index,
-						 segment_handle)))
+
+  if (app_worker_add_segment_notify (client_wrk, segment_handle))
     {
       clib_warning ("failed to notify client %u of new segment",
-		    ls->client_wrk_index);
+		    sct->client_wrk);
       segment_manager_segment_reader_unlock (sm);
-      app_worker_local_session_disconnect (ls->client_wrk_index, ls);
+      session_close (ss);
       is_fail = 1;
     }
   else
@@ -252,17 +99,44 @@ app_worker_local_session_connect_notify (local_session_t * ls)
       segment_manager_segment_reader_unlock (sm);
     }
 
-  client->cb_fns.session_connected_callback (client_wrk->wrk_index,
-					     ls->client_opaque,
-					     (session_t *) ls, is_fail);
+  /* Alloc client session */
+  cct = ct_connection_get (sct->peer_index);
+
+  cs = session_alloc (0);
+  ss = session_get (ss_index, 0);
+  cs->session_type = ss->session_type;
+  cs->connection_index = sct->c_c_index;
+  cs->listener_index = SESSION_INVALID_INDEX;
+  cs->session_state = SESSION_STATE_CONNECTING;
+  cs->app_wrk_index = client_wrk->wrk_index;
+  cs->connection_index = cct->c_c_index;
+
+  cct->c_s_index = cs->session_index;
+
+  /* This will allocate fifos for the session. They won't be used for
+   * exchanging data but they will be used to close the connection if
+   * the segment manager/worker is freed */
+  if (app_worker_init_connected (client_wrk, cs))
+    {
+      session_close (ss);
+      return -1;
+    }
+
+  if (app_worker_connect_notify (client_wrk, is_fail ? 0 : cs,
+				 sct->client_opaque))
+    {
+      session_close (ss);
+      return -1;
+    }
+
+  cs = session_get (cct->c_s_index, 0);
+  cs->session_state = SESSION_STATE_READY;
 
-  client_key = application_client_local_connect_key (ls);
-  hash_set (client_wrk->local_connects, client_key, client_key);
   return 0;
 }
 
 static void
-application_local_session_fix_eventds (svm_msg_q_t * sq, svm_msg_q_t * cq)
+ct_session_fix_eventds (svm_msg_q_t * sq, svm_msg_q_t * cq)
 {
   int fd;
 
@@ -279,22 +153,19 @@ application_local_session_fix_eventds (svm_msg_q_t * sq, svm_msg_q_t * cq)
 }
 
 int
-app_worker_local_session_connect (app_worker_t * client_wrk,
-				  app_worker_t * server_wrk,
-				  session_t * ll, u32 opaque)
+ct_init_local_session (app_worker_t * client_wrk, app_worker_t * server_wrk,
+		       ct_connection_t * ct, session_t * ls, session_t * ll)
 {
   u32 seg_size, evt_q_sz, evt_q_elts, margin = 16 << 10;
   u32 round_rx_fifo_sz, round_tx_fifo_sz, sm_index;
   segment_manager_properties_t *props, *cprops;
-  int rv, has_transport, seg_index;
   svm_fifo_segment_private_t *seg;
   application_t *server, *client;
   segment_manager_t *sm;
-  local_session_t *ls;
   svm_msg_q_t *sq, *cq;
   u64 segment_handle;
+  int seg_index, rv;
 
-  ls = app_worker_local_session_alloc (server_wrk);
   server = application_get (server_wrk->app_index);
   client = application_get (client_wrk->app_index);
 
@@ -306,24 +177,7 @@ app_worker_local_session_connect (app_worker_t * client_wrk,
   round_tx_fifo_sz = 1 << max_log2 (props->tx_fifo_size);
   seg_size = round_rx_fifo_sz + round_tx_fifo_sz + evt_q_sz + margin;
 
-  has_transport = session_has_transport (ll);
-  if (!has_transport)
-    {
-      /* Local sessions don't have backing transport */
-      transport_connection_t *tc;
-      tc = session_get_transport (ll);
-      ls->port = tc->lcl_port;
-      sm = app_worker_get_local_segment_manager (server_wrk);
-    }
-  else
-    {
-      session_t *sl = (session_t *) ll;
-      transport_connection_t *tc;
-      tc = listen_session_get_transport (sl);
-      ls->port = tc->lcl_port;
-      sm = app_worker_get_listen_segment_manager (server_wrk, sl);
-    }
-
+  sm = app_worker_get_listen_segment_manager (server_wrk, ll);
   seg_index = segment_manager_add_segment (sm, seg_size);
   if (seg_index < 0)
     {
@@ -335,19 +189,20 @@ app_worker_local_session_connect (app_worker_t * client_wrk,
   cq = segment_manager_alloc_queue (seg, cprops);
 
   if (props->use_mq_eventfd)
-    application_local_session_fix_eventds (sq, cq);
+    ct_session_fix_eventds (sq, cq);
 
-  ls->server_evt_q = pointer_to_uword (sq);
-  ls->client_evt_q = pointer_to_uword (cq);
+  ct->server_evt_q = pointer_to_uword (sq);
+  ct->client_evt_q = pointer_to_uword (cq);
   rv = segment_manager_try_alloc_fifos (seg, props->rx_fifo_size,
-					props->tx_fifo_size,
-					&ls->rx_fifo, &ls->tx_fifo);
+					props->tx_fifo_size, &ls->rx_fifo,
+					&ls->tx_fifo);
   if (rv)
     {
       clib_warning ("failed to add fifos in cut-through segment");
       segment_manager_segment_reader_unlock (sm);
       goto failed;
     }
+
   sm_index = segment_manager_index (sm);
   ls->rx_fifo->ct_session_index = ls->session_index;
   ls->tx_fifo->ct_session_index = ls->session_index;
@@ -356,162 +211,107 @@ app_worker_local_session_connect (app_worker_t * client_wrk,
   ls->rx_fifo->segment_index = seg_index;
   ls->tx_fifo->segment_index = seg_index;
   ls->svm_segment_index = seg_index;
-  ls->listener_index = ll->session_index;
-  ls->client_wrk_index = client_wrk->wrk_index;
-  ls->client_opaque = opaque;
-  ls->listener_session_type = ll->session_type;
-  ls->session_state = SESSION_STATE_READY;
 
   segment_handle = segment_manager_segment_handle (sm, seg);
-  if ((rv = server->cb_fns.add_segment_callback (server_wrk->api_client_index,
-						 segment_handle)))
+  if ((rv = app_worker_add_segment_notify (server_wrk, segment_handle)))
     {
       clib_warning ("failed to notify server of new segment");
       segment_manager_segment_reader_unlock (sm);
       goto failed;
     }
   segment_manager_segment_reader_unlock (sm);
-  if ((rv = server->cb_fns.session_accept_callback ((session_t *) ls)))
-    {
-      clib_warning ("failed to send accept cut-through notify to server");
-      goto failed;
-    }
-  if (server->flags & APP_OPTIONS_FLAGS_IS_BUILTIN)
-    app_worker_local_session_connect_notify (ls);
+  ct->segment_handle = segment_handle;
 
   return 0;
 
 failed:
-  if (!has_transport)
-    segment_manager_del_segment (sm, seg);
+  segment_manager_del_segment (sm, seg);
   return rv;
 }
 
 int
-app_worker_local_session_disconnect (u32 app_wrk_index, local_session_t * ls)
+ct_connect (app_worker_t * client_wrk, session_t * ll,
+	    session_endpoint_cfg_t * sep)
 {
-  app_worker_t *client_wrk, *server_wrk;
+  u32 cct_index, ll_index, ll_ct_index;
+  ct_connection_t *sct, *cct, *ll_ct;
+  app_worker_t *server_wrk;
+  session_t *ss;
 
-  client_wrk = app_worker_get_if_valid (ls->client_wrk_index);
-  server_wrk = app_worker_get (ls->app_wrk_index);
+  ll_index = ll->session_index;
+  ll_ct_index = ll->connection_index;
 
-  if (ls->session_state == SESSION_STATE_CLOSED)
-    return app_worker_local_session_cleanup (client_wrk, server_wrk, ls);
+  cct = ct_connection_alloc ();
+  cct_index = cct->c_c_index;
+  sct = ct_connection_alloc ();
+  ll_ct = ct_connection_get (ll_ct_index);
 
-  if (app_wrk_index == ls->client_wrk_index)
-    {
-      mq_send_local_session_disconnected_cb (ls->app_wrk_index, ls);
-    }
-  else
-    {
-      if (!client_wrk)
-	{
-	  return app_worker_local_session_cleanup (client_wrk, server_wrk,
-						   ls);
-	}
-      else if (ls->session_state < SESSION_STATE_READY)
-	{
-	  application_t *client = application_get (client_wrk->app_index);
-	  client->cb_fns.session_connected_callback (client_wrk->wrk_index,
-						     ls->client_opaque,
-						     (session_t *) ls,
-						     1 /* is_fail */ );
-	  ls->session_state = SESSION_STATE_CLOSED;
-	  return app_worker_local_session_cleanup (client_wrk, server_wrk,
-						   ls);
-	}
-      else
-	{
-	  mq_send_local_session_disconnected_cb (client_wrk->wrk_index, ls);
-	}
-    }
+  /*
+   * Alloc and init client transport
+   */
+  cct = ct_connection_get (cct_index);
+  cct->c_thread_index = 0;
+  cct->c_rmt_port = sep->port;
+  cct->c_lcl_port = 0;
+  cct->c_is_ip4 = sep->is_ip4;
+  clib_memcpy (&cct->c_rmt_ip, &sep->ip, sizeof (sep->ip));
+  cct->actual_tp = ll_ct->actual_tp;
 
-  ls->session_state = SESSION_STATE_CLOSED;
+  /*
+   * Init server transport
+   */
+  sct->c_thread_index = 0;
+  sct->c_rmt_port = 0;
+  sct->c_lcl_port = ll_ct->c_lcl_port;
+  sct->c_is_ip4 = sep->is_ip4;
+  clib_memcpy (&sct->c_lcl_ip, &ll_ct->c_lcl_ip, sizeof (ll_ct->c_lcl_ip));
+  sct->client_wrk = client_wrk->wrk_index;
+  sct->c_proto = TRANSPORT_PROTO_NONE;
+  sct->client_opaque = sep->opaque;
+  sct->actual_tp = ll_ct->actual_tp;
+
+  sct->peer_index = cct->c_c_index;
+  cct->peer_index = sct->c_c_index;
 
-  return 0;
-}
+  /*
+   * Accept server session. Client session is created only after
+   * server confirms accept.
+   */
+  ss = session_alloc (0);
+  ll = listen_session_get (ll_index);
+  ss->session_type = ll->session_type;
+  ss->connection_index = sct->c_c_index;
+  ss->listener_index = ll->session_index;
+  ss->session_state = SESSION_STATE_CREATED;
 
-int
-app_worker_local_session_disconnect_w_index (u32 app_wrk_index, u32 ls_index)
-{
-  app_worker_t *app_wrk;
-  local_session_t *ls;
-  app_wrk = app_worker_get (app_wrk_index);
-  ls = app_worker_get_local_session (app_wrk, ls_index);
-  return app_worker_local_session_disconnect (app_wrk_index, ls);
-}
+  server_wrk = application_listener_select_worker (ll);
+  ss->app_wrk_index = server_wrk->wrk_index;
 
-void
-app_worker_format_local_sessions (app_worker_t * app_wrk, int verbose)
-{
-  vlib_main_t *vm = vlib_get_main ();
-  app_worker_t *client_wrk;
-  local_session_t *ls;
-  transport_proto_t tp;
-  u8 *conn = 0;
+  sct->c_s_index = ss->session_index;
+  sct->server_wrk = ss->app_wrk_index;
 
-  /* Header */
-  if (app_wrk == 0)
+  if (ct_init_local_session (client_wrk, server_wrk, sct, ss, ll))
     {
-      vlib_cli_output (vm, "%-40s%-15s%-20s", "Connection", "ServerApp",
-		       "ClientApp");
-      return;
+      clib_warning ("failed");
+      ct_connection_free (sct);
+      session_free (ss);
+      return -1;
     }
 
-  if (!pool_elts (app_wrk->local_sessions)
-      && !pool_elts (app_wrk->local_connects))
-    return;
-
-  /* *INDENT-OFF* */
-  pool_foreach (ls, app_wrk->local_sessions, ({
-    tp = session_type_transport_proto(ls->listener_session_type);
-    conn = format (0, "[L][%U] *:%u", format_transport_proto_short, tp,
-                   ls->port);
-    client_wrk = app_worker_get (ls->client_wrk_index);
-    vlib_cli_output (vm, "%-40v%-15u%-20u", conn, ls->app_index,
-                     client_wrk->app_index);
-    vec_reset_length (conn);
-  }));
-  /* *INDENT-ON* */
-
-  vec_free (conn);
-}
-
-void
-app_worker_format_local_connects (app_worker_t * app, int verbose)
-{
-  vlib_main_t *vm = vlib_get_main ();
-  u32 app_wrk_index, session_index;
-  app_worker_t *server_wrk;
-  local_session_t *ls;
-  u64 client_key;
-  u64 value;
-
-  /* Header */
-  if (app == 0)
+  ss->session_state = SESSION_STATE_ACCEPTING;
+  if (app_worker_accept_notify (server_wrk, ss))
     {
-      if (verbose)
-	vlib_cli_output (vm, "%-40s%-15s%-20s%-10s", "Connection", "App",
-			 "Peer App", "SegManager");
-      else
-	vlib_cli_output (vm, "%-40s%-15s%-20s", "Connection", "App",
-			 "Peer App");
-      return;
+      clib_warning ("failed");
+      ct_connection_free (sct);
+      session_free_w_fifos (ss);
+      return -1;
     }
 
-  if (!app->local_connects)
-    return;
-
-  /* *INDENT-OFF* */
-  hash_foreach (client_key, value, app->local_connects, ({
-    application_client_local_connect_key_parse (client_key, &app_wrk_index,
-                                                &session_index);
-    server_wrk = app_worker_get (app_wrk_index);
-    ls = app_worker_get_local_session (server_wrk, session_index);
-    vlib_cli_output (vm, "%-40s%-15s%-20s", "TODO", ls->app_wrk_index,
-                     ls->client_wrk_index);
-  }));
-  /* *INDENT-ON* */
+  cct->client_evt_q = sct->client_evt_q;
+  cct->server_evt_q = sct->server_evt_q;
+  cct->segment_handle = sct->segment_handle;
+
+  return 0;
 }
 
 u32
@@ -545,6 +345,96 @@ ct_listener_get (u32 ct_index)
   return (transport_connection_t *) ct_connection_get (ct_index);
 }
 
+int
+ct_session_connect (transport_endpoint_cfg_t * tep)
+{
+  session_endpoint_cfg_t *sep_ext;
+  session_endpoint_t *sep;
+  app_worker_t *app_wrk;
+  session_handle_t lh;
+  application_t *app;
+  app_listener_t *al;
+  u32 table_index;
+  session_t *ll;
+  u8 fib_proto;
+
+  sep_ext = (session_endpoint_cfg_t *) tep;
+  sep = (session_endpoint_t *) tep;
+  app_wrk = app_worker_get (sep_ext->app_wrk_index);
+  app = application_get (app_wrk->app_index);
+
+  sep->transport_proto = sep_ext->original_tp;
+  table_index = application_local_session_table (app);
+  lh = session_lookup_local_endpoint (table_index, sep);
+  if (lh == SESSION_DROP_HANDLE)
+    return VNET_API_ERROR_APP_CONNECT_FILTERED;
+
+  if (lh == SESSION_INVALID_HANDLE)
+    goto global_scope;
+
+  ll = listen_session_get_from_handle (lh);
+  al = app_listener_get_w_session (ll);
+
+  /*
+   * Break loop if rule in local table points to connecting app. This
+   * can happen if client is a generic proxy. Route connect through
+   * global table instead.
+   */
+  if (al->app_index == app->app_index)
+    goto global_scope;
+
+  return ct_connect (app_wrk, ll, sep_ext);
+
+  /*
+   * If nothing found, check the global scope for locally attached
+   * destinations. Make sure first that we're allowed to.
+   */
+
+global_scope:
+  if (session_endpoint_is_local (sep))
+    return VNET_API_ERROR_SESSION_CONNECT;
+
+  if (!application_has_global_scope (app))
+    return VNET_API_ERROR_APP_CONNECT_SCOPE;
+
+  fib_proto = session_endpoint_fib_proto (sep);
+  table_index = application_session_table (app, fib_proto);
+  ll = session_lookup_listener (table_index, sep);
+
+  if (ll)
+    return ct_connect (app_wrk, ll, sep_ext);
+
+  /* Failed to connect but no error */
+  return 1;
+}
+
+void
+ct_session_close (u32 ct_index, u32 thread_index)
+{
+  ct_connection_t *ct, *peer_ct;
+  session_t *s;
+
+  ct = ct_connection_get (ct_index);
+  peer_ct = ct_connection_get (ct->peer_index);
+  if (peer_ct)
+    {
+      peer_ct->peer_index = ~0;
+      session_transport_closing_notify (&peer_ct->connection);
+    }
+
+  s = session_get (ct->c_s_index, 0);
+  app_worker_del_segment_notify (app_worker_get (s->app_wrk_index),
+				 ct->segment_handle);
+  session_free_w_fifos (s);
+  ct_connection_free (ct);
+}
+
+transport_connection_t *
+ct_session_get (u32 ct_index, u32 thread_index)
+{
+  return (transport_connection_t *) ct_connection_get (ct_index);
+}
+
 static u8 *
 format_ct_connection_id (u8 * s, va_list * args)
 {
@@ -583,14 +473,57 @@ format_ct_listener (u8 * s, va_list * args)
   return s;
 }
 
+u8 *
+format_ct_connection (u8 * s, va_list * args)
+{
+  ct_connection_t *ct = va_arg (*args, ct_connection_t *);
+  u32 verbose = va_arg (*args, u32);
+
+  if (!ct)
+    return s;
+  s = format (s, "%-50U", format_ct_connection_id, ct);
+  if (verbose)
+    {
+      s = format (s, "%-15s", "ESTABLISHED");
+      if (verbose > 1)
+	{
+	  s = format (s, "\n");
+	}
+    }
+  return s;
+}
+
+u8 *
+format_ct_session (u8 * s, va_list * args)
+{
+  u32 ct_index = va_arg (*args, u32);
+  u32 __clib_unused thread_index = va_arg (*args, u32);
+  u32 verbose = va_arg (*args, u32);
+  ct_connection_t *ct;
+
+  ct = ct_connection_get (ct_index);
+  if (!ct)
+    {
+      s = format (s, "empty\n");
+      return s;
+    }
+
+  s = format (s, "%U", format_ct_connection, ct, verbose);
+  return s;
+}
+
 /* *INDENT-OFF* */
 const static transport_proto_vft_t cut_thru_proto = {
   .start_listen = ct_start_listen,
   .stop_listen = ct_stop_listen,
   .get_listener = ct_listener_get,
+  .connect = ct_session_connect,
+  .close = ct_session_close,
+  .get_connection = ct_session_get,
   .tx_type = TRANSPORT_TX_INTERNAL,
   .service_type = TRANSPORT_SERVICE_APP,
   .format_listener = format_ct_listener,
+  .format_connection = format_ct_session,
 };
 /* *INDENT-ON* */