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