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_transport4 (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_transport6 (ip6_address_t * lcl, ip6_address_t * rmt,
 362                                   u16 lcl_port, u16 rmt_port, u8 proto,
 363                                   u32 my_thread_index)
 364 {
 365   session_manager_main_t *smm = &session_manager_main;
 366   stream_session_t *s;
 367   session_kv6_t kv6;
 368   int rv;
 369
 370   make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
 371   rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
 372   if (rv == 0)
 373     {
 374       s = stream_session_get_tsi (kv6.value, my_thread_index);
 375
 376       return tp_vfts[s->session_type].get_connection (s->connection_index,
 377                                                       my_thread_index);
 378     }
 379
 380   /* If nothing is found, check if any listener is available */
 381   s = stream_session_lookup_listener6 (lcl, lcl_port, proto);
 382   if (s)
 383     return tp_vfts[s->session_type].get_listener (s->connection_index);
 384
 385   /* Finally, try half-open connections */
 386   rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
 387   if (rv == 0)
 388     return tp_vfts[proto].get_half_open (kv6.value & 0xFFFFFFFF);
 389
 390   return 0;
 391 }
 392
 393 int
 394 stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
 395                          stream_session_t ** ret_s)
 396 {
 397   session_manager_main_t *smm = &session_manager_main;
 398   svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
 399   u32 fifo_segment_index;
 400   u32 pool_index;
 401   stream_session_t *s;
 402   u64 value;
 403   u32 thread_index = tc->thread_index;
 404   int rv;
 405
 406   ASSERT (thread_index == vlib_get_thread_index ());
 407
 408   if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo,
 409                                                  &server_tx_fifo,
 410                                                  &fifo_segment_index)))
 411     return rv;
 412
 413   /* Create the session */
 414   pool_get_aligned (smm->sessions[thread_index], s, CLIB_CACHE_LINE_BYTES);
 415   memset (s, 0, sizeof (*s));
 416
 417   /* Initialize backpointers */
 418   pool_index = s - smm->sessions[thread_index];
 419   server_rx_fifo->master_session_index = pool_index;
 420   server_rx_fifo->master_thread_index = thread_index;
 421
 422   server_tx_fifo->master_session_index = pool_index;
 423   server_tx_fifo->master_thread_index = thread_index;
 424
 425   s->server_rx_fifo = server_rx_fifo;
 426   s->server_tx_fifo = server_tx_fifo;
 427
 428   /* Initialize state machine, such as it is... */
 429   s->session_type = tc->proto;
 430   s->session_state = SESSION_STATE_CONNECTING;
 431   s->svm_segment_index = fifo_segment_index;
 432   s->thread_index = thread_index;
 433   s->session_index = pool_index;
 434
 435   /* Attach transport to session */
 436   s->connection_index = tc->c_index;
 437
 438   /* Attach session to transport */
 439   tc->s_index = s->session_index;
 440
 441   /* Add to the main lookup table */
 442   value = (((u64) thread_index) << 32) | (u64) s->session_index;
 443   stream_session_table_add_for_tc (tc, value);
 444
 445   *ret_s = s;
 446
 447   return 0;
 448 }
 449
 450 /** Enqueue buffer chain tail */
 451 always_inline int
 452 session_enqueue_chain_tail (stream_session_t * s, vlib_buffer_t * b,
 453                             u32 offset, u8 is_in_order)
 454 {
 455   vlib_buffer_t *chain_b;
 456   u32 chain_bi = b->next_buffer;
 457   vlib_main_t *vm = vlib_get_main ();
 458   u8 *data, len;
 459   u16 written = 0;
 460   int rv = 0;
 461
 462   do
 463     {
 464       chain_b = vlib_get_buffer (vm, chain_bi);
 465       data = vlib_buffer_get_current (chain_b);
 466       len = chain_b->current_length;
 467       if (is_in_order)
 468         {
 469           rv = svm_fifo_enqueue_nowait (s->server_rx_fifo, len, data);
 470           if (rv < len)
 471             {
 472               return (rv > 0) ? (written + rv) : written;
 473             }
 474           written += rv;
 475         }
 476       else
 477         {
 478           rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset, len,
 479                                              data);
 480           if (rv)
 481             return -1;
 482           offset += len;
 483         }
 484     }
 485   while ((chain_bi = (chain_b->flags & VLIB_BUFFER_NEXT_PRESENT)
 486           ? chain_b->next_buffer : 0));
 487
 488   if (is_in_order)
 489     return written;
 490
 491   return 0;
 492 }
 493
 494 /*
 495  * Enqueue data for delivery to session peer. Does not notify peer of enqueue
 496  * event but on request can queue notification events for later delivery by
 497  * calling stream_server_flush_enqueue_events().
 498  *
 499  * @param tc Transport connection which is to be enqueued data
 500  * @param b Buffer to be enqueued
 501  * @param offset Offset at which to start enqueueing if out-of-order
 502  * @param queue_event Flag to indicate if peer is to be notified or if event
 503  *                    is to be queued. The former is useful when more data is
 504  *                    enqueued and only one event is to be generated.
 505  * @param is_in_order Flag to indicate if data is in order
 506  * @return Number of bytes enqueued or a negative value if enqueueing failed.
 507  */
 508 int
 509 stream_session_enqueue_data (transport_connection_t * tc, vlib_buffer_t * b,
 510                              u32 offset, u8 queue_event, u8 is_in_order)
 511 {
 512   stream_session_t *s;
 513   int enqueued = 0, rv;
 514
 515   s = stream_session_get (tc->s_index, tc->thread_index);
 516
 517   if (is_in_order)
 518     {
 519       enqueued =
 520         svm_fifo_enqueue_nowait (s->server_rx_fifo, b->current_length,
 521                                  vlib_buffer_get_current (b));
 522       if (PREDICT_FALSE
 523           ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && enqueued > 0))
 524         {
 525           rv = session_enqueue_chain_tail (s, b, 0, 1);
 526           if (rv <= 0)
 527             return enqueued;
 528           enqueued += rv;
 529         }
 530     }
 531   else
 532     {
 533       rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset,
 534                                          b->current_length,
 535                                          vlib_buffer_get_current (b));
 536       if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && !rv))
 537         rv = session_enqueue_chain_tail (s, b, offset + b->current_length, 0);
 538       if (rv)
 539         return -1;
 540     }
 541
 542   if (queue_event)
 543     {
 544       /* Queue RX event on this fifo. Eventually these will need to be flushed
 545        * by calling stream_server_flush_enqueue_events () */
 546       session_manager_main_t *smm = vnet_get_session_manager_main ();
 547       u32 thread_index = s->thread_index;
 548       u32 my_enqueue_epoch = smm->current_enqueue_epoch[thread_index];
 549
 550       if (s->enqueue_epoch != my_enqueue_epoch)
 551         {
 552           s->enqueue_epoch = my_enqueue_epoch;
 553           vec_add1 (smm->session_indices_to_enqueue_by_thread[thread_index],
 554                     s - smm->sessions[thread_index]);
 555         }
 556     }
 557
 558   if (is_in_order)
 559     return enqueued;
 560
 561   return 0;
 562 }
 563
 564 /** Check if we have space in rx fifo to push more bytes */
 565 u8
 566 stream_session_no_space (transport_connection_t * tc, u32 thread_index,
 567                          u16 data_len)
 568 {
 569   stream_session_t *s = stream_session_get (tc->s_index, thread_index);
 570
 571   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY))
 572     return 1;
 573
 574   if (data_len > svm_fifo_max_enqueue (s->server_rx_fifo))
 575     return 1;
 576
 577   return 0;
 578 }
 579
 580 u32
 581 stream_session_tx_fifo_max_dequeue (transport_connection_t * tc)
 582 {
 583   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
 584   if (s->session_state != SESSION_STATE_READY)
 585     return 0;
 586   return svm_fifo_max_dequeue (s->server_tx_fifo);
 587 }
 588
 589 int
 590 stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
 591                            u32 offset, u32 max_bytes)
 592 {
 593   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
 594   return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
 595 }
 596
 597 u32
 598 stream_session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
 599 {
 600   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
 601   return svm_fifo_dequeue_drop (s->server_tx_fifo, max_bytes);
 602 }
 603
 604 /**
 605  * Notify session peer that new data has been enqueued.
 606  *
 607  * @param s Stream session for which the event is to be generated.
 608  * @param block Flag to indicate if call should block if event queue is full.
 609  *
 610  * @return 0 on succes or negative number if failed to send notification.
 611  */
 612 static int
 613 stream_session_enqueue_notify (stream_session_t * s, u8 block)
 614 {
 615   application_t *app;
 616   session_fifo_event_t evt;
 617   unix_shared_memory_queue_t *q;
 618   static u32 serial_number;
 619
 620   if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED))
 621     return 0;
 622
 623   /* Get session's server */
 624   app = application_get (s->app_index);
 625
 626   /* Built-in server? Hand event to the callback... */
 627   if (app->cb_fns.builtin_server_rx_callback)
 628     return app->cb_fns.builtin_server_rx_callback (s);
 629
 630   /* If no event, send one */
 631   if (svm_fifo_set_event (s->server_rx_fifo))
 632     {
 633       /* Fabricate event */
 634       evt.fifo = s->server_rx_fifo;
 635       evt.event_type = FIFO_EVENT_APP_RX;
 636       evt.event_id = serial_number++;
 637
 638       /* Add event to server's event queue */
 639       q = app->event_queue;
 640
 641       /* Based on request block (or not) for lack of space */
 642       if (block || PREDICT_TRUE (q->cursize < q->maxsize))
 643         unix_shared_memory_queue_add (app->event_queue, (u8 *) & evt,
 644                                       0 /* do wait for mutex */ );
 645       else
 646         {
 647           clib_warning ("fifo full");
 648           return -1;
 649         }
 650     }
 651
 652   /* *INDENT-OFF* */
 653   SESSION_EVT_DBG(SESSION_EVT_ENQ, s, ({
 654       ed->data[0] = evt.event_id;
 655       ed->data[1] = svm_fifo_max_dequeue (s->server_rx_fifo);
 656   }));
 657   /* *INDENT-ON* */
 658
 659   return 0;
 660 }
 661
 662 /**
 663  * Flushes queue of sessions that are to be notified of new data
 664  * enqueued events.
 665  *
 666  * @param thread_index Thread index for which the flush is to be performed.
 667  * @return 0 on success or a positive number indicating the number of
 668  *         failures due to API queue being full.
 669  */
 670 int
 671 session_manager_flush_enqueue_events (u32 thread_index)
 672 {
 673   session_manager_main_t *smm = &session_manager_main;
 674   u32 *session_indices_to_enqueue;
 675   int i, errors = 0;
 676
 677   session_indices_to_enqueue =
 678     smm->session_indices_to_enqueue_by_thread[thread_index];
 679
 680   for (i = 0; i < vec_len (session_indices_to_enqueue); i++)
 681     {
 682       stream_session_t *s0;
 683
 684       /* Get session */
 685       s0 = stream_session_get (session_indices_to_enqueue[i], thread_index);
 686       if (stream_session_enqueue_notify (s0, 0 /* don't block */ ))
 687         {
 688           errors++;
 689         }
 690     }
 691
 692   vec_reset_length (session_indices_to_enqueue);
 693
 694   smm->session_indices_to_enqueue_by_thread[thread_index] =
 695     session_indices_to_enqueue;
 696
 697   /* Increment enqueue epoch for next round */
 698   smm->current_enqueue_epoch[thread_index]++;
 699
 700   return errors;
 701 }
 702
 703 /**
 704  * Init fifo tail and head pointers
 705  *
 706  * Useful if transport uses absolute offsets for tracking ooo segments.
 707  */
 708 void
 709 stream_session_init_fifos_pointers (transport_connection_t * tc,
 710                                     u32 rx_pointer, u32 tx_pointer)
 711 {
 712   stream_session_t *s;
 713   s = stream_session_get (tc->s_index, tc->thread_index);
 714   svm_fifo_init_pointers (s->server_rx_fifo, rx_pointer);
 715   svm_fifo_init_pointers (s->server_tx_fifo, tx_pointer);
 716 }
 717
 718 int
 719 stream_session_connect_notify (transport_connection_t * tc, u8 sst,
 720                                u8 is_fail)
 721 {
 722   session_manager_main_t *smm = &session_manager_main;
 723   application_t *app;
 724   stream_session_t *new_s = 0;
 725   u64 handle;
 726   u32 api_context = 0;
 727   int error = 0;
 728
 729   handle = stream_session_half_open_lookup (smm, &tc->lcl_ip, &tc->rmt_ip,
 730                                             tc->lcl_port, tc->rmt_port,
 731                                             tc->proto);
 732   if (handle == HALF_OPEN_LOOKUP_INVALID_VALUE)
 733     {
 734       clib_warning ("This can't be good!");
 735       return -1;
 736     }
 737
 738   /* Get the app's index from the handle we stored when opening connection */
 739   app = application_get (handle >> 32);
 740   api_context = tc->s_index;
 741
 742   if (!is_fail)
 743     {
 744       segment_manager_t *sm;
 745       sm = application_get_connect_segment_manager (app);
 746
 747       /* Create new session (svm segments are allocated if needed) */
 748       if (stream_session_create_i (sm, tc, &new_s))
 749         {
 750           is_fail = 1;
 751           error = -1;
 752         }
 753       else
 754         new_s->app_index = app->index;
 755     }
 756
 757   /* Notify client */
 758   app->cb_fns.session_connected_callback (app->index, api_context, new_s,
 759                                           is_fail);
 760
 761   /* Cleanup session lookup */
 762   stream_session_half_open_table_del (smm, sst, tc);
 763
 764   return error;
 765 }
 766
 767 void
 768 stream_session_accept_notify (transport_connection_t * tc)
 769 {
 770   application_t *server;
 771   stream_session_t *s;
 772
 773   s = stream_session_get (tc->s_index, tc->thread_index);
 774   server = application_get (s->app_index);
 775   server->cb_fns.session_accept_callback (s);
 776 }
 777
 778 /**
 779  * Notification from transport that connection is being closed.
 780  *
 781  * A disconnect is sent to application but state is not removed. Once
 782  * disconnect is acknowledged by application, session disconnect is called.
 783  * Ultimately this leads to close being called on transport (passive close).
 784  */
 785 void
 786 stream_session_disconnect_notify (transport_connection_t * tc)
 787 {
 788   application_t *server;
 789   stream_session_t *s;
 790
 791   s = stream_session_get (tc->s_index, tc->thread_index);
 792   server = application_get (s->app_index);
 793   server->cb_fns.session_disconnect_callback (s);
 794 }
 795
 796 /**
 797  * Cleans up session and associated app if needed.
 798  */
 799 void
 800 stream_session_delete (stream_session_t * s)
 801 {
 802   session_manager_main_t *smm = vnet_get_session_manager_main ();
 803
 804   /* Delete from the main lookup table. */
 805   stream_session_table_del (smm, s);
 806
 807   /* Cleanup fifo segments */
 808   segment_manager_dealloc_fifos (s->svm_segment_index, s->server_rx_fifo,
 809                                  s->server_tx_fifo);
 810
 811   pool_put (smm->sessions[s->thread_index], s);
 812 }
 813
 814 /**
 815  * Notification from transport that connection is being deleted
 816  *
 817  * This should be called only on previously fully established sessions. For
 818  * instance failed connects should call stream_session_connect_notify and
 819  * indicate that the connect has failed.
 820  */
 821 void
 822 stream_session_delete_notify (transport_connection_t * tc)
 823 {
 824   stream_session_t *s;
 825
 826   /* App might've been removed already */
 827   s = stream_session_get_if_valid (tc->s_index, tc->thread_index);
 828   if (!s)
 829     {
 830       return;
 831     }
 832   stream_session_delete (s);
 833 }
 834
 835 /**
 836  * Notify application that connection has been reset.
 837  */
 838 void
 839 stream_session_reset_notify (transport_connection_t * tc)
 840 {
 841   stream_session_t *s;
 842   application_t *app;
 843   s = stream_session_get (tc->s_index, tc->thread_index);
 844
 845   app = application_get (s->app_index);
 846   app->cb_fns.session_reset_callback (s);
 847 }
 848
 849 /**
 850  * Accept a stream session. Optionally ping the server by callback.
 851  */
 852 int
 853 stream_session_accept (transport_connection_t * tc, u32 listener_index,
 854                        u8 sst, u8 notify)
 855 {
 856   application_t *server;
 857   stream_session_t *s, *listener;
 858   segment_manager_t *sm;
 859
 860   int rv;
 861
 862   /* Find the server */
 863   listener = listen_session_get (sst, listener_index);
 864   server = application_get (listener->app_index);
 865
 866   sm = application_get_listen_segment_manager (server, listener);
 867   if ((rv = stream_session_create_i (sm, tc, &s)))
 868     return rv;
 869
 870   s->app_index = server->index;
 871   s->listener_index = listener_index;
 872   s->session_state = SESSION_STATE_ACCEPTING;
 873
 874   /* Shoulder-tap the server */
 875   if (notify)
 876     {
 877       server->cb_fns.session_accept_callback (s);
 878     }
 879
 880   return 0;
 881 }
 882
 883 /**
 884  * Ask transport to open connection to remote transport endpoint.
 885  *
 886  * Stores handle for matching request with reply since the call can be
 887  * asynchronous. For instance, for TCP the 3-way handshake must complete
 888  * before reply comes. Session is only created once connection is established.
 889  *
 890  * @param app_index Index of the application requesting the connect
 891  * @param st Session type requested.
 892  * @param tep Remote transport endpoint
 893  * @param res Resulting transport connection .
 894  */
 895 int
 896 stream_session_open (u32 app_index, session_type_t st,
 897                      transport_endpoint_t * tep,
 898                      transport_connection_t ** res)
 899 {
 900   transport_connection_t *tc;
 901   int rv;
 902   u64 handle;
 903
 904   rv = tp_vfts[st].open (&tep->ip, tep->port);
 905   if (rv < 0)
 906     {
 907       clib_warning ("Transport failed to open connection.");
 908       return VNET_API_ERROR_SESSION_CONNECT_FAIL;
 909     }
 910
 911   tc = tp_vfts[st].get_half_open ((u32) rv);
 912
 913   /* Save app and tc index. The latter is needed to help establish the
 914    * connection while the former is needed when the connect notify comes
 915    * and we have to notify the external app */
 916   handle = (((u64) app_index) << 32) | (u64) tc->c_index;
 917
 918   /* Add to the half-open lookup table */
 919   stream_session_half_open_table_add (st, tc, handle);
 920
 921   *res = tc;
 922
 923   return 0;
 924 }
 925
 926 /**
 927  * Ask transport to listen on local transport endpoint.
 928  *
 929  * @param s Session for which listen will be called. Note that unlike
 930  *          established sessions, listen sessions are not associated to a
 931  *          thread.
 932  * @param tep Local endpoint to be listened on.
 933  */
 934 int
 935 stream_session_listen (stream_session_t * s, transport_endpoint_t * tep)
 936 {
 937   transport_connection_t *tc;
 938   u32 tci;
 939
 940   /* Transport bind/listen  */
 941   tci = tp_vfts[s->session_type].bind (s->session_index, &tep->ip, tep->port);
 942
 943   if (tci == (u32) ~ 0)
 944     return -1;
 945
 946   /* Attach transport to session */
 947   s->connection_index = tci;
 948   tc = tp_vfts[s->session_type].get_listener (tci);
 949
 950   /* Weird but handle it ... */
 951   if (tc == 0)
 952     return -1;
 953
 954   /* Add to the main lookup table */
 955   stream_session_table_add_for_tc (tc, s->session_index);
 956
 957   return 0;
 958 }
 959
 960 /**
 961  * Ask transport to stop listening on local transport endpoint.
 962  *
 963  * @param s Session to stop listening on. It must be in state LISTENING.
 964  */
 965 int
 966 stream_session_stop_listen (stream_session_t * s)
 967 {
 968   transport_connection_t *tc;
 969
 970   if (s->session_state != SESSION_STATE_LISTENING)
 971     {
 972       clib_warning ("not a listening session");
 973       return -1;
 974     }
 975
 976   tc = tp_vfts[s->session_type].get_listener (s->connection_index);
 977   if (!tc)
 978     {
 979       clib_warning ("no transport");
 980       return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
 981     }
 982
 983   stream_session_table_del_for_tc (tc);
 984   tp_vfts[s->session_type].unbind (s->connection_index);
 985   return 0;
 986 }
 987
 988 void
 989 session_send_session_evt_to_thread (u64 session_handle,
 990                                     fifo_event_type_t evt_type,
 991                                     u32 thread_index)
 992 {
 993   static u16 serial_number = 0;
 994   session_fifo_event_t evt;
 995   unix_shared_memory_queue_t *q;
 996
 997   /* Fabricate event */
 998   evt.session_handle = session_handle;
 999   evt.event_type = evt_type;
1000   evt.event_id = serial_number++;
1001
1002   q = session_manager_get_vpp_event_queue (thread_index);
1003
1004   /* Based on request block (or not) for lack of space */
1005   if (PREDICT_TRUE (q->cursize < q->maxsize))
1006     {
1007       if (unix_shared_memory_queue_add (q, (u8 *) & evt,
1008                                         1 /* do wait for mutex */ ))
1009         {
1010           clib_warning ("failed to enqueue evt");
1011         }
1012     }
1013   else
1014     {
1015       clib_warning ("queue full");
1016       return;
1017     }
1018 }
1019
1020 /**
1021  * Disconnect session and propagate to transport. This should eventually
1022  * result in a delete notification that allows us to cleanup session state.
1023  * Called for both active/passive disconnects.
1024  *
1025  * Should be called from the session's thread.
1026  */
1027 void
1028 stream_session_disconnect (stream_session_t * s)
1029 {
1030   s->session_state = SESSION_STATE_CLOSED;
1031   tp_vfts[s->session_type].close (s->connection_index, s->thread_index);
1032 }
1033
1034 /**
1035  * Cleanup transport and session state.
1036  *
1037  * Notify transport of the cleanup, wait for a delete notify to actually
1038  * remove the session state.
1039  */
1040 void
1041 stream_session_cleanup (stream_session_t * s)
1042 {
1043   session_manager_main_t *smm = &session_manager_main;
1044   int rv;
1045
1046   s->session_state = SESSION_STATE_CLOSED;
1047
1048   /* Delete from the main lookup table to avoid more enqueues */
1049   rv = stream_session_table_del (smm, s);
1050   if (rv)
1051     clib_warning ("hash delete error, rv %d", rv);
1052
1053   tp_vfts[s->session_type].cleanup (s->connection_index, s->thread_index);
1054 }
1055
1056 void
1057 session_register_transport (u8 type, const transport_proto_vft_t * vft)
1058 {
1059   session_manager_main_t *smm = vnet_get_session_manager_main ();
1060
1061   vec_validate (tp_vfts, type);
1062   tp_vfts[type] = *vft;
1063
1064   /* If an offset function is provided, then peek instead of dequeue */
1065   smm->session_tx_fns[type] =
1066     (vft->tx_fifo_offset) ? session_tx_fifo_peek_and_snd :
1067     session_tx_fifo_dequeue_and_snd;
1068 }
1069
1070 transport_proto_vft_t *
1071 session_get_transport_vft (u8 type)
1072 {
1073   if (type >= vec_len (tp_vfts))
1074     return 0;
1075   return &tp_vfts[type];
1076 }
1077
1078 /**
1079  * Allocate vpp event queue (once) per worker thread
1080  */
1081 void
1082 session_vpp_event_queue_allocate (session_manager_main_t * smm,
1083                                   u32 thread_index)
1084 {
1085   api_main_t *am = &api_main;
1086   void *oldheap;
1087   u32 event_queue_length = 2048;
1088
1089   if (smm->vpp_event_queues[thread_index] == 0)
1090     {
1091       /* Allocate event fifo in the /vpe-api shared-memory segment */
1092       oldheap = svm_push_data_heap (am->vlib_rp);
1093
1094       if (smm->configured_event_queue_length)
1095         event_queue_length = smm->configured_event_queue_length;
1096
1097       smm->vpp_event_queues[thread_index] =
1098         unix_shared_memory_queue_init
1099         (event_queue_length,
1100          sizeof (session_fifo_event_t), 0 /* consumer pid */ ,
1101          0 /* (do not) send signal when queue non-empty */ );
1102
1103       svm_pop_heap (oldheap);
1104     }
1105 }
1106
1107 session_type_t
1108 session_type_from_proto_and_ip (transport_proto_t proto, u8 is_ip4)
1109 {
1110   if (proto == TRANSPORT_PROTO_TCP)
1111     {
1112       if (is_ip4)
1113         return SESSION_TYPE_IP4_TCP;
1114       else
1115         return SESSION_TYPE_IP6_TCP;
1116     }
1117   else
1118     {
1119       if (is_ip4)
1120         return SESSION_TYPE_IP4_UDP;
1121       else
1122         return SESSION_TYPE_IP6_UDP;
1123     }
1124
1125   return SESSION_N_TYPES;
1126 }
1127
1128 static clib_error_t *
1129 session_manager_main_enable (vlib_main_t * vm)
1130 {
1131   session_manager_main_t *smm = &session_manager_main;
1132   vlib_thread_main_t *vtm = vlib_get_thread_main ();
1133   u32 num_threads;
1134   int i;
1135
1136   num_threads = 1 /* main thread */  + vtm->n_threads;
1137
1138   if (num_threads < 1)
1139     return clib_error_return (0, "n_thread_stacks not set");
1140
1141   /* $$$ config parameters */
1142   svm_fifo_segment_init (0x200000000ULL /* first segment base VA */ ,
1143                          20 /* timeout in seconds */ );
1144
1145   /* configure per-thread ** vectors */
1146   vec_validate (smm->sessions, num_threads - 1);
1147   vec_validate (smm->session_indices_to_enqueue_by_thread, num_threads - 1);
1148   vec_validate (smm->tx_buffers, num_threads - 1);
1149   vec_validate (smm->pending_event_vector, num_threads - 1);
1150   vec_validate (smm->free_event_vector, num_threads - 1);
1151   vec_validate (smm->current_enqueue_epoch, num_threads - 1);
1152   vec_validate (smm->vpp_event_queues, num_threads - 1);
1153
1154   for (i = 0; i < num_threads; i++)
1155     {
1156       vec_validate (smm->free_event_vector[i], 0);
1157       _vec_len (smm->free_event_vector[i]) = 0;
1158       vec_validate (smm->pending_event_vector[i], 0);
1159       _vec_len (smm->pending_event_vector[i]) = 0;
1160     }
1161
1162 #if SESSION_DBG
1163   vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1164 #endif
1165
1166   /* Allocate vpp event queues */
1167   for (i = 0; i < vec_len (smm->vpp_event_queues); i++)
1168     session_vpp_event_queue_allocate (smm, i);
1169
1170   /* $$$$ preallocate hack config parameter */
1171   for (i = 0; i < smm->preallocated_sessions; i++)
1172     {
1173       stream_session_t *ss __attribute__ ((unused));
1174       pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES);
1175     }
1176
1177   for (i = 0; i < smm->preallocated_sessions; i++)
1178     pool_put_index (smm->sessions[0], i);
1179
1180   clib_bihash_init_16_8 (&smm->v4_session_hash, "v4 session table",
1181                          200000 /* $$$$ config parameter nbuckets */ ,
1182                          (64 << 20) /*$$$ config parameter table size */ );
1183   clib_bihash_init_48_8 (&smm->v6_session_hash, "v6 session table",
1184                          200000 /* $$$$ config parameter nbuckets */ ,
1185                          (64 << 20) /*$$$ config parameter table size */ );
1186
1187   clib_bihash_init_16_8 (&smm->v4_half_open_hash, "v4 half-open table",
1188                          200000 /* $$$$ config parameter nbuckets */ ,
1189                          (64 << 20) /*$$$ config parameter table size */ );
1190   clib_bihash_init_48_8 (&smm->v6_half_open_hash, "v6 half-open table",
1191                          200000 /* $$$$ config parameter nbuckets */ ,
1192                          (64 << 20) /*$$$ config parameter table size */ );
1193
1194   smm->is_enabled = 1;
1195
1196   /* Enable TCP transport */
1197   vnet_tcp_enable_disable (vm, 1);
1198
1199   return 0;
1200 }
1201
1202 void
1203 session_node_enable_disable (u8 is_en)
1204 {
1205   u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1206   /* *INDENT-OFF* */
1207   foreach_vlib_main (({
1208     vlib_node_set_state (this_vlib_main, session_queue_node.index,
1209                          state);
1210   }));
1211   /* *INDENT-ON* */
1212 }
1213
1214 clib_error_t *
1215 vnet_session_enable_disable (vlib_main_t * vm, u8 is_en)
1216 {
1217   if (is_en)
1218     {
1219       if (session_manager_main.is_enabled)
1220         return 0;
1221
1222       session_node_enable_disable (is_en);
1223
1224       return session_manager_main_enable (vm);
1225     }
1226   else
1227     {
1228       session_manager_main.is_enabled = 0;
1229       session_node_enable_disable (is_en);
1230     }
1231
1232   return 0;
1233 }
1234
1235 clib_error_t *
1236 session_manager_main_init (vlib_main_t * vm)
1237 {
1238   session_manager_main_t *smm = &session_manager_main;
1239
1240   smm->vlib_main = vm;
1241   smm->vnet_main = vnet_get_main ();
1242   smm->is_enabled = 0;
1243
1244   return 0;
1245 }
1246
1247 VLIB_INIT_FUNCTION (session_manager_main_init);
1248
1249 static clib_error_t *
1250 session_config_fn (vlib_main_t * vm, unformat_input_t * input)
1251 {
1252   session_manager_main_t *smm = &session_manager_main;
1253   u32 nitems;
1254
1255   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
1256     {
1257       if (unformat (input, "event-queue-length %d", &nitems))
1258         {
1259           if (nitems >= 2048)
1260             smm->configured_event_queue_length = nitems;
1261           else
1262             clib_warning ("event queue length %d too small, ignored", nitems);
1263         }
1264       if (unformat (input, "preallocated-sessions %d",
1265                     &smm->preallocated_sessions))
1266         ;
1267       else
1268         return clib_error_return (0, "unknown input `%U'",
1269                                   format_unformat_error, input);
1270     }
1271   return 0;
1272 }
1273
1274 VLIB_CONFIG_FUNCTION (session_config_fn, "session");
1275
1276 /*
1277  * fd.io coding-style-patch-verification: ON
1278  *
1279  * Local Variables:
1280  * eval: (c-set-style "gnu")
1281  * End:
1282  */