src/vnet/tcp/builtin_server.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
  21 /* define message IDs */
  22 #include <vpp/api/vpe_msg_enum.h>
  23
  24 /* define message structures */
  25 #define vl_typedefs
  26 #include <vpp/api/vpe_all_api_h.h>
  27 #undef vl_typedefs
  28
  29 /* define generated endian-swappers */
  30 #define vl_endianfun
  31 #include <vpp/api/vpe_all_api_h.h>
  32 #undef vl_endianfun
  33
  34 /* instantiate all the print functions we know about */
  35 #define vl_print(handle, ...) vlib_cli_output (handle, __VA_ARGS__)
  36 #define vl_printfun
  37 #include <vpp/api/vpe_all_api_h.h>
  38 #undef vl_printfun
  39
  40 typedef struct
  41 {
  42   /*
  43    * Server app parameters
  44    */
  45   unix_shared_memory_queue_t **vpp_queue;
  46   unix_shared_memory_queue_t *vl_input_queue;   /**< Sever's event queue */
  47
  48   u32 app_index;                /**< Server app index */
  49   u32 my_client_index;          /**< API client handle */
  50   u32 node_index;               /**< process node index for evnt scheduling */
  51
  52   /*
  53    * Config params
  54    */
  55   u8 no_echo;                   /**< Don't echo traffic */
  56   u32 fifo_size;                /**< Fifo size */
  57   u32 rcv_buffer_size;          /**< Rcv buffer size */
  58   u32 prealloc_fifos;           /**< Preallocate fifos */
  59
  60   /*
  61    * Test state
  62    */
  63   u8 **rx_buf;                  /**< Per-thread RX buffer */
  64   u64 byte_index;
  65
  66   vlib_main_t *vlib_main;
  67 } builtin_server_main_t;
  68
  69 builtin_server_main_t builtin_server_main;
  70
  71 int
  72 builtin_session_accept_callback (stream_session_t * s)
  73 {
  74   builtin_server_main_t *bsm = &builtin_server_main;
  75
  76   bsm->vpp_queue[s->thread_index] =
  77     session_manager_get_vpp_event_queue (s->thread_index);
  78   s->session_state = SESSION_STATE_READY;
  79   bsm->byte_index = 0;
  80   return 0;
  81 }
  82
  83 void
  84 builtin_session_disconnect_callback (stream_session_t * s)
  85 {
  86   builtin_server_main_t *bsm = &builtin_server_main;
  87   vnet_disconnect_args_t _a, *a = &_a;
  88
  89   a->handle = stream_session_handle (s);
  90   a->app_index = bsm->app_index;
  91   vnet_disconnect_session (a);
  92 }
  93
  94 void
  95 builtin_session_reset_callback (stream_session_t * s)
  96 {
  97   clib_warning ("called.. ");
  98
  99   stream_session_cleanup (s);
 100 }
 101
 102
 103 int
 104 builtin_session_connected_callback (u32 app_index, u32 api_context,
 105                                     stream_session_t * s, u8 is_fail)
 106 {
 107   clib_warning ("called...");
 108   return -1;
 109 }
 110
 111 int
 112 builtin_add_segment_callback (u32 client_index,
 113                               const u8 * seg_name, u32 seg_size)
 114 {
 115   clib_warning ("called...");
 116   return -1;
 117 }
 118
 119 int
 120 builtin_redirect_connect_callback (u32 client_index, void *mp)
 121 {
 122   clib_warning ("called...");
 123   return -1;
 124 }
 125
 126 void
 127 test_bytes (builtin_server_main_t * bsm, int actual_transfer)
 128 {
 129   int i;
 130   u32 my_thread_id = vlib_get_thread_index ();
 131
 132   for (i = 0; i < actual_transfer; i++)
 133     {
 134       if (bsm->rx_buf[my_thread_id][i] != ((bsm->byte_index + i) & 0xff))
 135         {
 136           clib_warning ("at %lld expected %d got %d", bsm->byte_index + i,
 137                         (bsm->byte_index + i) & 0xff,
 138                         bsm->rx_buf[my_thread_id][i]);
 139         }
 140     }
 141   bsm->byte_index += actual_transfer;
 142 }
 143
 144 /*
 145  * If no-echo, just read the data and be done with it
 146  */
 147 int
 148 builtin_server_rx_callback_no_echo (stream_session_t * s)
 149 {
 150   builtin_server_main_t *bsm = &builtin_server_main;
 151   u32 my_thread_id = vlib_get_thread_index ();
 152   int actual_transfer;
 153   svm_fifo_t *rx_fifo;
 154
 155   rx_fifo = s->server_rx_fifo;
 156
 157   do
 158     {
 159       actual_transfer =
 160         svm_fifo_dequeue_nowait (rx_fifo, bsm->rcv_buffer_size,
 161                                  bsm->rx_buf[my_thread_id]);
 162     }
 163   while (actual_transfer > 0);
 164   return 0;
 165 }
 166
 167 int
 168 builtin_server_rx_callback (stream_session_t * s)
 169 {
 170   u32 n_written, max_dequeue, max_enqueue, max_transfer;
 171   int actual_transfer;
 172   svm_fifo_t *tx_fifo, *rx_fifo;
 173   builtin_server_main_t *bsm = &builtin_server_main;
 174   session_fifo_event_t evt;
 175   static int serial_number = 0;
 176   u32 my_thread_id = vlib_get_thread_index ();
 177
 178   rx_fifo = s->server_rx_fifo;
 179   tx_fifo = s->server_tx_fifo;
 180
 181   max_dequeue = svm_fifo_max_dequeue (s->server_rx_fifo);
 182   max_enqueue = svm_fifo_max_enqueue (s->server_tx_fifo);
 183
 184   if (PREDICT_FALSE (max_dequeue == 0))
 185     return 0;
 186
 187   /* Number of bytes we're going to copy */
 188   max_transfer = (max_dequeue < max_enqueue) ? max_dequeue : max_enqueue;
 189
 190   /* No space in tx fifo */
 191   if (PREDICT_FALSE (max_transfer == 0))
 192     {
 193       /* XXX timeout for session that are stuck */
 194
 195     rx_event:
 196       /* Program self-tap to retry */
 197       if (svm_fifo_set_event (rx_fifo))
 198         {
 199           unix_shared_memory_queue_t *q;
 200           evt.fifo = rx_fifo;
 201           evt.event_type = FIFO_EVENT_BUILTIN_RX;
 202           evt.event_id = 0;
 203
 204           q = bsm->vpp_queue[s->thread_index];
 205           if (PREDICT_FALSE (q->cursize == q->maxsize))
 206             clib_warning ("out of event queue space");
 207           else
 208             unix_shared_memory_queue_add (q, (u8 *) & evt,
 209                                           0 /* don't wait for mutex */ );
 210         }
 211
 212       return 0;
 213     }
 214
 215   _vec_len (bsm->rx_buf[my_thread_id]) = max_transfer;
 216
 217   actual_transfer = svm_fifo_dequeue_nowait (rx_fifo, max_transfer,
 218                                              bsm->rx_buf[my_thread_id]);
 219   ASSERT (actual_transfer == max_transfer);
 220
 221 //  test_bytes (bsm, actual_transfer);
 222
 223   /*
 224    * Echo back
 225    */
 226
 227   n_written = svm_fifo_enqueue_nowait (tx_fifo, actual_transfer,
 228                                        bsm->rx_buf[my_thread_id]);
 229
 230   if (n_written != max_transfer)
 231     clib_warning ("short trout!");
 232
 233   if (svm_fifo_set_event (tx_fifo))
 234     {
 235       /* Fabricate TX event, send to vpp */
 236       evt.fifo = tx_fifo;
 237       evt.event_type = FIFO_EVENT_APP_TX;
 238       evt.event_id = serial_number++;
 239
 240       unix_shared_memory_queue_add (bsm->vpp_queue[s->thread_index],
 241                                     (u8 *) & evt, 0 /* do wait for mutex */ );
 242     }
 243
 244   if (PREDICT_FALSE (max_enqueue < max_dequeue))
 245     goto rx_event;
 246
 247   return 0;
 248 }
 249
 250 static session_cb_vft_t builtin_session_cb_vft = {
 251   .session_accept_callback = builtin_session_accept_callback,
 252   .session_disconnect_callback = builtin_session_disconnect_callback,
 253   .session_connected_callback = builtin_session_connected_callback,
 254   .add_segment_callback = builtin_add_segment_callback,
 255   .redirect_connect_callback = builtin_redirect_connect_callback,
 256   .builtin_server_rx_callback = builtin_server_rx_callback,
 257   .session_reset_callback = builtin_session_reset_callback
 258 };
 259
 260 /* Abuse VPP's input queue */
 261 static int
 262 create_api_loopback (vlib_main_t * vm)
 263 {
 264   builtin_server_main_t *bsm = &builtin_server_main;
 265   vl_api_memclnt_create_t _m, *mp = &_m;
 266   extern void vl_api_memclnt_create_t_handler (vl_api_memclnt_create_t *);
 267   api_main_t *am = &api_main;
 268   vl_shmem_hdr_t *shmem_hdr;
 269   uword *event_data = 0, event_type;
 270   int resolved = 0;
 271
 272   /*
 273    * Create a "loopback" API client connection
 274    * Don't do things like this unless you know what you're doing...
 275    */
 276
 277   shmem_hdr = am->shmem_hdr;
 278   bsm->vl_input_queue = shmem_hdr->vl_input_queue;
 279   memset (mp, 0, sizeof (*mp));
 280   mp->_vl_msg_id = VL_API_MEMCLNT_CREATE;
 281   mp->context = 0xFEEDFACE;
 282   mp->input_queue = pointer_to_uword (bsm->vl_input_queue);
 283   strncpy ((char *) mp->name, "tcp_test_server", sizeof (mp->name) - 1);
 284
 285   vl_api_memclnt_create_t_handler (mp);
 286
 287   /* Wait for reply */
 288   bsm->node_index = vlib_get_current_process (vm)->node_runtime.node_index;
 289   vlib_process_wait_for_event_or_clock (vm, 1.0);
 290   event_type = vlib_process_get_events (vm, &event_data);
 291   switch (event_type)
 292     {
 293     case 1:
 294       resolved = 1;
 295       break;
 296     case ~0:
 297       /* timed out */
 298       break;
 299     default:
 300       clib_warning ("unknown event_type %d", event_type);
 301     }
 302   if (!resolved)
 303     return -1;
 304
 305   return 0;
 306 }
 307
 308 static int
 309 server_attach ()
 310 {
 311   builtin_server_main_t *bsm = &builtin_server_main;
 312   u8 segment_name[128];
 313   u64 options[SESSION_OPTIONS_N_OPTIONS];
 314   vnet_app_attach_args_t _a, *a = &_a;
 315
 316   memset (a, 0, sizeof (*a));
 317   memset (options, 0, sizeof (options));
 318
 319   if (bsm->no_echo)
 320     builtin_session_cb_vft.builtin_server_rx_callback =
 321       builtin_server_rx_callback_no_echo;
 322   else
 323     builtin_session_cb_vft.builtin_server_rx_callback =
 324       builtin_server_rx_callback;
 325   a->api_client_index = bsm->my_client_index;
 326   a->session_cb_vft = &builtin_session_cb_vft;
 327   a->options = options;
 328   a->options[SESSION_OPTIONS_SEGMENT_SIZE] = 512 << 20;
 329   a->options[SESSION_OPTIONS_RX_FIFO_SIZE] = bsm->fifo_size;
 330   a->options[SESSION_OPTIONS_TX_FIFO_SIZE] = bsm->fifo_size;
 331   a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
 332   a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
 333     bsm->prealloc_fifos ? bsm->prealloc_fifos : 1;
 334   a->segment_name = segment_name;
 335   a->segment_name_length = ARRAY_LEN (segment_name);
 336
 337   if (vnet_application_attach (a))
 338     {
 339       clib_warning ("failed to attach server");
 340       return -1;
 341     }
 342   bsm->app_index = a->app_index;
 343   return 0;
 344 }
 345
 346 static int
 347 server_listen ()
 348 {
 349   builtin_server_main_t *bsm = &builtin_server_main;
 350   vnet_bind_args_t _a, *a = &_a;
 351   memset (a, 0, sizeof (*a));
 352   a->app_index = bsm->app_index;
 353   a->uri = "tcp://0.0.0.0/1234";
 354   return vnet_bind_uri (a);
 355 }
 356
 357 static int
 358 server_create (vlib_main_t * vm)
 359 {
 360   builtin_server_main_t *bsm = &builtin_server_main;
 361   vlib_thread_main_t *vtm = vlib_get_thread_main ();
 362   u32 num_threads;
 363   int i;
 364
 365   if (bsm->my_client_index == (u32) ~ 0)
 366     {
 367       if (create_api_loopback (vm))
 368         {
 369           clib_warning ("failed to create api loopback");
 370           return -1;
 371         }
 372     }
 373
 374   num_threads = 1 /* main thread */  + vtm->n_threads;
 375   vec_validate (builtin_server_main.vpp_queue, num_threads - 1);
 376   vec_validate (bsm->rx_buf, num_threads - 1);
 377   for (i = 0; i < num_threads; i++)
 378     vec_validate (bsm->rx_buf[i], bsm->rcv_buffer_size);
 379
 380   if (server_attach ())
 381     {
 382       clib_warning ("failed to attach server");
 383       return -1;
 384     }
 385   if (server_listen ())
 386     {
 387       clib_warning ("failed to start listening");
 388       return -1;
 389     }
 390   return 0;
 391 }
 392
 393 /* Get our api client index */
 394 static void
 395 vl_api_memclnt_create_reply_t_handler (vl_api_memclnt_create_reply_t * mp)
 396 {
 397   vlib_main_t *vm = vlib_get_main ();
 398   builtin_server_main_t *bsm = &builtin_server_main;
 399   bsm->my_client_index = mp->index;
 400   vlib_process_signal_event (vm, bsm->node_index, 1 /* evt */ ,
 401                              0 /* data */ );
 402 }
 403
 404 #define foreach_tcp_builtin_server_api_msg                      \
 405 _(MEMCLNT_CREATE_REPLY, memclnt_create_reply)                   \
 406
 407 static clib_error_t *
 408 tcp_builtin_server_api_hookup (vlib_main_t * vm)
 409 {
 410   vl_msg_api_msg_config_t _c, *c = &_c;
 411
 412   /* Hook up client-side static APIs to our handlers */
 413 #define _(N,n) do {                                             \
 414     c->id = VL_API_##N;                                         \
 415     c->name = #n;                                               \
 416     c->handler = vl_api_##n##_t_handler;                        \
 417     c->cleanup = vl_noop_handler;                               \
 418     c->endian = vl_api_##n##_t_endian;                          \
 419     c->print = vl_api_##n##_t_print;                            \
 420     c->size = sizeof(vl_api_##n##_t);                           \
 421     c->traced = 1; /* trace, so these msgs print */             \
 422     c->replay = 0; /* don't replay client create/delete msgs */ \
 423     c->message_bounce = 0; /* don't bounce this message */      \
 424     vl_msg_api_config(c);} while (0);
 425
 426   foreach_tcp_builtin_server_api_msg;
 427 #undef _
 428
 429   return 0;
 430 }
 431
 432 static clib_error_t *
 433 server_create_command_fn (vlib_main_t * vm, unformat_input_t * input,
 434                           vlib_cli_command_t * cmd)
 435 {
 436   builtin_server_main_t *bsm = &builtin_server_main;
 437   int rv;
 438
 439   bsm->no_echo = 0;
 440   bsm->fifo_size = 64 << 10;
 441   bsm->rcv_buffer_size = 128 << 10;
 442   bsm->prealloc_fifos = 0;
 443
 444   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
 445     {
 446       if (unformat (input, "no-echo"))
 447         bsm->no_echo = 1;
 448       else if (unformat (input, "fifo-size %d", &bsm->fifo_size))
 449         bsm->fifo_size <<= 10;
 450       else if (unformat (input, "rcv-buf-size %d", &bsm->rcv_buffer_size))
 451         ;
 452       else if (unformat (input, "prealloc-fifos", &bsm->prealloc_fifos))
 453         ;
 454       else
 455         return clib_error_return (0, "unknown input `%U'",
 456                                   format_unformat_error, input);
 457     }
 458
 459   tcp_builtin_server_api_hookup (vm);
 460   vnet_session_enable_disable (vm, 1 /* turn on TCP, etc. */ );
 461
 462   rv = server_create (vm);
 463   switch (rv)
 464     {
 465     case 0:
 466       break;
 467     default:
 468       return clib_error_return (0, "server_create returned %d", rv);
 469     }
 470
 471   return 0;
 472 }
 473
 474 /* *INDENT-OFF* */
 475 VLIB_CLI_COMMAND (server_create_command, static) =
 476 {
 477   .path = "test tcp server",
 478   .short_help = "test tcp server",
 479   .function = server_create_command_fn,
 480 };
 481 /* *INDENT-ON* */
 482
 483 clib_error_t *
 484 builtin_tcp_server_main_init (vlib_main_t * vm)
 485 {
 486   builtin_server_main_t *bsm = &builtin_server_main;
 487   bsm->my_client_index = ~0;
 488   return 0;
 489 }
 490
 491 VLIB_INIT_FUNCTION (builtin_tcp_server_main_init);
 492
 493 /*
 494 * fd.io coding-style-patch-verification: ON
 495 *
 496 * Local Variables:
 497 * eval: (c-set-style "gnu")
 498 * End:
 499 */