src/vnet/session/session.c

   1 /*
   2  * Copyright (c) 2017 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  * @file
  17  * @brief Session and session manager
  18  */
  19
  20 #include <vnet/session/session.h>
  21 #include <vlibmemory/api.h>
  22 #include <vnet/dpo/load_balance.h>
  23 #include <vnet/fib/ip4_fib.h>
  24 #include <vnet/session/application.h>
  25 #include <vnet/tcp/tcp.h>
  26 #include <vnet/session/session_debug.h>
  27
  28 /**
  29  * Per-type vector of transport protocol virtual function tables
  30  */
  31 static transport_proto_vft_t *tp_vfts;
  32
  33 session_manager_main_t session_manager_main;
  34
  35 /*
  36  * Session lookup key; (src-ip, dst-ip, src-port, dst-port, session-type)
  37  * Value: (owner thread index << 32 | session_index);
  38  */
  39 void
  40 stream_session_table_add_for_tc (transport_connection_t * tc, u64 value)
  41 {
  42   session_manager_main_t *smm = &session_manager_main;
  43   session_kv4_t kv4;
  44   session_kv6_t kv6;
  45
  46   switch (tc->proto)
  47     {
  48     case SESSION_TYPE_IP4_UDP:
  49     case SESSION_TYPE_IP4_TCP:
  50       make_v4_ss_kv_from_tc (&kv4, tc);
  51       kv4.value = value;
  52       clib_bihash_add_del_16_8 (&smm->v4_session_hash, &kv4, 1 /* is_add */ );
  53       break;
  54     case SESSION_TYPE_IP6_UDP:
  55     case SESSION_TYPE_IP6_TCP:
  56       make_v6_ss_kv_from_tc (&kv6, tc);
  57       kv6.value = value;
  58       clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6, 1 /* is_add */ );
  59       break;
  60     default:
  61       clib_warning ("Session type not supported");
  62       ASSERT (0);
  63     }
  64 }
  65
  66 void
  67 stream_session_table_add (session_manager_main_t * smm, stream_session_t * s,
  68                           u64 value)
  69 {
  70   transport_connection_t *tc;
  71
  72   tc = tp_vfts[s->session_type].get_connection (s->connection_index,
  73                                                 s->thread_index);
  74   stream_session_table_add_for_tc (tc, value);
  75 }
  76
  77 static void
  78 stream_session_half_open_table_add (session_type_t sst,
  79                                     transport_connection_t * tc, u64 value)
  80 {
  81   session_manager_main_t *smm = &session_manager_main;
  82   session_kv4_t kv4;
  83   session_kv6_t kv6;
  84
  85   switch (sst)
  86     {
  87     case SESSION_TYPE_IP4_UDP:
  88     case SESSION_TYPE_IP4_TCP:
  89       make_v4_ss_kv_from_tc (&kv4, tc);
  90       kv4.value = value;
  91       clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4,
  92                                 1 /* is_add */ );
  93       break;
  94     case SESSION_TYPE_IP6_UDP:
  95     case SESSION_TYPE_IP6_TCP:
  96       make_v6_ss_kv_from_tc (&kv6, tc);
  97       kv6.value = value;
  98       clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6,
  99                                 1 /* is_add */ );
 100       break;
 101     default:
 102       clib_warning ("Session type not supported");
 103       ASSERT (0);
 104     }
 105 }
 106
 107 int
 108 stream_session_table_del_for_tc (transport_connection_t * tc)
 109 {
 110   session_manager_main_t *smm = &session_manager_main;
 111   session_kv4_t kv4;
 112   session_kv6_t kv6;
 113   switch (tc->proto)
 114     {
 115     case SESSION_TYPE_IP4_UDP:
 116     case SESSION_TYPE_IP4_TCP:
 117       make_v4_ss_kv_from_tc (&kv4, tc);
 118       return clib_bihash_add_del_16_8 (&smm->v4_session_hash, &kv4,
 119                                        0 /* is_add */ );
 120       break;
 121     case SESSION_TYPE_IP6_UDP:
 122     case SESSION_TYPE_IP6_TCP:
 123       make_v6_ss_kv_from_tc (&kv6, tc);
 124       return clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6,
 125                                        0 /* is_add */ );
 126       break;
 127     default:
 128       clib_warning ("Session type not supported");
 129       ASSERT (0);
 130     }
 131
 132   return 0;
 133 }
 134
 135 static int
 136 stream_session_table_del (session_manager_main_t * smm, stream_session_t * s)
 137 {
 138   transport_connection_t *ts;
 139
 140   ts = tp_vfts[s->session_type].get_connection (s->connection_index,
 141                                                 s->thread_index);
 142   return stream_session_table_del_for_tc (ts);
 143 }
 144
 145 static void
 146 stream_session_half_open_table_del (session_manager_main_t * smm, u8 sst,
 147                                     transport_connection_t * tc)
 148 {
 149   session_kv4_t kv4;
 150   session_kv6_t kv6;
 151
 152   switch (sst)
 153     {
 154     case SESSION_TYPE_IP4_UDP:
 155     case SESSION_TYPE_IP4_TCP:
 156       make_v4_ss_kv_from_tc (&kv4, tc);
 157       clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4,
 158                                 0 /* is_add */ );
 159       break;
 160     case SESSION_TYPE_IP6_UDP:
 161     case SESSION_TYPE_IP6_TCP:
 162       make_v6_ss_kv_from_tc (&kv6, tc);
 163       clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6,
 164                                 0 /* is_add */ );
 165       break;
 166     default:
 167       clib_warning ("Session type not supported");
 168       ASSERT (0);
 169     }
 170 }
 171
 172 stream_session_t *
 173 stream_session_lookup_listener4 (ip4_address_t * lcl, u16 lcl_port, u8 proto)
 174 {
 175   session_manager_main_t *smm = &session_manager_main;
 176   session_kv4_t kv4;
 177   int rv;
 178
 179   make_v4_listener_kv (&kv4, lcl, lcl_port, proto);
 180   rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
 181   if (rv == 0)
 182     return pool_elt_at_index (smm->listen_sessions[proto], (u32) kv4.value);
 183
 184   /* Zero out the lcl ip */
 185   kv4.key[0] = 0;
 186   rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
 187   if (rv == 0)
 188     return pool_elt_at_index (smm->listen_sessions[proto], kv4.value);
 189
 190   return 0;
 191 }
 192
 193 /** Looks up a session based on the 5-tuple passed as argument.
 194  *
 195  * First it tries to find an established session, if this fails, it tries
 196  * finding a listener session if this fails, it tries a lookup with a
 197  * wildcarded local source (listener bound to all interfaces)
 198  */
 199 stream_session_t *
 200 stream_session_lookup4 (ip4_address_t * lcl, ip4_address_t * rmt,
 201                         u16 lcl_port, u16 rmt_port, u8 proto)
 202 {
 203   session_manager_main_t *smm = &session_manager_main;
 204   session_kv4_t kv4;
 205   stream_session_t *s;
 206   int rv;
 207
 208   /* Lookup session amongst established ones */
 209   make_v4_ss_kv (&kv4, lcl, rmt, lcl_port, rmt_port, proto);
 210   rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
 211   if (rv == 0)
 212     return stream_session_get_from_handle (kv4.value);
 213
 214   /* If nothing is found, check if any listener is available */
 215   if ((s = stream_session_lookup_listener4 (lcl, lcl_port, proto)))
 216     return s;
 217
 218   /* Finally, try half-open connections */
 219   rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
 220   if (rv == 0)
 221     return stream_session_get_from_handle (kv4.value);
 222   return 0;
 223 }
 224
 225 stream_session_t *
 226 stream_session_lookup_listener6 (ip6_address_t * lcl, u16 lcl_port, u8 proto)
 227 {
 228   session_manager_main_t *smm = &session_manager_main;
 229   session_kv6_t kv6;
 230   int rv;
 231
 232   make_v6_listener_kv (&kv6, lcl, lcl_port, proto);
 233   rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
 234   if (rv == 0)
 235     return pool_elt_at_index (smm->listen_sessions[proto], kv6.value);
 236
 237   /* Zero out the lcl ip */
 238   kv6.key[0] = kv6.key[1] = 0;
 239   rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
 240   if (rv == 0)
 241     return pool_elt_at_index (smm->listen_sessions[proto], kv6.value);
 242
 243   return 0;
 244 }
 245
 246 /* Looks up a session based on the 5-tuple passed as argument.
 247  * First it tries to find an established session, if this fails, it tries
 248  * finding a listener session if this fails, it tries a lookup with a
 249  * wildcarded local source (listener bound to all interfaces) */
 250 stream_session_t *
 251 stream_session_lookup6 (ip6_address_t * lcl, ip6_address_t * rmt,
 252                         u16 lcl_port, u16 rmt_port, u8 proto)
 253 {
 254   session_manager_main_t *smm = vnet_get_session_manager_main ();
 255   session_kv6_t kv6;
 256   stream_session_t *s;
 257   int rv;
 258
 259   make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
 260   rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
 261   if (rv == 0)
 262     return stream_session_get_from_handle (kv6.value);
 263
 264   /* If nothing is found, check if any listener is available */
 265   if ((s = stream_session_lookup_listener6 (lcl, lcl_port, proto)))
 266     return s;
 267
 268   /* Finally, try half-open connections */
 269   rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
 270   if (rv == 0)
 271     return stream_session_get_from_handle (kv6.value);
 272   return 0;
 273 }
 274
 275 stream_session_t *
 276 stream_session_lookup_listener (ip46_address_t * lcl, u16 lcl_port, u8 proto)
 277 {
 278   switch (proto)
 279     {
 280     case SESSION_TYPE_IP4_UDP:
 281     case SESSION_TYPE_IP4_TCP:
 282       return stream_session_lookup_listener4 (&lcl->ip4, lcl_port, proto);
 283       break;
 284     case SESSION_TYPE_IP6_UDP:
 285     case SESSION_TYPE_IP6_TCP:
 286       return stream_session_lookup_listener6 (&lcl->ip6, lcl_port, proto);
 287       break;
 288     }
 289   return 0;
 290 }
 291
 292 static u64
 293 stream_session_half_open_lookup (session_manager_main_t * smm,
 294                                  ip46_address_t * lcl, ip46_address_t * rmt,
 295                                  u16 lcl_port, u16 rmt_port, u8 proto)
 296 {
 297   session_kv4_t kv4;
 298   session_kv6_t kv6;
 299   int rv;
 300
 301   switch (proto)
 302     {
 303     case SESSION_TYPE_IP4_UDP:
 304     case SESSION_TYPE_IP4_TCP:
 305       make_v4_ss_kv (&kv4, &lcl->ip4, &rmt->ip4, lcl_port, rmt_port, proto);
 306       rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
 307
 308       if (rv == 0)
 309         return kv4.value;
 310
 311       return (u64) ~ 0;
 312       break;
 313     case SESSION_TYPE_IP6_UDP:
 314     case SESSION_TYPE_IP6_TCP:
 315       make_v6_ss_kv (&kv6, &lcl->ip6, &rmt->ip6, lcl_port, rmt_port, proto);
 316       rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
 317
 318       if (rv == 0)
 319         return kv6.value;
 320
 321       return (u64) ~ 0;
 322       break;
 323     }
 324   return 0;
 325 }
 326
 327 transport_connection_t *
 328 stream_session_lookup_transport_wt4 (ip4_address_t * lcl, ip4_address_t * rmt,
 329                                      u16 lcl_port, u16 rmt_port, u8 proto,
 330                                      u32 my_thread_index)
 331 {
 332   session_manager_main_t *smm = &session_manager_main;
 333   session_kv4_t kv4;
 334   stream_session_t *s;
 335   int rv;
 336
 337   /* Lookup session amongst established ones */
 338   make_v4_ss_kv (&kv4, lcl, rmt, lcl_port, rmt_port, proto);
 339   rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
 340   if (rv == 0)
 341     {
 342       s = stream_session_get_tsi (kv4.value, my_thread_index);
 343
 344       return tp_vfts[s->session_type].get_connection (s->connection_index,
 345                                                       my_thread_index);
 346     }
 347
 348   /* If nothing is found, check if any listener is available */
 349   s = stream_session_lookup_listener4 (lcl, lcl_port, proto);
 350   if (s)
 351     return tp_vfts[s->session_type].get_listener (s->connection_index);
 352
 353   /* Finally, try half-open connections */
 354   rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
 355   if (rv == 0)
 356     return tp_vfts[proto].get_half_open (kv4.value & 0xFFFFFFFF);
 357   return 0;
 358 }
 359
 360 transport_connection_t *
 361 stream_session_lookup_transport4 (ip4_address_t * lcl, ip4_address_t * rmt,
 362                                   u16 lcl_port, u16 rmt_port, u8 proto)
 363 {
 364   session_manager_main_t *smm = &session_manager_main;
 365   session_kv4_t kv4;
 366   stream_session_t *s;
 367   int rv;
 368
 369   /* Lookup session amongst established ones */
 370   make_v4_ss_kv (&kv4, lcl, rmt, lcl_port, rmt_port, proto);
 371   rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
 372   if (rv == 0)
 373     {
 374       s = stream_session_get_from_handle (kv4.value);
 375       return tp_vfts[s->session_type].get_connection (s->connection_index,
 376                                                       s->thread_index);
 377     }
 378
 379   /* If nothing is found, check if any listener is available */
 380   s = stream_session_lookup_listener4 (lcl, lcl_port, proto);
 381   if (s)
 382     return tp_vfts[s->session_type].get_listener (s->connection_index);
 383
 384   /* Finally, try half-open connections */
 385   rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
 386   if (rv == 0)
 387     return tp_vfts[proto].get_half_open (kv4.value & 0xFFFFFFFF);
 388   return 0;
 389 }
 390
 391 transport_connection_t *
 392 stream_session_lookup_transport_wt6 (ip6_address_t * lcl, ip6_address_t * rmt,
 393                                      u16 lcl_port, u16 rmt_port, u8 proto,
 394                                      u32 my_thread_index)
 395 {
 396   session_manager_main_t *smm = &session_manager_main;
 397   stream_session_t *s;
 398   session_kv6_t kv6;
 399   int rv;
 400
 401   make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
 402   rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
 403   if (rv == 0)
 404     {
 405       s = stream_session_get_tsi (kv6.value, my_thread_index);
 406
 407       return tp_vfts[s->session_type].get_connection (s->connection_index,
 408                                                       my_thread_index);
 409     }
 410
 411   /* If nothing is found, check if any listener is available */
 412   s = stream_session_lookup_listener6 (lcl, lcl_port, proto);
 413   if (s)
 414     return tp_vfts[s->session_type].get_listener (s->connection_index);
 415
 416   /* Finally, try half-open connections */
 417   rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
 418   if (rv == 0)
 419     return tp_vfts[proto].get_half_open (kv6.value & 0xFFFFFFFF);
 420
 421   return 0;
 422 }
 423
 424 transport_connection_t *
 425 stream_session_lookup_transport6 (ip6_address_t * lcl, ip6_address_t * rmt,
 426                                   u16 lcl_port, u16 rmt_port, u8 proto)
 427 {
 428   session_manager_main_t *smm = &session_manager_main;
 429   stream_session_t *s;
 430   session_kv6_t kv6;
 431   int rv;
 432
 433   make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
 434   rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
 435   if (rv == 0)
 436     {
 437       s = stream_session_get_from_handle (kv6.value);
 438       return tp_vfts[s->session_type].get_connection (s->connection_index,
 439                                                       s->thread_index);
 440     }
 441
 442   /* If nothing is found, check if any listener is available */
 443   s = stream_session_lookup_listener6 (lcl, lcl_port, proto);
 444   if (s)
 445     return tp_vfts[s->session_type].get_listener (s->connection_index);
 446
 447   /* Finally, try half-open connections */
 448   rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
 449   if (rv == 0)
 450     return tp_vfts[proto].get_half_open (kv6.value & 0xFFFFFFFF);
 451
 452   return 0;
 453 }
 454
 455 int
 456 stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
 457                          stream_session_t ** ret_s)
 458 {
 459   session_manager_main_t *smm = &session_manager_main;
 460   svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
 461   u32 fifo_segment_index;
 462   u32 pool_index;
 463   stream_session_t *s;
 464   u64 value;
 465   u32 thread_index = tc->thread_index;
 466   int rv;
 467
 468   ASSERT (thread_index == vlib_get_thread_index ());
 469
 470   if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo,
 471                                                  &server_tx_fifo,
 472                                                  &fifo_segment_index)))
 473     return rv;
 474
 475   /* Create the session */
 476   pool_get_aligned (smm->sessions[thread_index], s, CLIB_CACHE_LINE_BYTES);
 477   memset (s, 0, sizeof (*s));
 478
 479   /* Initialize backpointers */
 480   pool_index = s - smm->sessions[thread_index];
 481   server_rx_fifo->master_session_index = pool_index;
 482   server_rx_fifo->master_thread_index = thread_index;
 483
 484   server_tx_fifo->master_session_index = pool_index;
 485   server_tx_fifo->master_thread_index = thread_index;
 486
 487   s->server_rx_fifo = server_rx_fifo;
 488   s->server_tx_fifo = server_tx_fifo;
 489
 490   /* Initialize state machine, such as it is... */
 491   s->session_type = tc->proto;
 492   s->session_state = SESSION_STATE_CONNECTING;
 493   s->svm_segment_index = fifo_segment_index;
 494   s->thread_index = thread_index;
 495   s->session_index = pool_index;
 496
 497   /* Attach transport to session */
 498   s->connection_index = tc->c_index;
 499
 500   /* Attach session to transport */
 501   tc->s_index = s->session_index;
 502
 503   /* Add to the main lookup table */
 504   value = (((u64) thread_index) << 32) | (u64) s->session_index;
 505   stream_session_table_add_for_tc (tc, value);
 506
 507   *ret_s = s;
 508
 509   return 0;
 510 }
 511
 512 /** Enqueue buffer chain tail */
 513 always_inline int
 514 session_enqueue_chain_tail (stream_session_t * s, vlib_buffer_t * b,
 515                             u32 offset, u8 is_in_order)
 516 {
 517   vlib_buffer_t *chain_b;
 518   u32 chain_bi = b->next_buffer;
 519   vlib_main_t *vm = vlib_get_main ();
 520   u8 *data, len;
 521   u16 written = 0;
 522   int rv = 0;
 523
 524   do
 525     {
 526       chain_b = vlib_get_buffer (vm, chain_bi);
 527       data = vlib_buffer_get_current (chain_b);
 528       len = chain_b->current_length;
 529       if (is_in_order)
 530         {
 531           rv = svm_fifo_enqueue_nowait (s->server_rx_fifo, len, data);
 532           if (rv < len)
 533             {
 534               return (rv > 0) ? (written + rv) : written;
 535             }
 536           written += rv;
 537         }
 538       else
 539         {
 540           rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset, len,
 541                                              data);
 542           if (rv)
 543             return -1;
 544           offset += len;
 545         }
 546     }
 547   while ((chain_bi = (chain_b->flags & VLIB_BUFFER_NEXT_PRESENT)
 548           ? chain_b->next_buffer : 0));
 549
 550   if (is_in_order)
 551     return written;
 552
 553   return 0;
 554 }
 555
 556 /*
 557  * Enqueue data for delivery to session peer. Does not notify peer of enqueue
 558  * event but on request can queue notification events for later delivery by
 559  * calling stream_server_flush_enqueue_events().
 560  *
 561  * @param tc Transport connection which is to be enqueued data
 562  * @param b Buffer to be enqueued
 563  * @param offset Offset at which to start enqueueing if out-of-order
 564  * @param queue_event Flag to indicate if peer is to be notified or if event
 565  *                    is to be queued. The former is useful when more data is
 566  *                    enqueued and only one event is to be generated.
 567  * @param is_in_order Flag to indicate if data is in order
 568  * @return Number of bytes enqueued or a negative value if enqueueing failed.
 569  */
 570 int
 571 stream_session_enqueue_data (transport_connection_t * tc, vlib_buffer_t * b,
 572                              u32 offset, u8 queue_event, u8 is_in_order)
 573 {
 574   stream_session_t *s;
 575   int enqueued = 0, rv;
 576
 577   s = stream_session_get (tc->s_index, tc->thread_index);
 578
 579   if (is_in_order)
 580     {
 581       enqueued =
 582         svm_fifo_enqueue_nowait (s->server_rx_fifo, b->current_length,
 583                                  vlib_buffer_get_current (b));
 584       if (PREDICT_FALSE
 585           ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && enqueued > 0))
 586         {
 587           rv = session_enqueue_chain_tail (s, b, 0, 1);
 588           if (rv <= 0)
 589             return enqueued;
 590           enqueued += rv;
 591         }
 592     }
 593   else
 594     {
 595       rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset,
 596                                          b->current_length,
 597                                          vlib_buffer_get_current (b));
 598       if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && !rv))
 599         rv = session_enqueue_chain_tail (s, b, offset + b->current_length, 0);
 600       if (rv)
 601         return -1;
 602     }
 603
 604   if (queue_event)
 605     {
 606       /* Queue RX event on this fifo. Eventually these will need to be flushed
 607        * by calling stream_server_flush_enqueue_events () */
 608       session_manager_main_t *smm = vnet_get_session_manager_main ();
 609       u32 thread_index = s->thread_index;
 610       u32 my_enqueue_epoch = smm->current_enqueue_epoch[thread_index];
 611
 612       if (s->enqueue_epoch != my_enqueue_epoch)
 613         {
 614           s->enqueue_epoch = my_enqueue_epoch;
 615           vec_add1 (smm->session_indices_to_enqueue_by_thread[thread_index],
 616                     s - smm->sessions[thread_index]);
 617         }
 618     }
 619
 620   if (is_in_order)
 621     return enqueued;
 622
 623   return 0;
 624 }
 625
 626 /** Check if we have space in rx fifo to push more bytes */
 627 u8
 628 stream_session_no_space (transport_connection_t * tc, u32 thread_index,
 629                          u16 data_len)
 630 {
 631   stream_session_t *s = stream_session_get (tc->s_index, thread_index);
 632
 633   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY))
 634     return 1;
 635
 636   if (data_len > svm_fifo_max_enqueue (s->server_rx_fifo))
 637     return 1;
 638
 639   return 0;
 640 }
 641
 642 u32
 643 stream_session_tx_fifo_max_dequeue (transport_connection_t * tc)
 644 {
 645   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
 646   if (s->session_state != SESSION_STATE_READY)
 647     return 0;
 648   return svm_fifo_max_dequeue (s->server_tx_fifo);
 649 }
 650
 651 int
 652 stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
 653                            u32 offset, u32 max_bytes)
 654 {
 655   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
 656   return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
 657 }
 658
 659 u32
 660 stream_session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
 661 {
 662   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
 663   return svm_fifo_dequeue_drop (s->server_tx_fifo, max_bytes);
 664 }
 665
 666 /**
 667  * Notify session peer that new data has been enqueued.
 668  *
 669  * @param s Stream session for which the event is to be generated.
 670  * @param block Flag to indicate if call should block if event queue is full.
 671  *
 672  * @return 0 on succes or negative number if failed to send notification.
 673  */
 674 static int
 675 stream_session_enqueue_notify (stream_session_t * s, u8 block)
 676 {
 677   application_t *app;
 678   session_fifo_event_t evt;
 679   unix_shared_memory_queue_t *q;
 680   static u32 serial_number;
 681
 682   if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED))
 683     return 0;
 684
 685   /* Get session's server */
 686   app = application_get (s->app_index);
 687
 688   /* Built-in server? Hand event to the callback... */
 689   if (app->cb_fns.builtin_server_rx_callback)
 690     return app->cb_fns.builtin_server_rx_callback (s);
 691
 692   /* If no event, send one */
 693   if (svm_fifo_set_event (s->server_rx_fifo))
 694     {
 695       /* Fabricate event */
 696       evt.fifo = s->server_rx_fifo;
 697       evt.event_type = FIFO_EVENT_APP_RX;
 698       evt.event_id = serial_number++;
 699
 700       /* Add event to server's event queue */
 701       q = app->event_queue;
 702
 703       /* Based on request block (or not) for lack of space */
 704       if (block || PREDICT_TRUE (q->cursize < q->maxsize))
 705         unix_shared_memory_queue_add (app->event_queue, (u8 *) & evt,
 706                                       0 /* do wait for mutex */ );
 707       else
 708         {
 709           clib_warning ("fifo full");
 710           return -1;
 711         }
 712     }
 713
 714   /* *INDENT-OFF* */
 715   SESSION_EVT_DBG(SESSION_EVT_ENQ, s, ({
 716       ed->data[0] = evt.event_id;
 717       ed->data[1] = svm_fifo_max_dequeue (s->server_rx_fifo);
 718   }));
 719   /* *INDENT-ON* */
 720
 721   return 0;
 722 }
 723
 724 /**
 725  * Flushes queue of sessions that are to be notified of new data
 726  * enqueued events.
 727  *
 728  * @param thread_index Thread index for which the flush is to be performed.
 729  * @return 0 on success or a positive number indicating the number of
 730  *         failures due to API queue being full.
 731  */
 732 int
 733 session_manager_flush_enqueue_events (u32 thread_index)
 734 {
 735   session_manager_main_t *smm = &session_manager_main;
 736   u32 *session_indices_to_enqueue;
 737   int i, errors = 0;
 738
 739   session_indices_to_enqueue =
 740     smm->session_indices_to_enqueue_by_thread[thread_index];
 741
 742   for (i = 0; i < vec_len (session_indices_to_enqueue); i++)
 743     {
 744       stream_session_t *s0;
 745
 746       /* Get session */
 747       s0 = stream_session_get (session_indices_to_enqueue[i], thread_index);
 748       if (stream_session_enqueue_notify (s0, 0 /* don't block */ ))
 749         {
 750           errors++;
 751         }
 752     }
 753
 754   vec_reset_length (session_indices_to_enqueue);
 755
 756   smm->session_indices_to_enqueue_by_thread[thread_index] =
 757     session_indices_to_enqueue;
 758
 759   /* Increment enqueue epoch for next round */
 760   smm->current_enqueue_epoch[thread_index]++;
 761
 762   return errors;
 763 }
 764
 765 /**
 766  * Init fifo tail and head pointers
 767  *
 768  * Useful if transport uses absolute offsets for tracking ooo segments.
 769  */
 770 void
 771 stream_session_init_fifos_pointers (transport_connection_t * tc,
 772                                     u32 rx_pointer, u32 tx_pointer)
 773 {
 774   stream_session_t *s;
 775   s = stream_session_get (tc->s_index, tc->thread_index);
 776   svm_fifo_init_pointers (s->server_rx_fifo, rx_pointer);
 777   svm_fifo_init_pointers (s->server_tx_fifo, tx_pointer);
 778 }
 779
 780 int
 781 stream_session_connect_notify (transport_connection_t * tc, u8 sst,
 782                                u8 is_fail)
 783 {
 784   session_manager_main_t *smm = &session_manager_main;
 785   application_t *app;
 786   stream_session_t *new_s = 0;
 787   u64 handle;
 788   u32 api_context = 0;
 789   int error = 0;
 790
 791   handle = stream_session_half_open_lookup (smm, &tc->lcl_ip, &tc->rmt_ip,
 792                                             tc->lcl_port, tc->rmt_port,
 793                                             tc->proto);
 794   if (handle == HALF_OPEN_LOOKUP_INVALID_VALUE)
 795     {
 796       clib_warning ("This can't be good!");
 797       return -1;
 798     }
 799
 800   /* Get the app's index from the handle we stored when opening connection */
 801   app = application_get (handle >> 32);
 802   api_context = tc->s_index;
 803
 804   if (!is_fail)
 805     {
 806       segment_manager_t *sm;
 807       sm = application_get_connect_segment_manager (app);
 808
 809       /* Create new session (svm segments are allocated if needed) */
 810       if (stream_session_create_i (sm, tc, &new_s))
 811         {
 812           is_fail = 1;
 813           error = -1;
 814         }
 815       else
 816         new_s->app_index = app->index;
 817     }
 818
 819   /* Notify client */
 820   app->cb_fns.session_connected_callback (app->index, api_context, new_s,
 821                                           is_fail);
 822
 823   /* Cleanup session lookup */
 824   stream_session_half_open_table_del (smm, sst, tc);
 825
 826   return error;
 827 }
 828
 829 void
 830 stream_session_accept_notify (transport_connection_t * tc)
 831 {
 832   application_t *server;
 833   stream_session_t *s;
 834
 835   s = stream_session_get (tc->s_index, tc->thread_index);
 836   server = application_get (s->app_index);
 837   server->cb_fns.session_accept_callback (s);
 838 }
 839
 840 /**
 841  * Notification from transport that connection is being closed.
 842  *
 843  * A disconnect is sent to application but state is not removed. Once
 844  * disconnect is acknowledged by application, session disconnect is called.
 845  * Ultimately this leads to close being called on transport (passive close).
 846  */
 847 void
 848 stream_session_disconnect_notify (transport_connection_t * tc)
 849 {
 850   application_t *server;
 851   stream_session_t *s;
 852
 853   s = stream_session_get (tc->s_index, tc->thread_index);
 854   server = application_get (s->app_index);
 855   server->cb_fns.session_disconnect_callback (s);
 856 }
 857
 858 /**
 859  * Cleans up session and associated app if needed.
 860  */
 861 void
 862 stream_session_delete (stream_session_t * s)
 863 {
 864   session_manager_main_t *smm = vnet_get_session_manager_main ();
 865
 866   /* Delete from the main lookup table. */
 867   stream_session_table_del (smm, s);
 868
 869   /* Cleanup fifo segments */
 870   segment_manager_dealloc_fifos (s->svm_segment_index, s->server_rx_fifo,
 871                                  s->server_tx_fifo);
 872
 873   pool_put (smm->sessions[s->thread_index], s);
 874 }
 875
 876 /**
 877  * Notification from transport that connection is being deleted
 878  *
 879  * This should be called only on previously fully established sessions. For
 880  * instance failed connects should call stream_session_connect_notify and
 881  * indicate that the connect has failed.
 882  */
 883 void
 884 stream_session_delete_notify (transport_connection_t * tc)
 885 {
 886   stream_session_t *s;
 887
 888   /* App might've been removed already */
 889   s = stream_session_get_if_valid (tc->s_index, tc->thread_index);
 890   if (!s)
 891     {
 892       return;
 893     }
 894   stream_session_delete (s);
 895 }
 896
 897 /**
 898  * Notify application that connection has been reset.
 899  */
 900 void
 901 stream_session_reset_notify (transport_connection_t * tc)
 902 {
 903   stream_session_t *s;
 904   application_t *app;
 905   s = stream_session_get (tc->s_index, tc->thread_index);
 906
 907   app = application_get (s->app_index);
 908   app->cb_fns.session_reset_callback (s);
 909 }
 910
 911 /**
 912  * Accept a stream session. Optionally ping the server by callback.
 913  */
 914 int
 915 stream_session_accept (transport_connection_t * tc, u32 listener_index,
 916                        u8 sst, u8 notify)
 917 {
 918   application_t *server;
 919   stream_session_t *s, *listener;
 920   segment_manager_t *sm;
 921
 922   int rv;
 923
 924   /* Find the server */
 925   listener = listen_session_get (sst, listener_index);
 926   server = application_get (listener->app_index);
 927
 928   sm = application_get_listen_segment_manager (server, listener);
 929   if ((rv = stream_session_create_i (sm, tc, &s)))
 930     return rv;
 931
 932   s->app_index = server->index;
 933   s->listener_index = listener_index;
 934   s->session_state = SESSION_STATE_ACCEPTING;
 935
 936   /* Shoulder-tap the server */
 937   if (notify)
 938     {
 939       server->cb_fns.session_accept_callback (s);
 940     }
 941
 942   return 0;
 943 }
 944
 945 /**
 946  * Ask transport to open connection to remote transport endpoint.
 947  *
 948  * Stores handle for matching request with reply since the call can be
 949  * asynchronous. For instance, for TCP the 3-way handshake must complete
 950  * before reply comes. Session is only created once connection is established.
 951  *
 952  * @param app_index Index of the application requesting the connect
 953  * @param st Session type requested.
 954  * @param tep Remote transport endpoint
 955  * @param res Resulting transport connection .
 956  */
 957 int
 958 stream_session_open (u32 app_index, session_type_t st,
 959                      transport_endpoint_t * tep,
 960                      transport_connection_t ** res)
 961 {
 962   transport_connection_t *tc;
 963   int rv;
 964   u64 handle;
 965
 966   rv = tp_vfts[st].open (&tep->ip, tep->port);
 967   if (rv < 0)
 968     {
 969       clib_warning ("Transport failed to open connection.");
 970       return VNET_API_ERROR_SESSION_CONNECT_FAIL;
 971     }
 972
 973   tc = tp_vfts[st].get_half_open ((u32) rv);
 974
 975   /* Save app and tc index. The latter is needed to help establish the
 976    * connection while the former is needed when the connect notify comes
 977    * and we have to notify the external app */
 978   handle = (((u64) app_index) << 32) | (u64) tc->c_index;
 979
 980   /* Add to the half-open lookup table */
 981   stream_session_half_open_table_add (st, tc, handle);
 982
 983   *res = tc;
 984
 985   return 0;
 986 }
 987
 988 /**
 989  * Ask transport to listen on local transport endpoint.
 990  *
 991  * @param s Session for which listen will be called. Note that unlike
 992  *          established sessions, listen sessions are not associated to a
 993  *          thread.
 994  * @param tep Local endpoint to be listened on.
 995  */
 996 int
 997 stream_session_listen (stream_session_t * s, transport_endpoint_t * tep)
 998 {
 999   transport_connection_t *tc;
1000   u32 tci;
1001
1002   /* Transport bind/listen  */
1003   tci = tp_vfts[s->session_type].bind (s->session_index, &tep->ip, tep->port);
1004
1005   if (tci == (u32) ~ 0)
1006     return -1;
1007
1008   /* Attach transport to session */
1009   s->connection_index = tci;
1010   tc = tp_vfts[s->session_type].get_listener (tci);
1011
1012   /* Weird but handle it ... */
1013   if (tc == 0)
1014     return -1;
1015
1016   /* Add to the main lookup table */
1017   stream_session_table_add_for_tc (tc, s->session_index);
1018
1019   return 0;
1020 }
1021
1022 /**
1023  * Ask transport to stop listening on local transport endpoint.
1024  *
1025  * @param s Session to stop listening on. It must be in state LISTENING.
1026  */
1027 int
1028 stream_session_stop_listen (stream_session_t * s)
1029 {
1030   transport_connection_t *tc;
1031
1032   if (s->session_state != SESSION_STATE_LISTENING)
1033     {
1034       clib_warning ("not a listening session");
1035       return -1;
1036     }
1037
1038   tc = tp_vfts[s->session_type].get_listener (s->connection_index);
1039   if (!tc)
1040     {
1041       clib_warning ("no transport");
1042       return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
1043     }
1044
1045   stream_session_table_del_for_tc (tc);
1046   tp_vfts[s->session_type].unbind (s->connection_index);
1047   return 0;
1048 }
1049
1050 void
1051 session_send_session_evt_to_thread (u64 session_handle,
1052                                     fifo_event_type_t evt_type,
1053                                     u32 thread_index)
1054 {
1055   static u16 serial_number = 0;
1056   session_fifo_event_t evt;
1057   unix_shared_memory_queue_t *q;
1058
1059   /* Fabricate event */
1060   evt.session_handle = session_handle;
1061   evt.event_type = evt_type;
1062   evt.event_id = serial_number++;
1063
1064   q = session_manager_get_vpp_event_queue (thread_index);
1065
1066   /* Based on request block (or not) for lack of space */
1067   if (PREDICT_TRUE (q->cursize < q->maxsize))
1068     {
1069       if (unix_shared_memory_queue_add (q, (u8 *) & evt,
1070                                         1 /* do wait for mutex */ ))
1071         {
1072           clib_warning ("failed to enqueue evt");
1073         }
1074     }
1075   else
1076     {
1077       clib_warning ("queue full");
1078       return;
1079     }
1080 }
1081
1082 /**
1083  * Disconnect session and propagate to transport. This should eventually
1084  * result in a delete notification that allows us to cleanup session state.
1085  * Called for both active/passive disconnects.
1086  *
1087  * Should be called from the session's thread.
1088  */
1089 void
1090 stream_session_disconnect (stream_session_t * s)
1091 {
1092   s->session_state = SESSION_STATE_CLOSED;
1093   tp_vfts[s->session_type].close (s->connection_index, s->thread_index);
1094 }
1095
1096 /**
1097  * Cleanup transport and session state.
1098  *
1099  * Notify transport of the cleanup, wait for a delete notify to actually
1100  * remove the session state.
1101  */
1102 void
1103 stream_session_cleanup (stream_session_t * s)
1104 {
1105   session_manager_main_t *smm = &session_manager_main;
1106   int rv;
1107
1108   s->session_state = SESSION_STATE_CLOSED;
1109
1110   /* Delete from the main lookup table to avoid more enqueues */
1111   rv = stream_session_table_del (smm, s);
1112   if (rv)
1113     clib_warning ("hash delete error, rv %d", rv);
1114
1115   tp_vfts[s->session_type].cleanup (s->connection_index, s->thread_index);
1116 }
1117
1118 void
1119 session_register_transport (u8 type, const transport_proto_vft_t * vft)
1120 {
1121   session_manager_main_t *smm = vnet_get_session_manager_main ();
1122
1123   vec_validate (tp_vfts, type);
1124   tp_vfts[type] = *vft;
1125
1126   /* If an offset function is provided, then peek instead of dequeue */
1127   smm->session_tx_fns[type] =
1128     (vft->tx_fifo_offset) ? session_tx_fifo_peek_and_snd :
1129     session_tx_fifo_dequeue_and_snd;
1130 }
1131
1132 transport_proto_vft_t *
1133 session_get_transport_vft (u8 type)
1134 {
1135   if (type >= vec_len (tp_vfts))
1136     return 0;
1137   return &tp_vfts[type];
1138 }
1139
1140 /**
1141  * Allocate vpp event queue (once) per worker thread
1142  */
1143 void
1144 session_vpp_event_queue_allocate (session_manager_main_t * smm,
1145                                   u32 thread_index)
1146 {
1147   api_main_t *am = &api_main;
1148   void *oldheap;
1149   u32 event_queue_length = 2048;
1150
1151   if (smm->vpp_event_queues[thread_index] == 0)
1152     {
1153       /* Allocate event fifo in the /vpe-api shared-memory segment */
1154       oldheap = svm_push_data_heap (am->vlib_rp);
1155
1156       if (smm->configured_event_queue_length)
1157         event_queue_length = smm->configured_event_queue_length;
1158
1159       smm->vpp_event_queues[thread_index] =
1160         unix_shared_memory_queue_init
1161         (event_queue_length,
1162          sizeof (session_fifo_event_t), 0 /* consumer pid */ ,
1163          0 /* (do not) send signal when queue non-empty */ );
1164
1165       svm_pop_heap (oldheap);
1166     }
1167 }
1168
1169 session_type_t
1170 session_type_from_proto_and_ip (transport_proto_t proto, u8 is_ip4)
1171 {
1172   if (proto == TRANSPORT_PROTO_TCP)
1173     {
1174       if (is_ip4)
1175         return SESSION_TYPE_IP4_TCP;
1176       else
1177         return SESSION_TYPE_IP6_TCP;
1178     }
1179   else
1180     {
1181       if (is_ip4)
1182         return SESSION_TYPE_IP4_UDP;
1183       else
1184         return SESSION_TYPE_IP6_UDP;
1185     }
1186
1187   return SESSION_N_TYPES;
1188 }
1189
1190 static clib_error_t *
1191 session_manager_main_enable (vlib_main_t * vm)
1192 {
1193   session_manager_main_t *smm = &session_manager_main;
1194   vlib_thread_main_t *vtm = vlib_get_thread_main ();
1195   u32 num_threads;
1196   int i;
1197
1198   num_threads = 1 /* main thread */  + vtm->n_threads;
1199
1200   if (num_threads < 1)
1201     return clib_error_return (0, "n_thread_stacks not set");
1202
1203   /* $$$ config parameters */
1204   svm_fifo_segment_init (0x200000000ULL /* first segment base VA */ ,
1205                          20 /* timeout in seconds */ );
1206
1207   /* configure per-thread ** vectors */
1208   vec_validate (smm->sessions, num_threads - 1);
1209   vec_validate (smm->session_indices_to_enqueue_by_thread, num_threads - 1);
1210   vec_validate (smm->tx_buffers, num_threads - 1);
1211   vec_validate (smm->pending_event_vector, num_threads - 1);
1212   vec_validate (smm->free_event_vector, num_threads - 1);
1213   vec_validate (smm->current_enqueue_epoch, num_threads - 1);
1214   vec_validate (smm->vpp_event_queues, num_threads - 1);
1215
1216   for (i = 0; i < num_threads; i++)
1217     {
1218       vec_validate (smm->free_event_vector[i], 0);
1219       _vec_len (smm->free_event_vector[i]) = 0;
1220       vec_validate (smm->pending_event_vector[i], 0);
1221       _vec_len (smm->pending_event_vector[i]) = 0;
1222     }
1223
1224 #if SESSION_DBG
1225   vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1226 #endif
1227
1228   /* Allocate vpp event queues */
1229   for (i = 0; i < vec_len (smm->vpp_event_queues); i++)
1230     session_vpp_event_queue_allocate (smm, i);
1231
1232   /* $$$$ preallocate hack config parameter */
1233   for (i = 0; i < smm->preallocated_sessions; i++)
1234     {
1235       stream_session_t *ss __attribute__ ((unused));
1236       pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES);
1237     }
1238
1239   for (i = 0; i < smm->preallocated_sessions; i++)
1240     pool_put_index (smm->sessions[0], i);
1241
1242   clib_bihash_init_16_8 (&smm->v4_session_hash, "v4 session table",
1243                          200000 /* $$$$ config parameter nbuckets */ ,
1244                          (64 << 20) /*$$$ config parameter table size */ );
1245   clib_bihash_init_48_8 (&smm->v6_session_hash, "v6 session table",
1246                          200000 /* $$$$ config parameter nbuckets */ ,
1247                          (64 << 20) /*$$$ config parameter table size */ );
1248
1249   clib_bihash_init_16_8 (&smm->v4_half_open_hash, "v4 half-open table",
1250                          200000 /* $$$$ config parameter nbuckets */ ,
1251                          (64 << 20) /*$$$ config parameter table size */ );
1252   clib_bihash_init_48_8 (&smm->v6_half_open_hash, "v6 half-open table",
1253                          200000 /* $$$$ config parameter nbuckets */ ,
1254                          (64 << 20) /*$$$ config parameter table size */ );
1255
1256   smm->is_enabled = 1;
1257
1258   /* Enable TCP transport */
1259   vnet_tcp_enable_disable (vm, 1);
1260
1261   return 0;
1262 }
1263
1264 void
1265 session_node_enable_disable (u8 is_en)
1266 {
1267   u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1268   /* *INDENT-OFF* */
1269   foreach_vlib_main (({
1270     vlib_node_set_state (this_vlib_main, session_queue_node.index,
1271                          state);
1272   }));
1273   /* *INDENT-ON* */
1274 }
1275
1276 clib_error_t *
1277 vnet_session_enable_disable (vlib_main_t * vm, u8 is_en)
1278 {
1279   if (is_en)
1280     {
1281       if (session_manager_main.is_enabled)
1282         return 0;
1283
1284       session_node_enable_disable (is_en);
1285
1286       return session_manager_main_enable (vm);
1287     }
1288   else
1289     {
1290       session_manager_main.is_enabled = 0;
1291       session_node_enable_disable (is_en);
1292     }
1293
1294   return 0;
1295 }
1296
1297 clib_error_t *
1298 session_manager_main_init (vlib_main_t * vm)
1299 {
1300   session_manager_main_t *smm = &session_manager_main;
1301
1302   smm->vlib_main = vm;
1303   smm->vnet_main = vnet_get_main ();
1304   smm->is_enabled = 0;
1305
1306   return 0;
1307 }
1308
1309 VLIB_INIT_FUNCTION (session_manager_main_init);
1310
1311 static clib_error_t *
1312 session_config_fn (vlib_main_t * vm, unformat_input_t * input)
1313 {
1314   session_manager_main_t *smm = &session_manager_main;
1315   u32 nitems;
1316
1317   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
1318     {
1319       if (unformat (input, "event-queue-length %d", &nitems))
1320         {
1321           if (nitems >= 2048)
1322             smm->configured_event_queue_length = nitems;
1323           else
1324             clib_warning ("event queue length %d too small, ignored", nitems);
1325         }
1326       if (unformat (input, "preallocated-sessions %d",
1327                     &smm->preallocated_sessions))
1328         ;
1329       else
1330         return clib_error_return (0, "unknown input `%U'",
1331                                   format_unformat_error, input);
1332     }
1333   return 0;
1334 }
1335
1336 VLIB_CONFIG_FUNCTION (session_config_fn, "session");
1337
1338 /*
1339  * fd.io coding-style-patch-verification: ON
1340  *
1341  * Local Variables:
1342  * eval: (c-set-style "gnu")
1343  * End:
1344  */