src/vnet/session-apps/proxy.c

   1 /*
   2 * Copyright (c) 2015-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 #include <vnet/vnet.h>
  17 #include <vlibmemory/api.h>
  18 #include <vnet/session/application.h>
  19 #include <vnet/session/application_interface.h>
  20 #include <vnet/session-apps/proxy.h>
  21
  22 proxy_main_t proxy_main;
  23
  24 static void
  25 delete_proxy_session (stream_session_t * s, int is_active_open)
  26 {
  27   proxy_main_t *pm = &proxy_main;
  28   proxy_session_t *ps = 0;
  29   vnet_disconnect_args_t _a, *a = &_a;
  30   stream_session_t *active_open_session = 0;
  31   stream_session_t *server_session = 0;
  32   uword *p;
  33   u64 handle;
  34
  35   handle = session_handle (s);
  36
  37   clib_spinlock_lock_if_init (&pm->sessions_lock);
  38   if (is_active_open)
  39     {
  40       active_open_session = s;
  41
  42       p = hash_get (pm->proxy_session_by_active_open_handle, handle);
  43       if (p == 0)
  44         {
  45           clib_warning ("proxy session for %s handle %lld (%llx) AWOL",
  46                         is_active_open ? "active open" : "server",
  47                         handle, handle);
  48         }
  49       else
  50         {
  51           ps = pool_elt_at_index (pm->sessions, p[0]);
  52           if (ps->vpp_server_handle != ~0)
  53             server_session = session_get_from_handle (ps->vpp_server_handle);
  54           else
  55             server_session = 0;
  56         }
  57     }
  58   else
  59     {
  60       server_session = s;
  61
  62       p = hash_get (pm->proxy_session_by_server_handle, handle);
  63       if (p == 0)
  64         {
  65           clib_warning ("proxy session for %s handle %lld (%llx) AWOL",
  66                         is_active_open ? "active open" : "server",
  67                         handle, handle);
  68         }
  69       else
  70         {
  71           ps = pool_elt_at_index (pm->sessions, p[0]);
  72           if (ps->vpp_server_handle != ~0)
  73             active_open_session = session_get_from_handle
  74               (ps->vpp_server_handle);
  75           else
  76             active_open_session = 0;
  77         }
  78     }
  79
  80   if (ps)
  81     {
  82       if (CLIB_DEBUG > 0)
  83         memset (ps, 0xFE, sizeof (*ps));
  84       pool_put (pm->sessions, ps);
  85     }
  86
  87   clib_spinlock_unlock_if_init (&pm->sessions_lock);
  88
  89   if (active_open_session)
  90     {
  91       a->handle = session_handle (active_open_session);
  92       a->app_index = pm->active_open_app_index;
  93       hash_unset (pm->proxy_session_by_active_open_handle,
  94                   session_handle (active_open_session));
  95       vnet_disconnect_session (a);
  96     }
  97
  98   if (server_session)
  99     {
 100       a->handle = session_handle (server_session);
 101       a->app_index = pm->server_app_index;
 102       hash_unset (pm->proxy_session_by_server_handle,
 103                   session_handle (server_session));
 104       vnet_disconnect_session (a);
 105     }
 106 }
 107
 108 static int
 109 proxy_accept_callback (stream_session_t * s)
 110 {
 111   proxy_main_t *pm = &proxy_main;
 112
 113   s->session_state = SESSION_STATE_READY;
 114
 115   clib_spinlock_lock_if_init (&pm->sessions_lock);
 116
 117   return 0;
 118 }
 119
 120 static void
 121 proxy_disconnect_callback (stream_session_t * s)
 122 {
 123   delete_proxy_session (s, 0 /* is_active_open */ );
 124 }
 125
 126 static void
 127 proxy_reset_callback (stream_session_t * s)
 128 {
 129   clib_warning ("Reset session %U", format_stream_session, s, 2);
 130   delete_proxy_session (s, 0 /* is_active_open */ );
 131 }
 132
 133 static int
 134 proxy_connected_callback (u32 app_index, u32 api_context,
 135                           stream_session_t * s, u8 is_fail)
 136 {
 137   clib_warning ("called...");
 138   return -1;
 139 }
 140
 141 static int
 142 proxy_add_segment_callback (u32 client_index, const ssvm_private_t * sp)
 143 {
 144   clib_warning ("called...");
 145   return -1;
 146 }
 147
 148 static int
 149 proxy_rx_callback (stream_session_t * s)
 150 {
 151   u32 max_dequeue;
 152   int actual_transfer __attribute__ ((unused));
 153   svm_fifo_t *tx_fifo, *rx_fifo;
 154   proxy_main_t *pm = &proxy_main;
 155   u32 thread_index = vlib_get_thread_index ();
 156   vnet_connect_args_t _a, *a = &_a;
 157   proxy_session_t *ps;
 158   int proxy_index;
 159   uword *p;
 160   svm_fifo_t *active_open_tx_fifo;
 161   session_fifo_event_t evt;
 162
 163   ASSERT (s->thread_index == thread_index);
 164
 165   clib_spinlock_lock_if_init (&pm->sessions_lock);
 166   p = hash_get (pm->proxy_session_by_server_handle, session_handle (s));
 167
 168   if (PREDICT_TRUE (p != 0))
 169     {
 170       clib_spinlock_unlock_if_init (&pm->sessions_lock);
 171       active_open_tx_fifo = s->server_rx_fifo;
 172
 173       /*
 174        * Send event for active open tx fifo
 175        */
 176       if (svm_fifo_set_event (active_open_tx_fifo))
 177         {
 178           evt.fifo = active_open_tx_fifo;
 179           evt.event_type = FIFO_EVENT_APP_TX;
 180           if (svm_queue_add
 181               (pm->active_open_event_queue[thread_index], (u8 *) & evt,
 182                0 /* do wait for mutex */ ))
 183             clib_warning ("failed to enqueue tx evt");
 184         }
 185     }
 186   else
 187     {
 188       rx_fifo = s->server_rx_fifo;
 189       tx_fifo = s->server_tx_fifo;
 190
 191       ASSERT (rx_fifo->master_thread_index == thread_index);
 192       ASSERT (tx_fifo->master_thread_index == thread_index);
 193
 194       max_dequeue = svm_fifo_max_dequeue (s->server_rx_fifo);
 195
 196       if (PREDICT_FALSE (max_dequeue == 0))
 197         return 0;
 198
 199       actual_transfer = svm_fifo_peek (rx_fifo, 0 /* relative_offset */ ,
 200                                        max_dequeue, pm->rx_buf[thread_index]);
 201
 202       /* $$$ your message in this space: parse url, etc. */
 203
 204       memset (a, 0, sizeof (*a));
 205
 206       clib_spinlock_lock_if_init (&pm->sessions_lock);
 207       pool_get (pm->sessions, ps);
 208       memset (ps, 0, sizeof (*ps));
 209       ps->server_rx_fifo = rx_fifo;
 210       ps->server_tx_fifo = tx_fifo;
 211       ps->vpp_server_handle = session_handle (s);
 212
 213       proxy_index = ps - pm->sessions;
 214
 215       hash_set (pm->proxy_session_by_server_handle, ps->vpp_server_handle,
 216                 proxy_index);
 217
 218       clib_spinlock_unlock_if_init (&pm->sessions_lock);
 219
 220       a->uri = (char *) pm->client_uri;
 221       a->api_context = proxy_index;
 222       a->app_index = pm->active_open_app_index;
 223       a->mp = 0;
 224       vnet_connect_uri (a);
 225     }
 226
 227   return 0;
 228 }
 229
 230 static session_cb_vft_t proxy_session_cb_vft = {
 231   .session_accept_callback = proxy_accept_callback,
 232   .session_disconnect_callback = proxy_disconnect_callback,
 233   .session_connected_callback = proxy_connected_callback,
 234   .add_segment_callback = proxy_add_segment_callback,
 235   .builtin_server_rx_callback = proxy_rx_callback,
 236   .session_reset_callback = proxy_reset_callback
 237 };
 238
 239 static int
 240 active_open_connected_callback (u32 app_index, u32 opaque,
 241                                 stream_session_t * s, u8 is_fail)
 242 {
 243   proxy_main_t *pm = &proxy_main;
 244   proxy_session_t *ps;
 245   u8 thread_index = vlib_get_thread_index ();
 246   session_fifo_event_t evt;
 247
 248   if (is_fail)
 249     {
 250       clib_warning ("connection %d failed!", opaque);
 251       return 0;
 252     }
 253
 254   /*
 255    * Setup proxy session handle.
 256    */
 257   clib_spinlock_lock_if_init (&pm->sessions_lock);
 258
 259   ps = pool_elt_at_index (pm->sessions, opaque);
 260   ps->vpp_active_open_handle = session_handle (s);
 261
 262   s->server_tx_fifo = ps->server_rx_fifo;
 263   s->server_rx_fifo = ps->server_tx_fifo;
 264
 265   /*
 266    * Reset the active-open tx-fifo master indices so the active-open session
 267    * will receive data, etc.
 268    */
 269   s->server_tx_fifo->master_session_index = s->session_index;
 270   s->server_tx_fifo->master_thread_index = s->thread_index;
 271
 272   /*
 273    * Account for the active-open session's use of the fifos
 274    * so they won't disappear until the last session which uses
 275    * them disappears
 276    */
 277   s->server_tx_fifo->refcnt++;
 278   s->server_rx_fifo->refcnt++;
 279
 280   hash_set (pm->proxy_session_by_active_open_handle,
 281             ps->vpp_active_open_handle, opaque);
 282
 283   clib_spinlock_unlock_if_init (&pm->sessions_lock);
 284
 285   /*
 286    * Send event for active open tx fifo
 287    */
 288   if (svm_fifo_set_event (s->server_tx_fifo))
 289     {
 290       evt.fifo = s->server_tx_fifo;
 291       evt.event_type = FIFO_EVENT_APP_TX;
 292       if (svm_queue_add
 293           (pm->active_open_event_queue[thread_index], (u8 *) & evt,
 294            0 /* do wait for mutex */ ))
 295         clib_warning ("failed to enqueue tx evt");
 296     }
 297
 298   return 0;
 299 }
 300
 301 static void
 302 active_open_reset_callback (stream_session_t * s)
 303 {
 304   delete_proxy_session (s, 1 /* is_active_open */ );
 305 }
 306
 307 static int
 308 active_open_create_callback (stream_session_t * s)
 309 {
 310   return 0;
 311 }
 312
 313 static void
 314 active_open_disconnect_callback (stream_session_t * s)
 315 {
 316   delete_proxy_session (s, 1 /* is_active_open */ );
 317 }
 318
 319 static int
 320 active_open_rx_callback (stream_session_t * s)
 321 {
 322   proxy_main_t *pm = &proxy_main;
 323   session_fifo_event_t evt;
 324   svm_fifo_t *server_rx_fifo;
 325   u32 thread_index = vlib_get_thread_index ();
 326
 327   server_rx_fifo = s->server_rx_fifo;
 328
 329   /*
 330    * Send event for server tx fifo
 331    */
 332   if (svm_fifo_set_event (server_rx_fifo))
 333     {
 334       evt.fifo = server_rx_fifo;
 335       evt.event_type = FIFO_EVENT_APP_TX;
 336       if (svm_queue_add
 337           (pm->server_event_queue[thread_index], (u8 *) & evt,
 338            0 /* do wait for mutex */ ))
 339         clib_warning ("failed to enqueue server rx evt");
 340     }
 341
 342   return 0;
 343 }
 344
 345 /* *INDENT-OFF* */
 346 static session_cb_vft_t active_open_clients = {
 347   .session_reset_callback = active_open_reset_callback,
 348   .session_connected_callback = active_open_connected_callback,
 349   .session_accept_callback = active_open_create_callback,
 350   .session_disconnect_callback = active_open_disconnect_callback,
 351   .builtin_server_rx_callback = active_open_rx_callback
 352 };
 353 /* *INDENT-ON* */
 354
 355
 356 static void
 357 create_api_loopbacks (vlib_main_t * vm)
 358 {
 359   proxy_main_t *pm = &proxy_main;
 360   api_main_t *am = &api_main;
 361   vl_shmem_hdr_t *shmem_hdr;
 362
 363   shmem_hdr = am->shmem_hdr;
 364   pm->vl_input_queue = shmem_hdr->vl_input_queue;
 365   pm->server_client_index =
 366     vl_api_memclnt_create_internal ("proxy_server", pm->vl_input_queue);
 367   pm->active_open_client_index =
 368     vl_api_memclnt_create_internal ("proxy_active_open", pm->vl_input_queue);
 369 }
 370
 371 static int
 372 proxy_server_attach ()
 373 {
 374   proxy_main_t *pm = &proxy_main;
 375   u64 options[APP_OPTIONS_N_OPTIONS];
 376   vnet_app_attach_args_t _a, *a = &_a;
 377   u32 segment_size = 512 << 20;
 378
 379   memset (a, 0, sizeof (*a));
 380   memset (options, 0, sizeof (options));
 381
 382   if (pm->private_segment_size)
 383     segment_size = pm->private_segment_size;
 384   a->api_client_index = pm->server_client_index;
 385   a->session_cb_vft = &proxy_session_cb_vft;
 386   a->options = options;
 387   a->options[APP_OPTIONS_SEGMENT_SIZE] = segment_size;
 388   a->options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size;
 389   a->options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size;
 390   a->options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count;
 391   a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
 392     pm->prealloc_fifos ? pm->prealloc_fifos : 1;
 393
 394   a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN;
 395
 396   if (vnet_application_attach (a))
 397     {
 398       clib_warning ("failed to attach server");
 399       return -1;
 400     }
 401   pm->server_app_index = a->app_index;
 402
 403   return 0;
 404 }
 405
 406 static int
 407 active_open_attach (void)
 408 {
 409   proxy_main_t *pm = &proxy_main;
 410   vnet_app_attach_args_t _a, *a = &_a;
 411   u64 options[16];
 412
 413   memset (a, 0, sizeof (*a));
 414   memset (options, 0, sizeof (options));
 415
 416   a->api_client_index = pm->active_open_client_index;
 417   a->session_cb_vft = &active_open_clients;
 418
 419   options[APP_OPTIONS_ACCEPT_COOKIE] = 0x12345678;
 420   options[APP_OPTIONS_SEGMENT_SIZE] = 512 << 20;
 421   options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size;
 422   options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size;
 423   options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count;
 424   options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
 425     pm->prealloc_fifos ? pm->prealloc_fifos : 1;
 426
 427   options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN
 428     | APP_OPTIONS_FLAGS_IS_PROXY;
 429
 430   a->options = options;
 431
 432   if (vnet_application_attach (a))
 433     return -1;
 434
 435   pm->active_open_app_index = a->app_index;
 436
 437   return 0;
 438 }
 439
 440 static int
 441 proxy_server_listen ()
 442 {
 443   proxy_main_t *pm = &proxy_main;
 444   vnet_bind_args_t _a, *a = &_a;
 445   memset (a, 0, sizeof (*a));
 446   a->app_index = pm->server_app_index;
 447   a->uri = (char *) pm->server_uri;
 448   return vnet_bind_uri (a);
 449 }
 450
 451 static int
 452 proxy_server_create (vlib_main_t * vm)
 453 {
 454   proxy_main_t *pm = &proxy_main;
 455   vlib_thread_main_t *vtm = vlib_get_thread_main ();
 456   u32 num_threads;
 457   int i;
 458
 459   if (pm->server_client_index == (u32) ~ 0)
 460     create_api_loopbacks (vm);
 461
 462   num_threads = 1 /* main thread */  + vtm->n_threads;
 463   vec_validate (proxy_main.server_event_queue, num_threads - 1);
 464   vec_validate (proxy_main.active_open_event_queue, num_threads - 1);
 465   vec_validate (pm->rx_buf, num_threads - 1);
 466
 467   for (i = 0; i < num_threads; i++)
 468     vec_validate (pm->rx_buf[i], pm->rcv_buffer_size);
 469
 470   if (proxy_server_attach ())
 471     {
 472       clib_warning ("failed to attach server app");
 473       return -1;
 474     }
 475   if (proxy_server_listen ())
 476     {
 477       clib_warning ("failed to start listening");
 478       return -1;
 479     }
 480   if (active_open_attach ())
 481     {
 482       clib_warning ("failed to attach active open app");
 483       return -1;
 484     }
 485
 486   for (i = 0; i < num_threads; i++)
 487     {
 488       pm->active_open_event_queue[i] =
 489         session_manager_get_vpp_event_queue (i);
 490
 491       ASSERT (pm->active_open_event_queue[i]);
 492
 493       pm->server_event_queue[i] = session_manager_get_vpp_event_queue (i);
 494     }
 495
 496   return 0;
 497 }
 498
 499 static clib_error_t *
 500 proxy_server_create_command_fn (vlib_main_t * vm, unformat_input_t * input,
 501                                 vlib_cli_command_t * cmd)
 502 {
 503   proxy_main_t *pm = &proxy_main;
 504   char *default_server_uri = "tcp://0.0.0.0/23";
 505   char *default_client_uri = "tcp://6.0.2.2/23";
 506   int rv;
 507   u64 tmp;
 508
 509   pm->fifo_size = 64 << 10;
 510   pm->rcv_buffer_size = 1024;
 511   pm->prealloc_fifos = 0;
 512   pm->private_segment_count = 0;
 513   pm->private_segment_size = 0;
 514   pm->server_uri = 0;
 515
 516   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
 517     {
 518       if (unformat (input, "fifo-size %d", &pm->fifo_size))
 519         pm->fifo_size <<= 10;
 520       else if (unformat (input, "rcv-buf-size %d", &pm->rcv_buffer_size))
 521         ;
 522       else if (unformat (input, "prealloc-fifos %d", &pm->prealloc_fifos))
 523         ;
 524       else if (unformat (input, "private-segment-count %d",
 525                          &pm->private_segment_count))
 526         ;
 527       else if (unformat (input, "private-segment-size %U",
 528                          unformat_memory_size, &tmp))
 529         {
 530           if (tmp >= 0x100000000ULL)
 531             return clib_error_return
 532               (0, "private segment size %lld (%llu) too large", tmp, tmp);
 533           pm->private_segment_size = tmp;
 534         }
 535       else if (unformat (input, "server-uri %s", &pm->server_uri))
 536         ;
 537       else if (unformat (input, "client-uri %s", &pm->client_uri))
 538         ;
 539       else
 540         return clib_error_return (0, "unknown input `%U'",
 541                                   format_unformat_error, input);
 542     }
 543
 544   if (!pm->server_uri)
 545     {
 546       clib_warning ("No server-uri provided, Using default: %s",
 547                     default_server_uri);
 548       pm->server_uri = format (0, "%s%c", default_server_uri, 0);
 549     }
 550   if (!pm->client_uri)
 551     {
 552       clib_warning ("No client-uri provided, Using default: %s",
 553                     default_client_uri);
 554       pm->client_uri = format (0, "%s%c", default_client_uri, 0);
 555     }
 556
 557   vnet_session_enable_disable (vm, 1 /* turn on session and transport */ );
 558
 559   rv = proxy_server_create (vm);
 560   switch (rv)
 561     {
 562     case 0:
 563       break;
 564     default:
 565       return clib_error_return (0, "server_create returned %d", rv);
 566     }
 567
 568   return 0;
 569 }
 570
 571 /* *INDENT-OFF* */
 572 VLIB_CLI_COMMAND (proxy_create_command, static) =
 573 {
 574   .path = "test proxy server",
 575   .short_help = "test proxy server [server-uri <tcp://ip/port>]"
 576       "[client-uri <tcp://ip/port>][fifo-size <nn>][rcv-buf-size <nn>]"
 577       "[prealloc-fifos <nn>][private-segment-size <mem>]"
 578       "[private-segment-count <nn>]",
 579   .function = proxy_server_create_command_fn,
 580 };
 581 /* *INDENT-ON* */
 582
 583 clib_error_t *
 584 proxy_main_init (vlib_main_t * vm)
 585 {
 586   proxy_main_t *pm = &proxy_main;
 587   pm->server_client_index = ~0;
 588   pm->active_open_client_index = ~0;
 589   pm->proxy_session_by_active_open_handle = hash_create (0, sizeof (uword));
 590   pm->proxy_session_by_server_handle = hash_create (0, sizeof (uword));
 591
 592   return 0;
 593 }
 594
 595 VLIB_INIT_FUNCTION (proxy_main_init);
 596
 597 /*
 598 * fd.io coding-style-patch-verification: ON
 599 *
 600 * Local Variables:
 601 * eval: (c-set-style "gnu")
 602 * End:
 603 */