src/tests/vnet/session/udp_echo.c

   1 /*
   2  * Copyright (c) 2016 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 <stdio.h>
  17 #include <setjmp.h>
  18 #include <signal.h>
  19 #include <vppinfra/clib.h>
  20 #include <vppinfra/format.h>
  21 #include <vppinfra/error.h>
  22 #include <vppinfra/time.h>
  23 #include <vppinfra/macros.h>
  24 #include <vnet/vnet.h>
  25 #include <vlib/vlib.h>
  26 #include <vlib/unix/unix.h>
  27 #include <vlibapi/api.h>
  28 #include <vlibmemory/api.h>
  29 #include <vpp/api/vpe_msg_enum.h>
  30 #include <svm/svm_fifo_segment.h>
  31 #include <pthread.h>
  32 #include <vnet/session/application_interface.h>
  33
  34 #define vl_typedefs             /* define message structures */
  35 #include <vpp/api/vpe_all_api_h.h>
  36 #undef vl_typedefs
  37
  38 /* declare message handlers for each api */
  39
  40 #define vl_endianfun            /* define message structures */
  41 #include <vpp/api/vpe_all_api_h.h>
  42 #undef vl_endianfun
  43
  44 /* instantiate all the print functions we know about */
  45 #define vl_print(handle, ...)
  46 #define vl_printfun
  47 #include <vpp/api/vpe_all_api_h.h>
  48 #undef vl_printfun
  49
  50 typedef enum
  51 {
  52   STATE_START,
  53   STATE_BOUND,
  54   STATE_READY,
  55   STATE_FAILED,
  56   STATE_DISCONNECTING,
  57 } connection_state_t;
  58
  59 typedef struct
  60 {
  61   svm_fifo_t *server_rx_fifo;
  62   svm_fifo_t *server_tx_fifo;
  63 } session_t;
  64
  65 typedef struct
  66 {
  67   /* vpe input queue */
  68   svm_queue_t *vl_input_queue;
  69
  70   /* API client handle */
  71   u32 my_client_index;
  72
  73   /* The URI we're playing with */
  74   u8 *uri;
  75
  76   /* Session pool */
  77   session_t *sessions;
  78
  79   /* Hash table for disconnect processing */
  80   uword *session_index_by_vpp_handles;
  81
  82   /* fifo segment */
  83   svm_fifo_segment_private_t *seg;
  84
  85   /* intermediate rx buffer */
  86   u8 *rx_buf;
  87
  88   /* URI for connect */
  89   u8 *connect_uri;
  90
  91   int i_am_master;
  92
  93   /* Our event queue */
  94   svm_queue_t *our_event_queue;
  95
  96   /* $$$ single thread only for the moment */
  97   svm_queue_t *vpp_event_queue;
  98
  99   /* $$$$ hack: cut-through session index */
 100   volatile u32 cut_through_session_index;
 101   volatile u32 connected_session;
 102
 103   /* unique segment name counter */
 104   u32 unique_segment_index;
 105
 106   pid_t my_pid;
 107
 108   /* pthread handle */
 109   pthread_t cut_through_thread_handle;
 110
 111   /* For deadman timers */
 112   clib_time_t clib_time;
 113
 114   /* State of the connection, shared between msg RX thread and main thread */
 115   volatile connection_state_t state;
 116
 117   volatile int time_to_stop;
 118   volatile int time_to_print_stats;
 119
 120   u32 configured_segment_size;
 121
 122   /* VNET_API_ERROR_FOO -> "Foo" hash table */
 123   uword *error_string_by_error_number;
 124
 125   /* convenience */
 126   svm_fifo_segment_main_t *segment_main;
 127
 128   u8 *connect_test_data;
 129
 130   uword *segments_table;
 131 } uri_udp_test_main_t;
 132
 133 #if CLIB_DEBUG > 0
 134 #define NITER 10000
 135 #else
 136 #define NITER 4000000
 137 #endif
 138
 139 uri_udp_test_main_t uri_udp_test_main;
 140
 141 static void
 142 stop_signal (int signum)
 143 {
 144   uri_udp_test_main_t *um = &uri_udp_test_main;
 145
 146   um->time_to_stop = 1;
 147 }
 148
 149 static void
 150 stats_signal (int signum)
 151 {
 152   uri_udp_test_main_t *um = &uri_udp_test_main;
 153
 154   um->time_to_print_stats = 1;
 155 }
 156
 157 static clib_error_t *
 158 setup_signal_handlers (void)
 159 {
 160   signal (SIGINT, stats_signal);
 161   signal (SIGQUIT, stop_signal);
 162   signal (SIGTERM, stop_signal);
 163
 164   return 0;
 165 }
 166
 167 void
 168 application_send_attach (uri_udp_test_main_t * utm)
 169 {
 170   vl_api_application_attach_t *bmp;
 171   u32 fifo_size = 1 << 20;
 172   bmp = vl_msg_api_alloc (sizeof (*bmp));
 173   memset (bmp, 0, sizeof (*bmp));
 174
 175   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH);
 176   bmp->client_index = utm->my_client_index;
 177   bmp->context = ntohl (0xfeedface);
 178   bmp->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_ADD_SEGMENT;
 179   bmp->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_USE_GLOBAL_SCOPE;
 180   bmp->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_USE_LOCAL_SCOPE;
 181   bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 2;
 182   bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = fifo_size;
 183   bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = fifo_size;
 184   bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
 185   bmp->options[APP_OPTIONS_SEGMENT_SIZE] = 256 << 20;
 186   bmp->options[APP_OPTIONS_EVT_QUEUE_SIZE] = 16768;
 187   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
 188 }
 189
 190 void
 191 application_detach (uri_udp_test_main_t * utm)
 192 {
 193   vl_api_application_detach_t *bmp;
 194   bmp = vl_msg_api_alloc (sizeof (*bmp));
 195   memset (bmp, 0, sizeof (*bmp));
 196
 197   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH);
 198   bmp->client_index = utm->my_client_index;
 199   bmp->context = ntohl (0xfeedface);
 200   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
 201 }
 202
 203 static void
 204 vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
 205                                            mp)
 206 {
 207   uri_udp_test_main_t *utm = &uri_udp_test_main;
 208   svm_fifo_segment_create_args_t _a = { 0 }, *a = &_a;
 209   int rv;
 210
 211   if (mp->retval)
 212     {
 213       clib_warning ("attach failed: %d", mp->retval);
 214       utm->state = STATE_FAILED;
 215       return;
 216     }
 217
 218   if (mp->segment_name_length == 0)
 219     {
 220       clib_warning ("segment_name_length zero");
 221       return;
 222     }
 223
 224   a->segment_name = (char *) mp->segment_name;
 225   a->segment_size = mp->segment_size;
 226
 227   ASSERT (mp->app_event_queue_address);
 228
 229   /* Attach to the segment vpp created */
 230   rv = svm_fifo_segment_attach (a);
 231   if (rv)
 232     {
 233       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 234                     mp->segment_name);
 235       return;
 236     }
 237
 238   utm->our_event_queue =
 239     uword_to_pointer (mp->app_event_queue_address, svm_queue_t *);
 240 }
 241
 242 static void
 243 vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t *
 244                                            mp)
 245 {
 246   if (mp->retval)
 247     clib_warning ("detach returned with err: %d", mp->retval);
 248 }
 249
 250 u8 *
 251 format_api_error (u8 * s, va_list * args)
 252 {
 253   uri_udp_test_main_t *utm = va_arg (*args, uri_udp_test_main_t *);
 254   i32 error = va_arg (*args, u32);
 255   uword *p;
 256
 257   p = hash_get (utm->error_string_by_error_number, -error);
 258
 259   if (p)
 260     s = format (s, "%s", p[0]);
 261   else
 262     s = format (s, "%d", error);
 263   return s;
 264 }
 265
 266 int
 267 wait_for_state_change (uri_udp_test_main_t * utm, connection_state_t state)
 268 {
 269 #if CLIB_DEBUG > 0
 270 #define TIMEOUT 600.0
 271 #else
 272 #define TIMEOUT 600.0
 273 #endif
 274
 275   f64 timeout = clib_time_now (&utm->clib_time) + TIMEOUT;
 276
 277   while (clib_time_now (&utm->clib_time) < timeout)
 278     {
 279       if (utm->state == state)
 280         return 0;
 281     }
 282   return -1;
 283 }
 284
 285 u64 server_bytes_received, server_bytes_sent;
 286
 287 static void *
 288 cut_through_thread_fn (void *arg)
 289 {
 290   session_t *s;
 291   svm_fifo_t *rx_fifo;
 292   svm_fifo_t *tx_fifo;
 293   u8 *my_copy_buffer = 0;
 294   uri_udp_test_main_t *utm = &uri_udp_test_main;
 295   i32 actual_transfer;
 296   int rv;
 297   u32 buffer_offset;
 298
 299   while (utm->cut_through_session_index == ~0)
 300     ;
 301
 302   s = pool_elt_at_index (utm->sessions, utm->cut_through_session_index);
 303
 304   rx_fifo = s->server_rx_fifo;
 305   tx_fifo = s->server_tx_fifo;
 306
 307   vec_validate (my_copy_buffer, 64 * 1024 - 1);
 308
 309   while (true)
 310     {
 311       /* We read from the tx fifo and write to the rx fifo */
 312       do
 313         {
 314           actual_transfer = svm_fifo_dequeue_nowait (rx_fifo,
 315                                                      vec_len (my_copy_buffer),
 316                                                      my_copy_buffer);
 317         }
 318       while (actual_transfer <= 0);
 319
 320       server_bytes_received += actual_transfer;
 321
 322       buffer_offset = 0;
 323       while (actual_transfer > 0)
 324         {
 325           rv = svm_fifo_enqueue_nowait (tx_fifo, actual_transfer,
 326                                         my_copy_buffer + buffer_offset);
 327           if (rv > 0)
 328             {
 329               actual_transfer -= rv;
 330               buffer_offset += rv;
 331               server_bytes_sent += rv;
 332             }
 333
 334         }
 335       if (PREDICT_FALSE (utm->time_to_stop))
 336         break;
 337     }
 338
 339   pthread_exit (0);
 340 }
 341
 342 static void
 343 udp_client_connect (uri_udp_test_main_t * utm)
 344 {
 345   vl_api_connect_uri_t *cmp;
 346   cmp = vl_msg_api_alloc (sizeof (*cmp));
 347   memset (cmp, 0, sizeof (*cmp));
 348
 349   cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
 350   cmp->client_index = utm->my_client_index;
 351   cmp->context = ntohl (0xfeedface);
 352   memcpy (cmp->uri, utm->connect_uri, vec_len (utm->connect_uri));
 353   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & cmp);
 354 }
 355
 356 static void
 357 client_send_cut_through (uri_udp_test_main_t * utm, session_t * session)
 358 {
 359   int i;
 360   u8 *test_data = 0;
 361   u64 bytes_received = 0, bytes_sent = 0;
 362   i32 bytes_to_read;
 363   int rv;
 364   f64 before, after, delta, bytes_per_second;
 365   svm_fifo_t *rx_fifo, *tx_fifo;
 366   int buffer_offset, bytes_to_send = 0;
 367
 368   /*
 369    * Prepare test data
 370    */
 371   vec_validate (test_data, 64 * 1024 - 1);
 372   for (i = 0; i < vec_len (test_data); i++)
 373     test_data[i] = i & 0xff;
 374
 375   rx_fifo = session->server_rx_fifo;
 376   tx_fifo = session->server_tx_fifo;
 377
 378   before = clib_time_now (&utm->clib_time);
 379
 380   vec_validate (utm->rx_buf, vec_len (test_data) - 1);
 381
 382   for (i = 0; i < NITER; i++)
 383     {
 384       bytes_to_send = vec_len (test_data);
 385       buffer_offset = 0;
 386       while (bytes_to_send > 0)
 387         {
 388           rv = svm_fifo_enqueue_nowait (tx_fifo, bytes_to_send,
 389                                         test_data + buffer_offset);
 390
 391           if (rv > 0)
 392             {
 393               bytes_to_send -= rv;
 394               buffer_offset += rv;
 395               bytes_sent += rv;
 396             }
 397         }
 398
 399       bytes_to_read = svm_fifo_max_dequeue (rx_fifo);
 400       bytes_to_read = vec_len (utm->rx_buf) > bytes_to_read ?
 401         bytes_to_read : vec_len (utm->rx_buf);
 402
 403       buffer_offset = 0;
 404       while (bytes_to_read > 0)
 405         {
 406           rv = svm_fifo_dequeue_nowait (rx_fifo,
 407                                         bytes_to_read,
 408                                         utm->rx_buf + buffer_offset);
 409           if (rv > 0)
 410             {
 411               bytes_to_read -= rv;
 412               buffer_offset += rv;
 413               bytes_received += rv;
 414             }
 415         }
 416     }
 417   while (bytes_received < bytes_sent)
 418     {
 419       rv =
 420         svm_fifo_dequeue_nowait (rx_fifo, vec_len (utm->rx_buf), utm->rx_buf);
 421       if (rv > 0)
 422         {
 423 #if CLIB_DEBUG > 0
 424           int j;
 425           for (j = 0; j < rv; j++)
 426             {
 427               if (utm->rx_buf[j] != ((bytes_received + j) & 0xff))
 428                 {
 429                   clib_warning ("error at byte %lld, 0x%x not 0x%x",
 430                                 bytes_received + j,
 431                                 utm->rx_buf[j],
 432                                 ((bytes_received + j) & 0xff));
 433                 }
 434             }
 435 #endif
 436           bytes_received += (u64) rv;
 437         }
 438     }
 439
 440   after = clib_time_now (&utm->clib_time);
 441   delta = after - before;
 442   bytes_per_second = 0.0;
 443
 444   if (delta > 0.0)
 445     bytes_per_second = (f64) bytes_received / delta;
 446
 447   fformat (stdout,
 448            "Done: %lld recv bytes in %.2f seconds, %.2f bytes/sec...\n\n",
 449            bytes_received, delta, bytes_per_second);
 450   fformat (stdout,
 451            "Done: %lld sent bytes in %.2f seconds, %.2f bytes/sec...\n\n",
 452            bytes_sent, delta, bytes_per_second);
 453   fformat (stdout,
 454            "client -> server -> client round trip: %.2f Gbit/sec \n\n",
 455            (bytes_per_second * 8.0) / 1e9);
 456 }
 457
 458 static void
 459 send_test_chunk (uri_udp_test_main_t * utm, svm_fifo_t * tx_fifo, int mypid,
 460                  u32 bytes)
 461 {
 462   u8 *test_data = utm->connect_test_data;
 463   u64 bytes_sent = 0;
 464   int test_buf_offset = 0;
 465   u32 bytes_to_snd;
 466   u32 queue_max_chunk = 128 << 10, actual_write;
 467   session_fifo_event_t evt;
 468   int rv;
 469
 470   bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes;
 471   if (bytes_to_snd > vec_len (test_data))
 472     bytes_to_snd = vec_len (test_data);
 473
 474   while (bytes_to_snd > 0 && !utm->time_to_stop)
 475     {
 476       actual_write = (bytes_to_snd > queue_max_chunk) ?
 477         queue_max_chunk : bytes_to_snd;
 478       rv = svm_fifo_enqueue_nowait (tx_fifo, actual_write,
 479                                     test_data + test_buf_offset);
 480
 481       if (rv > 0)
 482         {
 483           bytes_to_snd -= rv;
 484           test_buf_offset += rv;
 485           bytes_sent += rv;
 486
 487           if (svm_fifo_set_event (tx_fifo))
 488             {
 489               /* Fabricate TX event, send to vpp */
 490               evt.fifo = tx_fifo;
 491               evt.event_type = FIFO_EVENT_APP_TX;
 492
 493               svm_queue_add (utm->vpp_event_queue,
 494                              (u8 *) & evt, 0 /* do wait for mutex */ );
 495             }
 496         }
 497     }
 498 }
 499
 500 static void
 501 recv_test_chunk (uri_udp_test_main_t * utm, session_t * session)
 502 {
 503   svm_fifo_t *rx_fifo;
 504   int buffer_offset, bytes_to_read = 0, rv;
 505
 506   rx_fifo = session->server_rx_fifo;
 507   bytes_to_read = svm_fifo_max_dequeue (rx_fifo);
 508   bytes_to_read =
 509     vec_len (utm->rx_buf) > bytes_to_read ?
 510     bytes_to_read : vec_len (utm->rx_buf);
 511
 512   buffer_offset = 0;
 513   while (bytes_to_read > 0)
 514     {
 515       rv = svm_fifo_dequeue_nowait (rx_fifo, bytes_to_read,
 516                                     utm->rx_buf + buffer_offset);
 517       if (rv > 0)
 518         {
 519           bytes_to_read -= rv;
 520           buffer_offset += rv;
 521         }
 522     }
 523 }
 524
 525 void
 526 client_send_data (uri_udp_test_main_t * utm)
 527 {
 528   u8 *test_data;
 529   int mypid = getpid ();
 530   session_t *session;
 531   svm_fifo_t *tx_fifo;
 532   u32 n_iterations;
 533   int i;
 534
 535   vec_validate (utm->connect_test_data, 64 * 1024 - 1);
 536   for (i = 0; i < vec_len (utm->connect_test_data); i++)
 537     utm->connect_test_data[i] = i & 0xff;
 538
 539   test_data = utm->connect_test_data;
 540   session = pool_elt_at_index (utm->sessions, utm->connected_session);
 541   tx_fifo = session->server_tx_fifo;
 542
 543   ASSERT (vec_len (test_data) > 0);
 544
 545   vec_validate (utm->rx_buf, vec_len (test_data) - 1);
 546   n_iterations = NITER;
 547
 548   for (i = 0; i < n_iterations; i++)
 549     {
 550       send_test_chunk (utm, tx_fifo, mypid, 0);
 551       recv_test_chunk (utm, session);
 552       if (utm->time_to_stop)
 553         break;
 554     }
 555
 556   f64 timeout = clib_time_now (&utm->clib_time) + 5;
 557   while (clib_time_now (&utm->clib_time) < timeout)
 558     {
 559       recv_test_chunk (utm, session);
 560     }
 561
 562 }
 563
 564 static void
 565 client_test (uri_udp_test_main_t * utm)
 566 {
 567   session_t *session;
 568
 569   application_send_attach (utm);
 570   udp_client_connect (utm);
 571
 572   if (wait_for_state_change (utm, STATE_READY))
 573     {
 574       clib_warning ("timeout waiting for STATE_READY");
 575       return;
 576     }
 577
 578   if (utm->cut_through_session_index != ~0)
 579     {
 580       session = pool_elt_at_index (utm->sessions,
 581                                    utm->cut_through_session_index);
 582       client_send_cut_through (utm, session);
 583     }
 584   else
 585     {
 586       session = pool_elt_at_index (utm->sessions, utm->connected_session);
 587       client_send_data (utm);
 588     }
 589
 590   application_detach (utm);
 591 }
 592
 593 static void
 594 vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
 595 {
 596   uri_udp_test_main_t *utm = &uri_udp_test_main;
 597
 598   if (mp->retval)
 599     {
 600       clib_warning ("bind failed: %d", mp->retval);
 601       utm->state = STATE_FAILED;
 602       return;
 603     }
 604
 605   utm->state = STATE_BOUND;
 606 }
 607
 608 static void
 609 vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
 610 {
 611   uri_udp_test_main_t *utm = &uri_udp_test_main;
 612   svm_fifo_segment_create_args_t _a, *a = &_a;
 613   svm_fifo_segment_private_t *seg;
 614   u8 *seg_name;
 615   int rv;
 616
 617   memset (a, 0, sizeof (*a));
 618   a->segment_name = (char *) mp->segment_name;
 619   a->segment_size = mp->segment_size;
 620   /* Attach to the segment vpp created */
 621   rv = svm_fifo_segment_attach (a);
 622   if (rv)
 623     {
 624       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 625                     mp->segment_name);
 626       return;
 627     }
 628   seg = svm_fifo_segment_get_segment (a->new_segment_indices[0]);
 629   clib_warning ("Mapped new segment '%s' size %d", seg->ssvm.name,
 630                 seg->ssvm.ssvm_size);
 631   seg_name = format (0, "%s", (char *) mp->segment_name);
 632   hash_set_mem (utm->segments_table, seg_name, a->new_segment_indices[0]);
 633   vec_free (seg_name);
 634 }
 635
 636 static void
 637 vl_api_unmap_segment_t_handler (vl_api_unmap_segment_t * mp)
 638 {
 639   uri_udp_test_main_t *utm = &uri_udp_test_main;
 640   svm_fifo_segment_private_t *seg;
 641   u64 *seg_indexp;
 642   u8 *seg_name;
 643
 644
 645   seg_name = format (0, "%s", mp->segment_name);
 646   seg_indexp = hash_get_mem (utm->segments_table, seg_name);
 647   if (!seg_indexp)
 648     {
 649       clib_warning ("segment not mapped: %s", seg_name);
 650       return;
 651     }
 652   hash_unset_mem (utm->segments_table, seg_name);
 653   seg = svm_fifo_segment_get_segment ((u32) seg_indexp[0]);
 654   svm_fifo_segment_delete (seg);
 655   clib_warning ("Unmapped segment '%s'", seg_name);
 656   vec_free (seg_name);
 657 }
 658
 659 /**
 660  * Acting as server for redirected connect requests
 661  */
 662 static void
 663 vl_api_connect_uri_t_handler (vl_api_connect_uri_t * mp)
 664 {
 665   u32 segment_index;
 666   uri_udp_test_main_t *utm = &uri_udp_test_main;
 667   svm_fifo_segment_main_t *sm = &svm_fifo_segment_main;
 668   svm_fifo_segment_create_args_t _a, *a = &_a;
 669   svm_fifo_segment_private_t *seg;
 670   svm_queue_t *client_q;
 671   vl_api_connect_session_reply_t *rmp;
 672   session_t *session = 0;
 673   int rv = 0;
 674
 675   /* Create the segment */
 676   a->segment_name = (char *) format (0, "%d:segment%d%c", utm->my_pid,
 677                                      utm->unique_segment_index++, 0);
 678   a->segment_size = utm->configured_segment_size;
 679
 680   rv = svm_fifo_segment_create (a);
 681   if (rv)
 682     {
 683       clib_warning ("sm_fifo_segment_create ('%s') failed", a->segment_name);
 684       rv = VNET_API_ERROR_URI_FIFO_CREATE_FAILED;
 685       goto send_reply;
 686     }
 687
 688   vec_add2 (utm->seg, seg, 1);
 689
 690   segment_index = vec_len (sm->segments) - 1;
 691   memcpy (seg, sm->segments + segment_index, sizeof (utm->seg[0]));
 692
 693   pool_get (utm->sessions, session);
 694
 695   session->server_rx_fifo = svm_fifo_segment_alloc_fifo
 696     (utm->seg, 128 * 1024, FIFO_SEGMENT_RX_FREELIST);
 697   ASSERT (session->server_rx_fifo);
 698
 699   session->server_tx_fifo = svm_fifo_segment_alloc_fifo
 700     (utm->seg, 128 * 1024, FIFO_SEGMENT_TX_FREELIST);
 701   ASSERT (session->server_tx_fifo);
 702
 703   session->server_rx_fifo->master_session_index = session - utm->sessions;
 704   session->server_tx_fifo->master_session_index = session - utm->sessions;
 705   utm->cut_through_session_index = session - utm->sessions;
 706
 707   rv = pthread_create (&utm->cut_through_thread_handle,
 708                        NULL /*attr */ , cut_through_thread_fn, 0);
 709   if (rv)
 710     {
 711       clib_warning ("pthread_create returned %d", rv);
 712       rv = VNET_API_ERROR_SYSCALL_ERROR_1;
 713     }
 714
 715 send_reply:
 716   rmp = vl_msg_api_alloc (sizeof (*rmp));
 717   memset (rmp, 0, sizeof (*rmp));
 718
 719   rmp->_vl_msg_id = ntohs (VL_API_CONNECT_SESSION_REPLY);
 720   rmp->context = mp->context;
 721   rmp->retval = ntohl (rv);
 722   rmp->segment_name_length = vec_len (a->segment_name);
 723   if (session)
 724     {
 725       rmp->server_rx_fifo = pointer_to_uword (session->server_rx_fifo);
 726       rmp->server_tx_fifo = pointer_to_uword (session->server_tx_fifo);
 727     }
 728
 729   memcpy (rmp->segment_name, a->segment_name, vec_len (a->segment_name));
 730
 731   vec_free (a->segment_name);
 732
 733   client_q = uword_to_pointer (mp->client_queue_address, svm_queue_t *);
 734   vl_msg_api_send_shmem (client_q, (u8 *) & rmp);
 735 }
 736
 737 static void
 738 vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp)
 739 {
 740   uri_udp_test_main_t *utm = &uri_udp_test_main;
 741
 742   if (mp->retval != 0)
 743     clib_warning ("returned %d", ntohl (mp->retval));
 744
 745   utm->state = STATE_START;
 746 }
 747
 748 static void
 749 vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
 750 {
 751   uri_udp_test_main_t *utm = &uri_udp_test_main;
 752   vl_api_accept_session_reply_t *rmp;
 753   svm_fifo_t *rx_fifo, *tx_fifo;
 754   session_t *session;
 755   static f64 start_time;
 756   u32 session_index;
 757   int rv = 0;
 758
 759   if (start_time == 0.0)
 760     start_time = clib_time_now (&utm->clib_time);
 761
 762   utm->vpp_event_queue =
 763     uword_to_pointer (mp->vpp_event_queue_address, svm_queue_t *);
 764
 765   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 766   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 767
 768   pool_get (utm->sessions, session);
 769   memset (session, 0, sizeof (*session));
 770   session_index = session - utm->sessions;
 771
 772   /* Cut-through case */
 773   if (mp->server_event_queue_address)
 774     {
 775       clib_warning ("cut-through session");
 776       utm->our_event_queue = uword_to_pointer (mp->server_event_queue_address,
 777                                                svm_queue_t *);
 778       rx_fifo->master_session_index = session_index;
 779       tx_fifo->master_session_index = session_index;
 780       utm->cut_through_session_index = session_index;
 781       session->server_rx_fifo = rx_fifo;
 782       session->server_tx_fifo = tx_fifo;
 783
 784       rv = pthread_create (&utm->cut_through_thread_handle,
 785                            NULL /*attr */ , cut_through_thread_fn, 0);
 786       if (rv)
 787         {
 788           clib_warning ("pthread_create returned %d", rv);
 789           rv = VNET_API_ERROR_SYSCALL_ERROR_1;
 790         }
 791     }
 792   else
 793     {
 794       rx_fifo->client_session_index = session_index;
 795       tx_fifo->client_session_index = session_index;
 796       session->server_rx_fifo = rx_fifo;
 797       session->server_tx_fifo = tx_fifo;
 798     }
 799
 800   hash_set (utm->session_index_by_vpp_handles, mp->handle, session_index);
 801   if (pool_elts (utm->sessions) && (pool_elts (utm->sessions) % 20000) == 0)
 802     {
 803       f64 now = clib_time_now (&utm->clib_time);
 804       fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
 805                pool_elts (utm->sessions), now - start_time,
 806                (f64) pool_elts (utm->sessions) / (now - start_time));
 807     }
 808
 809   rmp = vl_msg_api_alloc (sizeof (*rmp));
 810   memset (rmp, 0, sizeof (*rmp));
 811   rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY);
 812   rmp->handle = mp->handle;
 813   rmp->context = mp->context;
 814   rmp->retval = rv;
 815   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
 816
 817   CLIB_MEMORY_BARRIER ();
 818   utm->state = STATE_READY;
 819 }
 820
 821 static void
 822 vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp)
 823 {
 824   uri_udp_test_main_t *utm = &uri_udp_test_main;
 825   session_t *session;
 826   vl_api_disconnect_session_reply_t *rmp;
 827   uword *p;
 828   int rv = 0;
 829
 830   p = hash_get (utm->session_index_by_vpp_handles, mp->handle);
 831
 832   if (p)
 833     {
 834       session = pool_elt_at_index (utm->sessions, p[0]);
 835       hash_unset (utm->session_index_by_vpp_handles, mp->handle);
 836       pool_put (utm->sessions, session);
 837     }
 838   else
 839     {
 840       clib_warning ("couldn't find session key %llx", mp->handle);
 841       rv = -11;
 842     }
 843
 844   rmp = vl_msg_api_alloc (sizeof (*rmp));
 845   memset (rmp, 0, sizeof (*rmp));
 846   rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY);
 847   rmp->retval = rv;
 848   rmp->handle = mp->handle;
 849   rmp->context = mp->context;
 850   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
 851 }
 852
 853 static void
 854 vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
 855 {
 856   uri_udp_test_main_t *utm = &uri_udp_test_main;
 857   session_t *session;
 858
 859   ASSERT (utm->i_am_master == 0);
 860
 861   if (mp->retval)
 862     {
 863       clib_warning ("failed connect");
 864       return;
 865     }
 866
 867   pool_get (utm->sessions, session);
 868   session->server_rx_fifo = uword_to_pointer (mp->server_rx_fifo,
 869                                               svm_fifo_t *);
 870   ASSERT (session->server_rx_fifo);
 871   session->server_tx_fifo = uword_to_pointer (mp->server_tx_fifo,
 872                                               svm_fifo_t *);
 873   ASSERT (session->server_tx_fifo);
 874
 875   /* Cut-through case */
 876   if (mp->client_event_queue_address)
 877     {
 878       clib_warning ("cut-through session");
 879       utm->cut_through_session_index = session - utm->sessions;
 880       utm->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address,
 881                                                svm_queue_t *);
 882       utm->our_event_queue = uword_to_pointer (mp->client_event_queue_address,
 883                                                svm_queue_t *);
 884     }
 885   else
 886     {
 887       utm->connected_session = session - utm->sessions;
 888       utm->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address,
 889                                                svm_queue_t *);
 890     }
 891   utm->state = STATE_READY;
 892 }
 893
 894 #define foreach_tcp_echo_msg                            \
 895 _(BIND_URI_REPLY, bind_uri_reply)                       \
 896 _(CONNECT_URI, connect_uri)                             \
 897 _(CONNECT_SESSION_REPLY, connect_session_reply)         \
 898 _(UNBIND_URI_REPLY, unbind_uri_reply)                   \
 899 _(ACCEPT_SESSION, accept_session)                       \
 900 _(DISCONNECT_SESSION, disconnect_session)               \
 901 _(MAP_ANOTHER_SEGMENT, map_another_segment)             \
 902 _(UNMAP_SEGMENT, unmap_segment)                         \
 903 _(APPLICATION_ATTACH_REPLY, application_attach_reply)   \
 904 _(APPLICATION_DETACH_REPLY, application_detach_reply)   \
 905
 906 void
 907 tcp_echo_api_hookup (uri_udp_test_main_t * utm)
 908 {
 909 #define _(N,n)                                                  \
 910     vl_msg_api_set_handlers(VL_API_##N, #n,                     \
 911                            vl_api_##n##_t_handler,              \
 912                            vl_noop_handler,                     \
 913                            vl_api_##n##_t_endian,               \
 914                            vl_api_##n##_t_print,                \
 915                            sizeof(vl_api_##n##_t), 1);
 916   foreach_tcp_echo_msg;
 917 #undef _
 918
 919 }
 920
 921 int
 922 connect_to_vpp (char *name)
 923 {
 924   uri_udp_test_main_t *utm = &uri_udp_test_main;
 925   api_main_t *am = &api_main;
 926
 927   if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0)
 928     return -1;
 929
 930   utm->vl_input_queue = am->shmem_hdr->vl_input_queue;
 931   utm->my_client_index = am->my_client_index;
 932
 933   return 0;
 934 }
 935
 936 void
 937 vlib_cli_output (struct vlib_main_t *vm, char *fmt, ...)
 938 {
 939   clib_warning ("BUG");
 940 }
 941
 942 static void
 943 init_error_string_table (uri_udp_test_main_t * utm)
 944 {
 945   utm->error_string_by_error_number = hash_create (0, sizeof (uword));
 946
 947 #define _(n,v,s) hash_set (utm->error_string_by_error_number, -v, s);
 948   foreach_vnet_api_error;
 949 #undef _
 950
 951   hash_set (utm->error_string_by_error_number, 99, "Misc");
 952 }
 953
 954 void
 955 server_handle_fifo_event_rx (uri_udp_test_main_t * utm,
 956                              session_fifo_event_t * e)
 957 {
 958   svm_fifo_t *rx_fifo, *tx_fifo;
 959   int nbytes;
 960   session_fifo_event_t evt;
 961   svm_queue_t *q;
 962   int rv;
 963
 964   rx_fifo = e->fifo;
 965   tx_fifo = utm->sessions[rx_fifo->client_session_index].server_tx_fifo;
 966   svm_fifo_unset_event (rx_fifo);
 967
 968   do
 969     {
 970       nbytes = svm_fifo_dequeue_nowait (rx_fifo, vec_len (utm->rx_buf),
 971                                         utm->rx_buf);
 972     }
 973   while (nbytes <= 0);
 974   do
 975     {
 976       rv = svm_fifo_enqueue_nowait (tx_fifo, nbytes, utm->rx_buf);
 977     }
 978   while (rv == -2);
 979
 980   if (svm_fifo_set_event (tx_fifo))
 981     {
 982       /* Fabricate TX event, send to vpp */
 983       evt.fifo = tx_fifo;
 984       evt.event_type = FIFO_EVENT_APP_TX;
 985       q = utm->vpp_event_queue;
 986       svm_queue_add (q, (u8 *) & evt, 0 /* do wait for mutex */ );
 987     }
 988 }
 989
 990 void
 991 server_handle_event_queue (uri_udp_test_main_t * utm)
 992 {
 993   session_fifo_event_t _e, *e = &_e;
 994
 995   while (utm->state != STATE_READY)
 996     sleep (5);
 997
 998   while (1)
 999     {
1000       svm_queue_sub (utm->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
1001       switch (e->event_type)
1002         {
1003         case FIFO_EVENT_APP_RX:
1004           server_handle_fifo_event_rx (utm, e);
1005           break;
1006
1007         case FIFO_EVENT_DISCONNECT:
1008           return;
1009
1010         default:
1011           clib_warning ("unknown event type %d", e->event_type);
1012           break;
1013         }
1014       if (PREDICT_FALSE (utm->time_to_stop == 1))
1015         return;
1016       if (PREDICT_FALSE (utm->time_to_print_stats == 1))
1017         {
1018           utm->time_to_print_stats = 0;
1019           fformat (stdout, "%d connections\n", pool_elts (utm->sessions));
1020         }
1021     }
1022 }
1023
1024 static void
1025 server_unbind (uri_udp_test_main_t * utm)
1026 {
1027   vl_api_unbind_uri_t *ump;
1028
1029   ump = vl_msg_api_alloc (sizeof (*ump));
1030   memset (ump, 0, sizeof (*ump));
1031
1032   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
1033   ump->client_index = utm->my_client_index;
1034   memcpy (ump->uri, utm->uri, vec_len (utm->uri));
1035   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & ump);
1036 }
1037
1038 static void
1039 server_bind (uri_udp_test_main_t * utm)
1040 {
1041   vl_api_bind_uri_t *bmp;
1042
1043   bmp = vl_msg_api_alloc (sizeof (*bmp));
1044   memset (bmp, 0, sizeof (*bmp));
1045
1046   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
1047   bmp->client_index = utm->my_client_index;
1048   bmp->context = ntohl (0xfeedface);
1049   memcpy (bmp->uri, utm->uri, vec_len (utm->uri));
1050   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
1051 }
1052
1053 void
1054 udp_server_test (uri_udp_test_main_t * utm)
1055 {
1056
1057   application_send_attach (utm);
1058
1059   /* Bind to uri */
1060   server_bind (utm);
1061
1062   if (wait_for_state_change (utm, STATE_BOUND))
1063     {
1064       clib_warning ("timeout waiting for STATE_BOUND");
1065       return;
1066     }
1067
1068   server_handle_event_queue (utm);
1069
1070   /* Cleanup */
1071   server_unbind (utm);
1072
1073   if (wait_for_state_change (utm, STATE_START))
1074     {
1075       clib_warning ("timeout waiting for STATE_START");
1076       return;
1077     }
1078
1079   application_detach (utm);
1080
1081   fformat (stdout, "Test complete...\n");
1082 }
1083
1084 int
1085 main (int argc, char **argv)
1086 {
1087   uri_udp_test_main_t *utm = &uri_udp_test_main;
1088   u8 *bind_name = (u8 *) "udp://0.0.0.0/1234";
1089   unformat_input_t _argv, *a = &_argv;
1090   int i_am_master = 1;
1091   session_t *session;
1092   u8 *chroot_prefix;
1093   char *app_name;
1094   mheap_t *h;
1095   u8 *heap;
1096   u32 tmp;
1097   int i;
1098
1099   clib_mem_init (0, 256 << 20);
1100
1101   heap = clib_mem_get_per_cpu_heap ();
1102   h = mheap_header (heap);
1103
1104   /* make the main heap thread-safe */
1105   h->flags |= MHEAP_FLAG_THREAD_SAFE;
1106
1107   vec_validate (utm->rx_buf, 8192);
1108
1109   utm->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
1110   utm->my_pid = getpid ();
1111   utm->configured_segment_size = 1 << 20;
1112   utm->segments_table = hash_create_vec (0, sizeof (u8), sizeof (u64));
1113
1114   clib_time_init (&utm->clib_time);
1115   init_error_string_table (utm);
1116   svm_fifo_segment_main_init (0x200000000ULL, 20);
1117   unformat_init_command_line (a, argv);
1118
1119   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
1120     {
1121       if (unformat (a, "chroot prefix %s", &chroot_prefix))
1122         {
1123           vl_set_memory_root_path ((char *) chroot_prefix);
1124         }
1125       else if (unformat (a, "uri %s", &bind_name))
1126         ;
1127       else if (unformat (a, "segment-size %dM", &tmp))
1128         utm->configured_segment_size = tmp << 20;
1129       else if (unformat (a, "segment-size %dG", &tmp))
1130         utm->configured_segment_size = tmp << 30;
1131       else if (unformat (a, "master"))
1132         i_am_master = 1;
1133       else if (unformat (a, "slave"))
1134         i_am_master = 0;
1135       else
1136         {
1137           fformat (stderr, "%s: usage [master|slave]\n");
1138           exit (1);
1139         }
1140     }
1141
1142   utm->cut_through_session_index = ~0;
1143   utm->uri = format (0, "%s%c", bind_name, 0);
1144   utm->i_am_master = i_am_master;
1145   utm->segment_main = &svm_fifo_segment_main;
1146   utm->connect_uri = format (0, "udp://6.0.1.2/1234%c", 0);
1147
1148   setup_signal_handlers ();
1149   tcp_echo_api_hookup (utm);
1150
1151   app_name = i_am_master ? "udp_echo_master" : "udp_echo_slave";
1152   if (connect_to_vpp (app_name) < 0)
1153     {
1154       svm_region_exit ();
1155       fformat (stderr, "Couldn't connect to vpe, exiting...\n");
1156       exit (1);
1157     }
1158
1159   if (i_am_master == 0)
1160     {
1161       client_test (utm);
1162       exit (0);
1163     }
1164
1165   /* $$$$ hack preallocation */
1166   for (i = 0; i < 200000; i++)
1167     {
1168       pool_get (utm->sessions, session);
1169       memset (session, 0, sizeof (*session));
1170     }
1171   for (i = 0; i < 200000; i++)
1172     pool_put_index (utm->sessions, i);
1173
1174   udp_server_test (utm);
1175
1176   vl_client_disconnect_from_vlib ();
1177   exit (0);
1178 }
1179
1180 #undef vl_api_version
1181 #define vl_api_version(n,v) static u32 vpe_api_version = v;
1182 #include <vpp/api/vpe.api.h>
1183 #undef vl_api_version
1184
1185 void
1186 vl_client_add_api_signatures (vl_api_memclnt_create_t * mp)
1187 {
1188   /*
1189    * Send the main API signature in slot 0. This bit of code must
1190    * match the checks in ../vpe/api/api.c: vl_msg_api_version_check().
1191    */
1192   mp->api_versions[0] = clib_host_to_net_u32 (vpe_api_version);
1193 }
1194
1195 u32
1196 vl (void *p)
1197 {
1198   return vec_len (p);
1199 }
1200
1201 /*
1202  * fd.io coding-style-patch-verification: ON
1203  *
1204  * Local Variables:
1205  * eval: (c-set-style "gnu")
1206  * End:
1207  */