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