src/vnet/session/application_local.c

   1 /*
   2  * Copyright (c) 2019 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <vnet/session/application_local.h>
  17 #include <vnet/session/session.h>
  18
  19 typedef struct ct_main_
  20 {
  21   ct_connection_t **connections;        /**< Per-worker connection pools */
  22   u32 n_workers;                        /**< Number of vpp workers */
  23   u32 n_sessions;                       /**< Cumulative sessions counter */
  24   u32 *ho_reusable;                     /**< Vector of reusable ho indices */
  25   clib_spinlock_t ho_reuseable_lock;    /**< Lock for reusable ho indices */
  26 } ct_main_t;
  27
  28 static ct_main_t ct_main;
  29
  30 static ct_connection_t *
  31 ct_connection_alloc (u32 thread_index)
  32 {
  33   ct_connection_t *ct;
  34
  35   pool_get_zero (ct_main.connections[thread_index], ct);
  36   ct->c_c_index = ct - ct_main.connections[thread_index];
  37   ct->c_thread_index = thread_index;
  38   ct->client_wrk = ~0;
  39   ct->server_wrk = ~0;
  40   return ct;
  41 }
  42
  43 static ct_connection_t *
  44 ct_connection_get (u32 ct_index, u32 thread_index)
  45 {
  46   if (pool_is_free_index (ct_main.connections[thread_index], ct_index))
  47     return 0;
  48   return pool_elt_at_index (ct_main.connections[thread_index], ct_index);
  49 }
  50
  51 static void
  52 ct_connection_free (ct_connection_t * ct)
  53 {
  54   if (CLIB_DEBUG)
  55     {
  56       u32 thread_index = ct->c_thread_index;
  57       memset (ct, 0xfc, sizeof (*ct));
  58       pool_put (ct_main.connections[thread_index], ct);
  59       return;
  60     }
  61   pool_put (ct_main.connections[ct->c_thread_index], ct);
  62 }
  63
  64 static ct_connection_t *
  65 ct_half_open_alloc (void)
  66 {
  67   ct_main_t *cm = &ct_main;
  68   u32 *hip;
  69
  70   clib_spinlock_lock (&cm->ho_reuseable_lock);
  71   vec_foreach (hip, cm->ho_reusable)
  72     pool_put_index (cm->connections[0], *hip);
  73   vec_reset_length (cm->ho_reusable);
  74   clib_spinlock_unlock (&cm->ho_reuseable_lock);
  75
  76   return ct_connection_alloc (0);
  77 }
  78
  79 void
  80 ct_half_open_add_reusable (u32 ho_index)
  81 {
  82   ct_main_t *cm = &ct_main;
  83
  84   clib_spinlock_lock (&cm->ho_reuseable_lock);
  85   vec_add1 (cm->ho_reusable, ho_index);
  86   clib_spinlock_unlock (&cm->ho_reuseable_lock);
  87 }
  88
  89 session_t *
  90 ct_session_get_peer (session_t * s)
  91 {
  92   ct_connection_t *ct, *peer_ct;
  93   ct = ct_connection_get (s->connection_index, s->thread_index);
  94   peer_ct = ct_connection_get (ct->peer_index, s->thread_index);
  95   return session_get (peer_ct->c_s_index, s->thread_index);
  96 }
  97
  98 void
  99 ct_session_endpoint (session_t * ll, session_endpoint_t * sep)
 100 {
 101   ct_connection_t *ct;
 102   ct = (ct_connection_t *) session_get_transport (ll);
 103   sep->transport_proto = ct->actual_tp;
 104   sep->port = ct->c_lcl_port;
 105   sep->is_ip4 = ct->c_is_ip4;
 106   ip_copy (&sep->ip, &ct->c_lcl_ip, ct->c_is_ip4);
 107 }
 108
 109 int
 110 ct_session_connect_notify (session_t * ss)
 111 {
 112   u32 ss_index, opaque, thread_index;
 113   ct_connection_t *sct, *cct;
 114   app_worker_t *client_wrk;
 115   segment_manager_t *sm;
 116   fifo_segment_t *seg;
 117   u64 segment_handle;
 118   int err = 0;
 119   session_t *cs;
 120
 121   ss_index = ss->session_index;
 122   thread_index = ss->thread_index;
 123   sct = (ct_connection_t *) session_get_transport (ss);
 124   client_wrk = app_worker_get (sct->client_wrk);
 125   opaque = sct->client_opaque;
 126
 127   sm = segment_manager_get (ss->rx_fifo->segment_manager);
 128   seg = segment_manager_get_segment_w_lock (sm, ss->rx_fifo->segment_index);
 129   segment_handle = segment_manager_segment_handle (sm, seg);
 130
 131   if ((err = app_worker_add_segment_notify (client_wrk, segment_handle)))
 132     {
 133       clib_warning ("failed to notify client %u of new segment",
 134                     sct->client_wrk);
 135       segment_manager_segment_reader_unlock (sm);
 136       session_close (ss);
 137       goto error;
 138     }
 139   else
 140     {
 141       segment_manager_segment_reader_unlock (sm);
 142     }
 143
 144   /*
 145    * Alloc client session
 146    */
 147   cct = ct_connection_get (sct->peer_index, thread_index);
 148
 149   /* Client closed while waiting for reply from server */
 150   if (!cct)
 151     {
 152       session_transport_closing_notify (&sct->connection);
 153       session_transport_delete_notify (&sct->connection);
 154       ct_connection_free (sct);
 155       return 0;
 156     }
 157
 158   session_half_open_delete_notify (&cct->connection);
 159   cct->flags &= ~CT_CONN_F_HALF_OPEN;
 160
 161   cs = session_alloc (thread_index);
 162   ss = session_get (ss_index, thread_index);
 163   cs->session_type = ss->session_type;
 164   cs->listener_handle = SESSION_INVALID_HANDLE;
 165   cs->session_state = SESSION_STATE_CONNECTING;
 166   cs->app_wrk_index = client_wrk->wrk_index;
 167   cs->connection_index = cct->c_c_index;
 168
 169   cct->c_s_index = cs->session_index;
 170   cct->client_rx_fifo = ss->tx_fifo;
 171   cct->client_tx_fifo = ss->rx_fifo;
 172
 173   cct->client_rx_fifo->refcnt++;
 174   cct->client_tx_fifo->refcnt++;
 175
 176   /* This will allocate fifos for the session. They won't be used for
 177    * exchanging data but they will be used to close the connection if
 178    * the segment manager/worker is freed */
 179   if ((err = app_worker_init_connected (client_wrk, cs)))
 180     {
 181       session_close (ss);
 182       session_free (cs);
 183       goto error;
 184     }
 185
 186   cs->session_state = SESSION_STATE_CONNECTING;
 187
 188   if (app_worker_connect_notify (client_wrk, cs, err, opaque))
 189     {
 190       session_close (ss);
 191       segment_manager_dealloc_fifos (cs->rx_fifo, cs->tx_fifo);
 192       session_free (cs);
 193       return -1;
 194     }
 195
 196   cs = session_get (cct->c_s_index, cct->c_thread_index);
 197   cs->session_state = SESSION_STATE_READY;
 198
 199   return 0;
 200
 201 error:
 202   app_worker_connect_notify (client_wrk, 0, err, opaque);
 203   return -1;
 204 }
 205
 206 static int
 207 ct_init_accepted_session (app_worker_t * server_wrk,
 208                           ct_connection_t * ct, session_t * ls,
 209                           session_t * ll)
 210 {
 211   u32 round_rx_fifo_sz, round_tx_fifo_sz, sm_index, seg_size;
 212   segment_manager_props_t *props;
 213   application_t *server;
 214   segment_manager_t *sm;
 215   u32 margin = 16 << 10;
 216   fifo_segment_t *seg;
 217   u64 segment_handle;
 218   int seg_index, rv;
 219
 220   server = application_get (server_wrk->app_index);
 221
 222   props = application_segment_manager_properties (server);
 223   round_rx_fifo_sz = 1 << max_log2 (props->rx_fifo_size);
 224   round_tx_fifo_sz = 1 << max_log2 (props->tx_fifo_size);
 225   /* Increase size because of inefficient chunk allocations. Depending on
 226    * how data is consumed, it may happen that more chunks than needed are
 227    * allocated.
 228    * TODO should remove once allocations are done more efficiently */
 229   seg_size = 4 * (round_rx_fifo_sz + round_tx_fifo_sz + margin);
 230
 231   sm = app_worker_get_listen_segment_manager (server_wrk, ll);
 232   seg_index = segment_manager_add_segment (sm, seg_size, 0);
 233   if (seg_index < 0)
 234     {
 235       clib_warning ("failed to add new cut-through segment");
 236       return seg_index;
 237     }
 238   seg = segment_manager_get_segment_w_lock (sm, seg_index);
 239
 240   rv = segment_manager_try_alloc_fifos (seg, ls->thread_index,
 241                                         props->rx_fifo_size,
 242                                         props->tx_fifo_size, &ls->rx_fifo,
 243                                         &ls->tx_fifo);
 244   if (rv)
 245     {
 246       clib_warning ("failed to add fifos in cut-through segment");
 247       segment_manager_segment_reader_unlock (sm);
 248       goto failed;
 249     }
 250
 251   sm_index = segment_manager_index (sm);
 252   ls->rx_fifo->shr->master_session_index = ls->session_index;
 253   ls->tx_fifo->shr->master_session_index = ls->session_index;
 254   ls->rx_fifo->master_thread_index = ls->thread_index;
 255   ls->tx_fifo->master_thread_index = ls->thread_index;
 256   ls->rx_fifo->segment_manager = sm_index;
 257   ls->tx_fifo->segment_manager = sm_index;
 258   ls->rx_fifo->segment_index = seg_index;
 259   ls->tx_fifo->segment_index = seg_index;
 260
 261   segment_handle = segment_manager_segment_handle (sm, seg);
 262   if ((rv = app_worker_add_segment_notify (server_wrk, segment_handle)))
 263     {
 264       clib_warning ("failed to notify server of new segment");
 265       segment_manager_segment_reader_unlock (sm);
 266       goto failed;
 267     }
 268   segment_manager_segment_reader_unlock (sm);
 269   ct->segment_handle = segment_handle;
 270
 271   return 0;
 272
 273 failed:
 274   segment_manager_lock_and_del_segment (sm, seg_index);
 275   return rv;
 276 }
 277
 278 static void
 279 ct_accept_rpc_wrk_handler (void *accept_args)
 280 {
 281   u32 cct_index, ho_index, thread_index, ll_index;
 282   ct_connection_t *sct, *cct, *ll_ct, *ho;
 283   app_worker_t *server_wrk;
 284   session_t *ss, *ll;
 285
 286   /*
 287    * Alloc client ct and initialize from ho
 288    */
 289   thread_index = vlib_get_thread_index ();
 290   cct = ct_connection_alloc (thread_index);
 291   cct_index = cct->c_c_index;
 292
 293   ho_index = pointer_to_uword (accept_args);
 294   ho = ct_connection_get (ho_index, 0);
 295
 296   /* Unlikely but half-open session and transport could have been freed */
 297   if (PREDICT_FALSE (!ho))
 298     {
 299       ct_connection_free (cct);
 300       return;
 301     }
 302
 303   clib_memcpy (cct, ho, sizeof (*ho));
 304   cct->c_c_index = cct_index;
 305   cct->c_thread_index = thread_index;
 306   cct->flags |= CT_CONN_F_HALF_OPEN;
 307
 308   /* Notify session layer that half-open is on a different thread
 309    * and mark ho connection index reusable. Avoids another rpc
 310    */
 311   session_half_open_migrate_notify (&cct->connection);
 312   session_half_open_migrated_notify (&cct->connection);
 313   ct_half_open_add_reusable (ho_index);
 314
 315   /*
 316    * Alloc and init server transport
 317    */
 318
 319   ll_index = cct->peer_index;
 320   ll = listen_session_get (ll_index);
 321   sct = ct_connection_alloc (thread_index);
 322   ll_ct = ct_connection_get (ll->connection_index, 0 /* listener thread */);
 323
 324   /* Make sure cct is valid after sct alloc */
 325   cct = ct_connection_get (cct_index, thread_index);
 326   cct->actual_tp = ll_ct->actual_tp;
 327
 328   sct->c_rmt_port = 0;
 329   sct->c_lcl_port = ll_ct->c_lcl_port;
 330   sct->c_is_ip4 = cct->c_is_ip4;
 331   clib_memcpy (&sct->c_lcl_ip, &ll_ct->c_lcl_ip, sizeof (ll_ct->c_lcl_ip));
 332   sct->client_wrk = cct->client_wrk;
 333   sct->c_proto = TRANSPORT_PROTO_NONE;
 334   sct->client_opaque = cct->client_opaque;
 335   sct->actual_tp = ll_ct->actual_tp;
 336
 337   sct->peer_index = cct->c_c_index;
 338   cct->peer_index = sct->c_c_index;
 339
 340   /*
 341    * Accept server session. Client session is created only after
 342    * server confirms accept.
 343    */
 344   ss = session_alloc (thread_index);
 345   ll = listen_session_get (ll_index);
 346   ss->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE,
 347                                                      sct->c_is_ip4);
 348   ss->connection_index = sct->c_c_index;
 349   ss->listener_handle = listen_session_get_handle (ll);
 350   ss->session_state = SESSION_STATE_CREATED;
 351
 352   server_wrk = application_listener_select_worker (ll);
 353   ss->app_wrk_index = server_wrk->wrk_index;
 354
 355   sct->c_s_index = ss->session_index;
 356   sct->server_wrk = ss->app_wrk_index;
 357
 358   if (ct_init_accepted_session (server_wrk, sct, ss, ll))
 359     {
 360       ct_connection_free (sct);
 361       session_free (ss);
 362       return;
 363     }
 364
 365   ss->session_state = SESSION_STATE_ACCEPTING;
 366   if (app_worker_accept_notify (server_wrk, ss))
 367     {
 368       ct_connection_free (sct);
 369       segment_manager_dealloc_fifos (ss->rx_fifo, ss->tx_fifo);
 370       session_free (ss);
 371       return;
 372     }
 373
 374   cct->segment_handle = sct->segment_handle;
 375 }
 376
 377 static int
 378 ct_connect (app_worker_t * client_wrk, session_t * ll,
 379             session_endpoint_cfg_t * sep)
 380 {
 381   u32 thread_index, ho_index;
 382   ct_main_t *cm = &ct_main;
 383   ct_connection_t *ho;
 384
 385   /* Simple round-robin policy for spreading sessions over workers. We skip
 386    * thread index 0, i.e., offset the index by 1, when we have workers as it
 387    * is the one dedicated to main thread. Note that n_workers does not include
 388    * main thread */
 389   cm->n_sessions += 1;
 390   thread_index = cm->n_workers ? (cm->n_sessions % cm->n_workers) + 1 : 0;
 391
 392   /*
 393    * Alloc and init client half-open transport
 394    */
 395
 396   ho = ct_half_open_alloc ();
 397   ho_index = ho->c_c_index;
 398   ho->c_rmt_port = sep->port;
 399   ho->c_lcl_port = 0;
 400   ho->c_is_ip4 = sep->is_ip4;
 401   ho->client_opaque = sep->opaque;
 402   ho->client_wrk = client_wrk->wrk_index;
 403   ho->peer_index = ll->session_index;
 404   ho->c_proto = TRANSPORT_PROTO_NONE;
 405   ho->c_flags |= TRANSPORT_CONNECTION_F_NO_LOOKUP;
 406   clib_memcpy (&ho->c_rmt_ip, &sep->ip, sizeof (&sep->ip));
 407   ho->flags |= CT_CONN_F_CLIENT;
 408   ho->c_s_index = ~0;
 409
 410   /*
 411    * Accept connection on thread selected above. Connected reply comes
 412    * after server accepts the connection.
 413    */
 414
 415   session_send_rpc_evt_to_thread_force (thread_index,
 416                                         ct_accept_rpc_wrk_handler,
 417                                         uword_to_pointer (ho_index, void *));
 418
 419   return ho_index;
 420 }
 421
 422 static u32
 423 ct_start_listen (u32 app_listener_index, transport_endpoint_t * tep)
 424 {
 425   session_endpoint_cfg_t *sep;
 426   ct_connection_t *ct;
 427
 428   sep = (session_endpoint_cfg_t *) tep;
 429   ct = ct_connection_alloc (0);
 430   ct->server_wrk = sep->app_wrk_index;
 431   ct->c_is_ip4 = sep->is_ip4;
 432   clib_memcpy (&ct->c_lcl_ip, &sep->ip, sizeof (sep->ip));
 433   ct->c_lcl_port = sep->port;
 434   ct->c_s_index = app_listener_index;
 435   ct->actual_tp = sep->transport_proto;
 436   return ct->c_c_index;
 437 }
 438
 439 static u32
 440 ct_stop_listen (u32 ct_index)
 441 {
 442   ct_connection_t *ct;
 443   ct = ct_connection_get (ct_index, 0);
 444   ct_connection_free (ct);
 445   return 0;
 446 }
 447
 448 static transport_connection_t *
 449 ct_listener_get (u32 ct_index)
 450 {
 451   return (transport_connection_t *) ct_connection_get (ct_index, 0);
 452 }
 453
 454 static transport_connection_t *
 455 ct_half_open_get (u32 ct_index)
 456 {
 457   return (transport_connection_t *) ct_connection_get (ct_index, 0);
 458 }
 459
 460 static void
 461 ct_session_cleanup (u32 conn_index, u32 thread_index)
 462 {
 463   ct_connection_t *ct, *peer_ct;
 464
 465   ct = ct_connection_get (conn_index, thread_index);
 466   if (!ct)
 467     return;
 468
 469   peer_ct = ct_connection_get (ct->peer_index, thread_index);
 470   if (peer_ct)
 471     peer_ct->peer_index = ~0;
 472
 473   ct_connection_free (ct);
 474 }
 475
 476 static void
 477 ct_cleanup_ho (u32 ho_index)
 478 {
 479   ct_connection_free (ct_connection_get (ho_index, 0));
 480 }
 481
 482 static int
 483 ct_session_connect (transport_endpoint_cfg_t * tep)
 484 {
 485   session_endpoint_cfg_t *sep_ext;
 486   session_endpoint_t _sep, *sep = &_sep;
 487   app_worker_t *app_wrk;
 488   session_handle_t lh;
 489   application_t *app;
 490   app_listener_t *al;
 491   u32 table_index;
 492   session_t *ll;
 493   u8 fib_proto;
 494
 495   sep_ext = (session_endpoint_cfg_t *) tep;
 496   _sep = *(session_endpoint_t *) tep;
 497   app_wrk = app_worker_get (sep_ext->app_wrk_index);
 498   app = application_get (app_wrk->app_index);
 499
 500   sep->transport_proto = sep_ext->original_tp;
 501   table_index = application_local_session_table (app);
 502   lh = session_lookup_local_endpoint (table_index, sep);
 503   if (lh == SESSION_DROP_HANDLE)
 504     return SESSION_E_FILTERED;
 505
 506   if (lh == SESSION_INVALID_HANDLE)
 507     goto global_scope;
 508
 509   ll = listen_session_get_from_handle (lh);
 510   al = app_listener_get_w_session (ll);
 511
 512   /*
 513    * Break loop if rule in local table points to connecting app. This
 514    * can happen if client is a generic proxy. Route connect through
 515    * global table instead.
 516    */
 517   if (al->app_index == app->app_index)
 518     goto global_scope;
 519
 520   return ct_connect (app_wrk, ll, sep_ext);
 521
 522   /*
 523    * If nothing found, check the global scope for locally attached
 524    * destinations. Make sure first that we're allowed to.
 525    */
 526
 527 global_scope:
 528   if (session_endpoint_is_local (sep))
 529     return SESSION_E_NOROUTE;
 530
 531   if (!application_has_global_scope (app))
 532     return SESSION_E_SCOPE;
 533
 534   fib_proto = session_endpoint_fib_proto (sep);
 535   table_index = session_lookup_get_index_for_fib (fib_proto, sep->fib_index);
 536   ll = session_lookup_listener_wildcard (table_index, sep);
 537
 538   if (ll)
 539     return ct_connect (app_wrk, ll, sep_ext);
 540
 541   /* Failed to connect but no error */
 542   return SESSION_E_LOCAL_CONNECT;
 543 }
 544
 545 static void
 546 ct_session_close (u32 ct_index, u32 thread_index)
 547 {
 548   ct_connection_t *ct, *peer_ct;
 549   app_worker_t *app_wrk;
 550   session_t *s;
 551
 552   ct = ct_connection_get (ct_index, thread_index);
 553   peer_ct = ct_connection_get (ct->peer_index, thread_index);
 554   if (peer_ct)
 555     {
 556       peer_ct->peer_index = ~0;
 557       /* Make sure session was allocated */
 558       if (peer_ct->flags & CT_CONN_F_HALF_OPEN)
 559         {
 560           app_wrk = app_worker_get (peer_ct->client_wrk);
 561           app_worker_connect_notify (app_wrk, 0, SESSION_E_REFUSED,
 562                                      peer_ct->client_opaque);
 563         }
 564       else if (peer_ct->c_s_index != ~0)
 565         session_transport_closing_notify (&peer_ct->connection);
 566       else
 567         ct_connection_free (peer_ct);
 568     }
 569
 570   s = session_get (ct->c_s_index, ct->c_thread_index);
 571   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
 572   if (app_wrk)
 573     app_worker_del_segment_notify (app_wrk, ct->segment_handle);
 574   session_free_w_fifos (s);
 575   if (ct->flags & CT_CONN_F_CLIENT)
 576     segment_manager_dealloc_fifos (ct->client_rx_fifo, ct->client_tx_fifo);
 577
 578   ct_connection_free (ct);
 579 }
 580
 581 static transport_connection_t *
 582 ct_session_get (u32 ct_index, u32 thread_index)
 583 {
 584   return (transport_connection_t *) ct_connection_get (ct_index,
 585                                                        thread_index);
 586 }
 587
 588 static u8 *
 589 format_ct_connection_id (u8 * s, va_list * args)
 590 {
 591   ct_connection_t *ct = va_arg (*args, ct_connection_t *);
 592   if (!ct)
 593     return s;
 594   if (ct->c_is_ip4)
 595     {
 596       s = format (s, "[%d:%d][CT:%U] %U:%d->%U:%d", ct->c_thread_index,
 597                   ct->c_s_index, format_transport_proto_short, ct->actual_tp,
 598                   format_ip4_address, &ct->c_lcl_ip4,
 599                   clib_net_to_host_u16 (ct->c_lcl_port), format_ip4_address,
 600                   &ct->c_rmt_ip4, clib_net_to_host_u16 (ct->c_rmt_port));
 601     }
 602   else
 603     {
 604       s = format (s, "[%d:%d][CT:%U] %U:%d->%U:%d", ct->c_thread_index,
 605                   ct->c_s_index, format_transport_proto_short, ct->actual_tp,
 606                   format_ip6_address, &ct->c_lcl_ip6,
 607                   clib_net_to_host_u16 (ct->c_lcl_port), format_ip6_address,
 608                   &ct->c_rmt_ip6, clib_net_to_host_u16 (ct->c_rmt_port));
 609     }
 610
 611   return s;
 612 }
 613
 614 static int
 615 ct_custom_tx (void *session, transport_send_params_t * sp)
 616 {
 617   session_t *s = (session_t *) session;
 618   if (session_has_transport (s))
 619     return 0;
 620   /* If event enqueued towards peer, remove from scheduler and remove
 621    * session tx flag, i.e., accept new tx events. Unset fifo flag now to
 622    * avoid missing events if peer did not clear fifo flag yet, which is
 623    * interpreted as successful notification and session is descheduled. */
 624   svm_fifo_unset_event (s->tx_fifo);
 625   if (!ct_session_tx (s))
 626     sp->flags = TRANSPORT_SND_F_DESCHED;
 627
 628   /* The scheduler uses packet count as a means of upper bounding the amount
 629    * of work done per dispatch. So make it look like we have sent something */
 630   return 1;
 631 }
 632
 633 static int
 634 ct_app_rx_evt (transport_connection_t * tc)
 635 {
 636   ct_connection_t *ct = (ct_connection_t *) tc, *peer_ct;
 637   session_t *ps;
 638
 639   peer_ct = ct_connection_get (ct->peer_index, tc->thread_index);
 640   if (!peer_ct)
 641     return -1;
 642   ps = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
 643   return session_dequeue_notify (ps);
 644 }
 645
 646 static u8 *
 647 format_ct_listener (u8 * s, va_list * args)
 648 {
 649   u32 tc_index = va_arg (*args, u32);
 650   u32 __clib_unused thread_index = va_arg (*args, u32);
 651   u32 __clib_unused verbose = va_arg (*args, u32);
 652   ct_connection_t *ct = ct_connection_get (tc_index, 0);
 653   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
 654   if (verbose)
 655     s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "LISTEN");
 656   return s;
 657 }
 658
 659 static u8 *
 660 format_ct_half_open (u8 *s, va_list *args)
 661 {
 662   u32 ho_index = va_arg (*args, u32);
 663   u32 verbose = va_arg (*args, u32);
 664   ct_connection_t *ct = ct_connection_get (ho_index, 0);
 665   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
 666   if (verbose)
 667     s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "HALF-OPEN");
 668   return s;
 669 }
 670
 671 static u8 *
 672 format_ct_connection (u8 * s, va_list * args)
 673 {
 674   ct_connection_t *ct = va_arg (*args, ct_connection_t *);
 675   u32 verbose = va_arg (*args, u32);
 676
 677   if (!ct)
 678     return s;
 679   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
 680   if (verbose)
 681     {
 682       s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "ESTABLISHED");
 683       if (verbose > 1)
 684         {
 685           s = format (s, "\n");
 686         }
 687     }
 688   return s;
 689 }
 690
 691 static u8 *
 692 format_ct_session (u8 * s, va_list * args)
 693 {
 694   u32 ct_index = va_arg (*args, u32);
 695   u32 thread_index = va_arg (*args, u32);
 696   u32 verbose = va_arg (*args, u32);
 697   ct_connection_t *ct;
 698
 699   ct = ct_connection_get (ct_index, thread_index);
 700   if (!ct)
 701     {
 702       s = format (s, "empty\n");
 703       return s;
 704     }
 705
 706   s = format (s, "%U", format_ct_connection, ct, verbose);
 707   return s;
 708 }
 709
 710 clib_error_t *
 711 ct_enable_disable (vlib_main_t * vm, u8 is_en)
 712 {
 713   ct_main_t *cm = &ct_main;
 714
 715   cm->n_workers = vlib_num_workers ();
 716   vec_validate (cm->connections, cm->n_workers);
 717   clib_spinlock_init (&cm->ho_reuseable_lock);
 718
 719   return 0;
 720 }
 721
 722 /* *INDENT-OFF* */
 723 static const transport_proto_vft_t cut_thru_proto = {
 724   .enable = ct_enable_disable,
 725   .start_listen = ct_start_listen,
 726   .stop_listen = ct_stop_listen,
 727   .get_connection = ct_session_get,
 728   .get_listener = ct_listener_get,
 729   .get_half_open = ct_half_open_get,
 730   .cleanup = ct_session_cleanup,
 731   .cleanup_ho = ct_cleanup_ho,
 732   .connect = ct_session_connect,
 733   .close = ct_session_close,
 734   .custom_tx = ct_custom_tx,
 735   .app_rx_evt = ct_app_rx_evt,
 736   .format_listener = format_ct_listener,
 737   .format_half_open = format_ct_half_open,
 738   .format_connection = format_ct_session,
 739   .transport_options = {
 740     .name = "ct",
 741     .short_name = "C",
 742     .tx_type = TRANSPORT_TX_INTERNAL,
 743     .service_type = TRANSPORT_SERVICE_VC,
 744   },
 745 };
 746 /* *INDENT-ON* */
 747
 748 int
 749 ct_session_tx (session_t * s)
 750 {
 751   ct_connection_t *ct, *peer_ct;
 752   session_t *peer_s;
 753
 754   ct = (ct_connection_t *) session_get_transport (s);
 755   peer_ct = ct_connection_get (ct->peer_index, ct->c_thread_index);
 756   if (!peer_ct)
 757     return 0;
 758   peer_s = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
 759   if (peer_s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
 760     return 0;
 761   return session_enqueue_notify (peer_s);
 762 }
 763
 764 static clib_error_t *
 765 ct_transport_init (vlib_main_t * vm)
 766 {
 767   transport_register_protocol (TRANSPORT_PROTO_NONE, &cut_thru_proto,
 768                                FIB_PROTOCOL_IP4, ~0);
 769   transport_register_protocol (TRANSPORT_PROTO_NONE, &cut_thru_proto,
 770                                FIB_PROTOCOL_IP6, ~0);
 771   return 0;
 772 }
 773
 774 VLIB_INIT_FUNCTION (ct_transport_init);
 775
 776 /*
 777  * fd.io coding-style-patch-verification: ON
 778  *
 779  * Local Variables:
 780  * eval: (c-set-style "gnu")
 781  * End:
 782  */