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       vnet_connect_uri (a);
 224     }
 225
 226   return 0;
 227 }
 228
 229 static session_cb_vft_t proxy_session_cb_vft = {
 230   .session_accept_callback = proxy_accept_callback,
 231   .session_disconnect_callback = proxy_disconnect_callback,
 232   .session_connected_callback = proxy_connected_callback,
 233   .add_segment_callback = proxy_add_segment_callback,
 234   .builtin_app_rx_callback = proxy_rx_callback,
 235   .session_reset_callback = proxy_reset_callback
 236 };
 237
 238 static int
 239 active_open_connected_callback (u32 app_index, u32 opaque,
 240                                 stream_session_t * s, u8 is_fail)
 241 {
 242   proxy_main_t *pm = &proxy_main;
 243   proxy_session_t *ps;
 244   u8 thread_index = vlib_get_thread_index ();
 245   session_fifo_event_t evt;
 246
 247   if (is_fail)
 248     {
 249       clib_warning ("connection %d failed!", opaque);
 250       return 0;
 251     }
 252
 253   /*
 254    * Setup proxy session handle.
 255    */
 256   clib_spinlock_lock_if_init (&pm->sessions_lock);
 257
 258   ps = pool_elt_at_index (pm->sessions, opaque);
 259   ps->vpp_active_open_handle = session_handle (s);
 260
 261   s->server_tx_fifo = ps->server_rx_fifo;
 262   s->server_rx_fifo = ps->server_tx_fifo;
 263
 264   /*
 265    * Reset the active-open tx-fifo master indices so the active-open session
 266    * will receive data, etc.
 267    */
 268   s->server_tx_fifo->master_session_index = s->session_index;
 269   s->server_tx_fifo->master_thread_index = s->thread_index;
 270
 271   /*
 272    * Account for the active-open session's use of the fifos
 273    * so they won't disappear until the last session which uses
 274    * them disappears
 275    */
 276   s->server_tx_fifo->refcnt++;
 277   s->server_rx_fifo->refcnt++;
 278
 279   hash_set (pm->proxy_session_by_active_open_handle,
 280             ps->vpp_active_open_handle, opaque);
 281
 282   clib_spinlock_unlock_if_init (&pm->sessions_lock);
 283
 284   /*
 285    * Send event for active open tx fifo
 286    */
 287   if (svm_fifo_set_event (s->server_tx_fifo))
 288     {
 289       evt.fifo = s->server_tx_fifo;
 290       evt.event_type = FIFO_EVENT_APP_TX;
 291       if (svm_queue_add
 292           (pm->active_open_event_queue[thread_index], (u8 *) & evt,
 293            0 /* do wait for mutex */ ))
 294         clib_warning ("failed to enqueue tx evt");
 295     }
 296
 297   return 0;
 298 }
 299
 300 static void
 301 active_open_reset_callback (stream_session_t * s)
 302 {
 303   delete_proxy_session (s, 1 /* is_active_open */ );
 304 }
 305
 306 static int
 307 active_open_create_callback (stream_session_t * s)
 308 {
 309   return 0;
 310 }
 311
 312 static void
 313 active_open_disconnect_callback (stream_session_t * s)
 314 {
 315   delete_proxy_session (s, 1 /* is_active_open */ );
 316 }
 317
 318 static int
 319 active_open_rx_callback (stream_session_t * s)
 320 {
 321   proxy_main_t *pm = &proxy_main;
 322   session_fifo_event_t evt;
 323   svm_fifo_t *server_rx_fifo;
 324   u32 thread_index = vlib_get_thread_index ();
 325
 326   server_rx_fifo = s->server_rx_fifo;
 327
 328   /*
 329    * Send event for server tx fifo
 330    */
 331   if (svm_fifo_set_event (server_rx_fifo))
 332     {
 333       evt.fifo = server_rx_fifo;
 334       evt.event_type = FIFO_EVENT_APP_TX;
 335       if (svm_queue_add
 336           (pm->server_event_queue[thread_index], (u8 *) & evt,
 337            0 /* do wait for mutex */ ))
 338         clib_warning ("failed to enqueue server rx evt");
 339     }
 340
 341   return 0;
 342 }
 343
 344 /* *INDENT-OFF* */
 345 static session_cb_vft_t active_open_clients = {
 346   .session_reset_callback = active_open_reset_callback,
 347   .session_connected_callback = active_open_connected_callback,
 348   .session_accept_callback = active_open_create_callback,
 349   .session_disconnect_callback = active_open_disconnect_callback,
 350   .builtin_app_rx_callback = active_open_rx_callback
 351 };
 352 /* *INDENT-ON* */
 353
 354
 355 static void
 356 create_api_loopbacks (vlib_main_t * vm)
 357 {
 358   proxy_main_t *pm = &proxy_main;
 359   api_main_t *am = &api_main;
 360   vl_shmem_hdr_t *shmem_hdr;
 361
 362   shmem_hdr = am->shmem_hdr;
 363   pm->vl_input_queue = shmem_hdr->vl_input_queue;
 364   pm->server_client_index =
 365     vl_api_memclnt_create_internal ("proxy_server", pm->vl_input_queue);
 366   pm->active_open_client_index =
 367     vl_api_memclnt_create_internal ("proxy_active_open", pm->vl_input_queue);
 368 }
 369
 370 static int
 371 proxy_server_attach ()
 372 {
 373   proxy_main_t *pm = &proxy_main;
 374   u64 options[APP_OPTIONS_N_OPTIONS];
 375   vnet_app_attach_args_t _a, *a = &_a;
 376   u32 segment_size = 512 << 20;
 377
 378   memset (a, 0, sizeof (*a));
 379   memset (options, 0, sizeof (options));
 380
 381   if (pm->private_segment_size)
 382     segment_size = pm->private_segment_size;
 383   a->api_client_index = pm->server_client_index;
 384   a->session_cb_vft = &proxy_session_cb_vft;
 385   a->options = options;
 386   a->options[APP_OPTIONS_SEGMENT_SIZE] = segment_size;
 387   a->options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size;
 388   a->options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size;
 389   a->options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count;
 390   a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
 391     pm->prealloc_fifos ? pm->prealloc_fifos : 1;
 392
 393   a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN;
 394
 395   if (vnet_application_attach (a))
 396     {
 397       clib_warning ("failed to attach server");
 398       return -1;
 399     }
 400   pm->server_app_index = a->app_index;
 401
 402   return 0;
 403 }
 404
 405 static int
 406 active_open_attach (void)
 407 {
 408   proxy_main_t *pm = &proxy_main;
 409   vnet_app_attach_args_t _a, *a = &_a;
 410   u64 options[16];
 411
 412   memset (a, 0, sizeof (*a));
 413   memset (options, 0, sizeof (options));
 414
 415   a->api_client_index = pm->active_open_client_index;
 416   a->session_cb_vft = &active_open_clients;
 417
 418   options[APP_OPTIONS_ACCEPT_COOKIE] = 0x12345678;
 419   options[APP_OPTIONS_SEGMENT_SIZE] = 512 << 20;
 420   options[APP_OPTIONS_RX_FIFO_SIZE] = pm->fifo_size;
 421   options[APP_OPTIONS_TX_FIFO_SIZE] = pm->fifo_size;
 422   options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = pm->private_segment_count;
 423   options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
 424     pm->prealloc_fifos ? pm->prealloc_fifos : 1;
 425
 426   options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_IS_BUILTIN
 427     | APP_OPTIONS_FLAGS_IS_PROXY;
 428
 429   a->options = options;
 430
 431   if (vnet_application_attach (a))
 432     return -1;
 433
 434   pm->active_open_app_index = a->app_index;
 435
 436   return 0;
 437 }
 438
 439 static int
 440 proxy_server_listen ()
 441 {
 442   proxy_main_t *pm = &proxy_main;
 443   vnet_bind_args_t _a, *a = &_a;
 444   memset (a, 0, sizeof (*a));
 445   a->app_index = pm->server_app_index;
 446   a->uri = (char *) pm->server_uri;
 447   return vnet_bind_uri (a);
 448 }
 449
 450 static int
 451 proxy_server_create (vlib_main_t * vm)
 452 {
 453   proxy_main_t *pm = &proxy_main;
 454   vlib_thread_main_t *vtm = vlib_get_thread_main ();
 455   u32 num_threads;
 456   int i;
 457
 458   if (pm->server_client_index == (u32) ~ 0)
 459     create_api_loopbacks (vm);
 460
 461   num_threads = 1 /* main thread */  + vtm->n_threads;
 462   vec_validate (proxy_main.server_event_queue, num_threads - 1);
 463   vec_validate (proxy_main.active_open_event_queue, num_threads - 1);
 464   vec_validate (pm->rx_buf, num_threads - 1);
 465
 466   for (i = 0; i < num_threads; i++)
 467     vec_validate (pm->rx_buf[i], pm->rcv_buffer_size);
 468
 469   if (proxy_server_attach ())
 470     {
 471       clib_warning ("failed to attach server app");
 472       return -1;
 473     }
 474   if (proxy_server_listen ())
 475     {
 476       clib_warning ("failed to start listening");
 477       return -1;
 478     }
 479   if (active_open_attach ())
 480     {
 481       clib_warning ("failed to attach active open app");
 482       return -1;
 483     }
 484
 485   for (i = 0; i < num_threads; i++)
 486     {
 487       pm->active_open_event_queue[i] =
 488         session_manager_get_vpp_event_queue (i);
 489
 490       ASSERT (pm->active_open_event_queue[i]);
 491
 492       pm->server_event_queue[i] = session_manager_get_vpp_event_queue (i);
 493     }
 494
 495   return 0;
 496 }
 497
 498 static clib_error_t *
 499 proxy_server_create_command_fn (vlib_main_t * vm, unformat_input_t * input,
 500                                 vlib_cli_command_t * cmd)
 501 {
 502   proxy_main_t *pm = &proxy_main;
 503   char *default_server_uri = "tcp://0.0.0.0/23";
 504   char *default_client_uri = "tcp://6.0.2.2/23";
 505   int rv;
 506   u64 tmp;
 507
 508   pm->fifo_size = 64 << 10;
 509   pm->rcv_buffer_size = 1024;
 510   pm->prealloc_fifos = 0;
 511   pm->private_segment_count = 0;
 512   pm->private_segment_size = 0;
 513   pm->server_uri = 0;
 514
 515   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
 516     {
 517       if (unformat (input, "fifo-size %d", &pm->fifo_size))
 518         pm->fifo_size <<= 10;
 519       else if (unformat (input, "rcv-buf-size %d", &pm->rcv_buffer_size))
 520         ;
 521       else if (unformat (input, "prealloc-fifos %d", &pm->prealloc_fifos))
 522         ;
 523       else if (unformat (input, "private-segment-count %d",
 524                          &pm->private_segment_count))
 525         ;
 526       else if (unformat (input, "private-segment-size %U",
 527                          unformat_memory_size, &tmp))
 528         {
 529           if (tmp >= 0x100000000ULL)
 530             return clib_error_return
 531               (0, "private segment size %lld (%llu) too large", tmp, tmp);
 532           pm->private_segment_size = tmp;
 533         }
 534       else if (unformat (input, "server-uri %s", &pm->server_uri))
 535         ;
 536       else if (unformat (input, "client-uri %s", &pm->client_uri))
 537         ;
 538       else
 539         return clib_error_return (0, "unknown input `%U'",
 540                                   format_unformat_error, input);
 541     }
 542
 543   if (!pm->server_uri)
 544     {
 545       clib_warning ("No server-uri provided, Using default: %s",
 546                     default_server_uri);
 547       pm->server_uri = format (0, "%s%c", default_server_uri, 0);
 548     }
 549   if (!pm->client_uri)
 550     {
 551       clib_warning ("No client-uri provided, Using default: %s",
 552                     default_client_uri);
 553       pm->client_uri = format (0, "%s%c", default_client_uri, 0);
 554     }
 555
 556   vnet_session_enable_disable (vm, 1 /* turn on session and transport */ );
 557
 558   rv = proxy_server_create (vm);
 559   switch (rv)
 560     {
 561     case 0:
 562       break;
 563     default:
 564       return clib_error_return (0, "server_create returned %d", rv);
 565     }
 566
 567   return 0;
 568 }
 569
 570 /* *INDENT-OFF* */
 571 VLIB_CLI_COMMAND (proxy_create_command, static) =
 572 {
 573   .path = "test proxy server",
 574   .short_help = "test proxy server [server-uri <tcp://ip/port>]"
 575       "[client-uri <tcp://ip/port>][fifo-size <nn>][rcv-buf-size <nn>]"
 576       "[prealloc-fifos <nn>][private-segment-size <mem>]"
 577       "[private-segment-count <nn>]",
 578   .function = proxy_server_create_command_fn,
 579 };
 580 /* *INDENT-ON* */
 581
 582 clib_error_t *
 583 proxy_main_init (vlib_main_t * vm)
 584 {
 585   proxy_main_t *pm = &proxy_main;
 586   pm->server_client_index = ~0;
 587   pm->active_open_client_index = ~0;
 588   pm->proxy_session_by_active_open_handle = hash_create (0, sizeof (uword));
 589   pm->proxy_session_by_server_handle = hash_create (0, sizeof (uword));
 590
 591   return 0;
 592 }
 593
 594 VLIB_INIT_FUNCTION (proxy_main_init);
 595
 596 /*
 597 * fd.io coding-style-patch-verification: ON
 598 *
 599 * Local Variables:
 600 * eval: (c-set-style "gnu")
 601 * End:
 602 */