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, *ho;
 283   transport_connection_t *ll_ct;
 284   app_worker_t *server_wrk;
 285   session_t *ss, *ll;
 286
 287   /*
 288    * Alloc client ct and initialize from ho
 289    */
 290   thread_index = vlib_get_thread_index ();
 291   cct = ct_connection_alloc (thread_index);
 292   cct_index = cct->c_c_index;
 293
 294   ho_index = pointer_to_uword (accept_args);
 295   ho = ct_connection_get (ho_index, 0);
 296
 297   /* Unlikely but half-open session and transport could have been freed */
 298   if (PREDICT_FALSE (!ho))
 299     {
 300       ct_connection_free (cct);
 301       return;
 302     }
 303
 304   clib_memcpy (cct, ho, sizeof (*ho));
 305   cct->c_c_index = cct_index;
 306   cct->c_thread_index = thread_index;
 307   cct->flags |= CT_CONN_F_HALF_OPEN;
 308
 309   /* Notify session layer that half-open is on a different thread
 310    * and mark ho connection index reusable. Avoids another rpc
 311    */
 312   session_half_open_migrate_notify (&cct->connection);
 313   session_half_open_migrated_notify (&cct->connection);
 314   ct_half_open_add_reusable (ho_index);
 315
 316   /*
 317    * Alloc and init server transport
 318    */
 319
 320   ll_index = cct->peer_index;
 321   ll = listen_session_get (ll_index);
 322   sct = ct_connection_alloc (thread_index);
 323   /* Transport not necessarily ct but it might, so grab after sct alloc */
 324   ll_ct = listen_session_get_transport (ll);
 325
 326   /* Make sure cct is valid after sct alloc */
 327   cct = ct_connection_get (cct_index, thread_index);
 328
 329   sct->c_rmt_port = 0;
 330   sct->c_lcl_port = ll_ct->lcl_port;
 331   sct->c_is_ip4 = cct->c_is_ip4;
 332   clib_memcpy (&sct->c_lcl_ip, &ll_ct->lcl_ip, sizeof (ll_ct->lcl_ip));
 333   sct->client_wrk = cct->client_wrk;
 334   sct->c_proto = TRANSPORT_PROTO_NONE;
 335   sct->client_opaque = cct->client_opaque;
 336   sct->actual_tp = cct->actual_tp;
 337
 338   sct->peer_index = cct->c_c_index;
 339   cct->peer_index = sct->c_c_index;
 340
 341   /*
 342    * Accept server session. Client session is created only after
 343    * server confirms accept.
 344    */
 345   ss = session_alloc (thread_index);
 346   ll = listen_session_get (ll_index);
 347   ss->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE,
 348                                                      sct->c_is_ip4);
 349   ss->connection_index = sct->c_c_index;
 350   ss->listener_handle = listen_session_get_handle (ll);
 351   ss->session_state = SESSION_STATE_CREATED;
 352
 353   server_wrk = application_listener_select_worker (ll);
 354   ss->app_wrk_index = server_wrk->wrk_index;
 355
 356   sct->c_s_index = ss->session_index;
 357   sct->server_wrk = ss->app_wrk_index;
 358
 359   if (ct_init_accepted_session (server_wrk, sct, ss, ll))
 360     {
 361       ct_connection_free (sct);
 362       session_free (ss);
 363       return;
 364     }
 365
 366   ss->session_state = SESSION_STATE_ACCEPTING;
 367   if (app_worker_accept_notify (server_wrk, ss))
 368     {
 369       ct_connection_free (sct);
 370       segment_manager_dealloc_fifos (ss->rx_fifo, ss->tx_fifo);
 371       session_free (ss);
 372       return;
 373     }
 374
 375   cct->segment_handle = sct->segment_handle;
 376 }
 377
 378 static int
 379 ct_connect (app_worker_t * client_wrk, session_t * ll,
 380             session_endpoint_cfg_t * sep)
 381 {
 382   u32 thread_index, ho_index;
 383   ct_main_t *cm = &ct_main;
 384   ct_connection_t *ho;
 385
 386   /* Simple round-robin policy for spreading sessions over workers. We skip
 387    * thread index 0, i.e., offset the index by 1, when we have workers as it
 388    * is the one dedicated to main thread. Note that n_workers does not include
 389    * main thread */
 390   cm->n_sessions += 1;
 391   thread_index = cm->n_workers ? (cm->n_sessions % cm->n_workers) + 1 : 0;
 392
 393   /*
 394    * Alloc and init client half-open transport
 395    */
 396
 397   ho = ct_half_open_alloc ();
 398   ho_index = ho->c_c_index;
 399   ho->c_rmt_port = sep->port;
 400   ho->c_lcl_port = 0;
 401   ho->c_is_ip4 = sep->is_ip4;
 402   ho->client_opaque = sep->opaque;
 403   ho->client_wrk = client_wrk->wrk_index;
 404   ho->peer_index = ll->session_index;
 405   ho->c_proto = TRANSPORT_PROTO_NONE;
 406   ho->c_flags |= TRANSPORT_CONNECTION_F_NO_LOOKUP;
 407   clib_memcpy (&ho->c_rmt_ip, &sep->ip, sizeof (sep->ip));
 408   ho->flags |= CT_CONN_F_CLIENT;
 409   ho->c_s_index = ~0;
 410   ho->actual_tp = sep->transport_proto;
 411
 412   /*
 413    * Accept connection on thread selected above. Connected reply comes
 414    * after server accepts the connection.
 415    */
 416
 417   session_send_rpc_evt_to_thread_force (thread_index,
 418                                         ct_accept_rpc_wrk_handler,
 419                                         uword_to_pointer (ho_index, void *));
 420
 421   return ho_index;
 422 }
 423
 424 static u32
 425 ct_start_listen (u32 app_listener_index, transport_endpoint_t * tep)
 426 {
 427   session_endpoint_cfg_t *sep;
 428   ct_connection_t *ct;
 429
 430   sep = (session_endpoint_cfg_t *) tep;
 431   ct = ct_connection_alloc (0);
 432   ct->server_wrk = sep->app_wrk_index;
 433   ct->c_is_ip4 = sep->is_ip4;
 434   clib_memcpy (&ct->c_lcl_ip, &sep->ip, sizeof (sep->ip));
 435   ct->c_lcl_port = sep->port;
 436   ct->c_s_index = app_listener_index;
 437   ct->actual_tp = sep->transport_proto;
 438   return ct->c_c_index;
 439 }
 440
 441 static u32
 442 ct_stop_listen (u32 ct_index)
 443 {
 444   ct_connection_t *ct;
 445   ct = ct_connection_get (ct_index, 0);
 446   ct_connection_free (ct);
 447   return 0;
 448 }
 449
 450 static transport_connection_t *
 451 ct_listener_get (u32 ct_index)
 452 {
 453   return (transport_connection_t *) ct_connection_get (ct_index, 0);
 454 }
 455
 456 static transport_connection_t *
 457 ct_half_open_get (u32 ct_index)
 458 {
 459   return (transport_connection_t *) ct_connection_get (ct_index, 0);
 460 }
 461
 462 static void
 463 ct_session_cleanup (u32 conn_index, u32 thread_index)
 464 {
 465   ct_connection_t *ct, *peer_ct;
 466
 467   ct = ct_connection_get (conn_index, thread_index);
 468   if (!ct)
 469     return;
 470
 471   peer_ct = ct_connection_get (ct->peer_index, thread_index);
 472   if (peer_ct)
 473     peer_ct->peer_index = ~0;
 474
 475   ct_connection_free (ct);
 476 }
 477
 478 static void
 479 ct_cleanup_ho (u32 ho_index)
 480 {
 481   ct_connection_free (ct_connection_get (ho_index, 0));
 482 }
 483
 484 static int
 485 ct_session_connect (transport_endpoint_cfg_t * tep)
 486 {
 487   session_endpoint_cfg_t *sep_ext;
 488   session_endpoint_t _sep, *sep = &_sep;
 489   app_worker_t *app_wrk;
 490   session_handle_t lh;
 491   application_t *app;
 492   app_listener_t *al;
 493   u32 table_index;
 494   session_t *ll;
 495   u8 fib_proto;
 496
 497   sep_ext = (session_endpoint_cfg_t *) tep;
 498   _sep = *(session_endpoint_t *) tep;
 499   app_wrk = app_worker_get (sep_ext->app_wrk_index);
 500   app = application_get (app_wrk->app_index);
 501
 502   sep->transport_proto = sep_ext->original_tp;
 503   table_index = application_local_session_table (app);
 504   lh = session_lookup_local_endpoint (table_index, sep);
 505   if (lh == SESSION_DROP_HANDLE)
 506     return SESSION_E_FILTERED;
 507
 508   if (lh == SESSION_INVALID_HANDLE)
 509     goto global_scope;
 510
 511   ll = listen_session_get_from_handle (lh);
 512   al = app_listener_get_w_session (ll);
 513
 514   /*
 515    * Break loop if rule in local table points to connecting app. This
 516    * can happen if client is a generic proxy. Route connect through
 517    * global table instead.
 518    */
 519   if (al->app_index == app->app_index)
 520     goto global_scope;
 521
 522   return ct_connect (app_wrk, ll, sep_ext);
 523
 524   /*
 525    * If nothing found, check the global scope for locally attached
 526    * destinations. Make sure first that we're allowed to.
 527    */
 528
 529 global_scope:
 530   if (session_endpoint_is_local (sep))
 531     return SESSION_E_NOROUTE;
 532
 533   if (!application_has_global_scope (app))
 534     return SESSION_E_SCOPE;
 535
 536   fib_proto = session_endpoint_fib_proto (sep);
 537   table_index = session_lookup_get_index_for_fib (fib_proto, sep->fib_index);
 538   ll = session_lookup_listener_wildcard (table_index, sep);
 539
 540   /* Avoid connecting app to own listener */
 541   if (ll && ll->app_index != app->app_index)
 542     return ct_connect (app_wrk, ll, sep_ext);
 543
 544   /* Failed to connect but no error */
 545   return SESSION_E_LOCAL_CONNECT;
 546 }
 547
 548 static void
 549 ct_session_close (u32 ct_index, u32 thread_index)
 550 {
 551   ct_connection_t *ct, *peer_ct;
 552   app_worker_t *app_wrk;
 553   session_t *s;
 554
 555   ct = ct_connection_get (ct_index, thread_index);
 556   peer_ct = ct_connection_get (ct->peer_index, thread_index);
 557   if (peer_ct)
 558     {
 559       peer_ct->peer_index = ~0;
 560       /* Make sure session was allocated */
 561       if (peer_ct->flags & CT_CONN_F_HALF_OPEN)
 562         {
 563           app_wrk = app_worker_get (peer_ct->client_wrk);
 564           app_worker_connect_notify (app_wrk, 0, SESSION_E_REFUSED,
 565                                      peer_ct->client_opaque);
 566         }
 567       else if (peer_ct->c_s_index != ~0)
 568         session_transport_closing_notify (&peer_ct->connection);
 569       else
 570         ct_connection_free (peer_ct);
 571     }
 572
 573   s = session_get (ct->c_s_index, ct->c_thread_index);
 574   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
 575   if (app_wrk)
 576     app_worker_del_segment_notify (app_wrk, ct->segment_handle);
 577   session_free_w_fifos (s);
 578   if (ct->flags & CT_CONN_F_CLIENT)
 579     segment_manager_dealloc_fifos (ct->client_rx_fifo, ct->client_tx_fifo);
 580
 581   ct_connection_free (ct);
 582 }
 583
 584 static transport_connection_t *
 585 ct_session_get (u32 ct_index, u32 thread_index)
 586 {
 587   return (transport_connection_t *) ct_connection_get (ct_index,
 588                                                        thread_index);
 589 }
 590
 591 static u8 *
 592 format_ct_connection_id (u8 * s, va_list * args)
 593 {
 594   ct_connection_t *ct = va_arg (*args, ct_connection_t *);
 595   if (!ct)
 596     return s;
 597   if (ct->c_is_ip4)
 598     {
 599       s = format (s, "[%d:%d][CT:%U] %U:%d->%U:%d", ct->c_thread_index,
 600                   ct->c_s_index, format_transport_proto_short, ct->actual_tp,
 601                   format_ip4_address, &ct->c_lcl_ip4,
 602                   clib_net_to_host_u16 (ct->c_lcl_port), format_ip4_address,
 603                   &ct->c_rmt_ip4, clib_net_to_host_u16 (ct->c_rmt_port));
 604     }
 605   else
 606     {
 607       s = format (s, "[%d:%d][CT:%U] %U:%d->%U:%d", ct->c_thread_index,
 608                   ct->c_s_index, format_transport_proto_short, ct->actual_tp,
 609                   format_ip6_address, &ct->c_lcl_ip6,
 610                   clib_net_to_host_u16 (ct->c_lcl_port), format_ip6_address,
 611                   &ct->c_rmt_ip6, clib_net_to_host_u16 (ct->c_rmt_port));
 612     }
 613
 614   return s;
 615 }
 616
 617 static int
 618 ct_custom_tx (void *session, transport_send_params_t * sp)
 619 {
 620   session_t *s = (session_t *) session;
 621   if (session_has_transport (s))
 622     return 0;
 623   /* If event enqueued towards peer, remove from scheduler and remove
 624    * session tx flag, i.e., accept new tx events. Unset fifo flag now to
 625    * avoid missing events if peer did not clear fifo flag yet, which is
 626    * interpreted as successful notification and session is descheduled. */
 627   svm_fifo_unset_event (s->tx_fifo);
 628   if (!ct_session_tx (s))
 629     sp->flags = TRANSPORT_SND_F_DESCHED;
 630
 631   /* The scheduler uses packet count as a means of upper bounding the amount
 632    * of work done per dispatch. So make it look like we have sent something */
 633   return 1;
 634 }
 635
 636 static int
 637 ct_app_rx_evt (transport_connection_t * tc)
 638 {
 639   ct_connection_t *ct = (ct_connection_t *) tc, *peer_ct;
 640   session_t *ps;
 641
 642   peer_ct = ct_connection_get (ct->peer_index, tc->thread_index);
 643   if (!peer_ct)
 644     return -1;
 645   ps = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
 646   return session_dequeue_notify (ps);
 647 }
 648
 649 static u8 *
 650 format_ct_listener (u8 * s, va_list * args)
 651 {
 652   u32 tc_index = va_arg (*args, u32);
 653   u32 __clib_unused thread_index = va_arg (*args, u32);
 654   u32 __clib_unused verbose = va_arg (*args, u32);
 655   ct_connection_t *ct = ct_connection_get (tc_index, 0);
 656   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
 657   if (verbose)
 658     s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "LISTEN");
 659   return s;
 660 }
 661
 662 static u8 *
 663 format_ct_half_open (u8 *s, va_list *args)
 664 {
 665   u32 ho_index = va_arg (*args, u32);
 666   u32 verbose = va_arg (*args, u32);
 667   ct_connection_t *ct = ct_connection_get (ho_index, 0);
 668   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
 669   if (verbose)
 670     s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "HALF-OPEN");
 671   return s;
 672 }
 673
 674 static u8 *
 675 format_ct_connection (u8 * s, va_list * args)
 676 {
 677   ct_connection_t *ct = va_arg (*args, ct_connection_t *);
 678   u32 verbose = va_arg (*args, u32);
 679
 680   if (!ct)
 681     return s;
 682   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
 683   if (verbose)
 684     {
 685       s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "ESTABLISHED");
 686       if (verbose > 1)
 687         {
 688           s = format (s, "\n");
 689         }
 690     }
 691   return s;
 692 }
 693
 694 static u8 *
 695 format_ct_session (u8 * s, va_list * args)
 696 {
 697   u32 ct_index = va_arg (*args, u32);
 698   u32 thread_index = va_arg (*args, u32);
 699   u32 verbose = va_arg (*args, u32);
 700   ct_connection_t *ct;
 701
 702   ct = ct_connection_get (ct_index, thread_index);
 703   if (!ct)
 704     {
 705       s = format (s, "empty\n");
 706       return s;
 707     }
 708
 709   s = format (s, "%U", format_ct_connection, ct, verbose);
 710   return s;
 711 }
 712
 713 clib_error_t *
 714 ct_enable_disable (vlib_main_t * vm, u8 is_en)
 715 {
 716   ct_main_t *cm = &ct_main;
 717
 718   cm->n_workers = vlib_num_workers ();
 719   vec_validate (cm->connections, cm->n_workers);
 720   clib_spinlock_init (&cm->ho_reuseable_lock);
 721
 722   return 0;
 723 }
 724
 725 /* *INDENT-OFF* */
 726 static const transport_proto_vft_t cut_thru_proto = {
 727   .enable = ct_enable_disable,
 728   .start_listen = ct_start_listen,
 729   .stop_listen = ct_stop_listen,
 730   .get_connection = ct_session_get,
 731   .get_listener = ct_listener_get,
 732   .get_half_open = ct_half_open_get,
 733   .cleanup = ct_session_cleanup,
 734   .cleanup_ho = ct_cleanup_ho,
 735   .connect = ct_session_connect,
 736   .close = ct_session_close,
 737   .custom_tx = ct_custom_tx,
 738   .app_rx_evt = ct_app_rx_evt,
 739   .format_listener = format_ct_listener,
 740   .format_half_open = format_ct_half_open,
 741   .format_connection = format_ct_session,
 742   .transport_options = {
 743     .name = "ct",
 744     .short_name = "C",
 745     .tx_type = TRANSPORT_TX_INTERNAL,
 746     .service_type = TRANSPORT_SERVICE_VC,
 747   },
 748 };
 749 /* *INDENT-ON* */
 750
 751 int
 752 ct_session_tx (session_t * s)
 753 {
 754   ct_connection_t *ct, *peer_ct;
 755   session_t *peer_s;
 756
 757   ct = (ct_connection_t *) session_get_transport (s);
 758   peer_ct = ct_connection_get (ct->peer_index, ct->c_thread_index);
 759   if (!peer_ct)
 760     return 0;
 761   peer_s = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
 762   if (peer_s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
 763     return 0;
 764   return session_enqueue_notify (peer_s);
 765 }
 766
 767 static clib_error_t *
 768 ct_transport_init (vlib_main_t * vm)
 769 {
 770   transport_register_protocol (TRANSPORT_PROTO_NONE, &cut_thru_proto,
 771                                FIB_PROTOCOL_IP4, ~0);
 772   transport_register_protocol (TRANSPORT_PROTO_NONE, &cut_thru_proto,
 773                                FIB_PROTOCOL_IP6, ~0);
 774   return 0;
 775 }
 776
 777 VLIB_INIT_FUNCTION (ct_transport_init);
 778
 779 /*
 780  * fd.io coding-style-patch-verification: ON
 781  *
 782  * Local Variables:
 783  * eval: (c-set-style "gnu")
 784  * End:
 785  */