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