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   rmp->context = mp->context;
 483   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
 484
 485   if (session)
 486     session_print_stats (em, session);
 487 }
 488
 489 static void
 490 vl_api_reset_session_t_handler (vl_api_reset_session_t * mp)
 491 {
 492   echo_main_t *em = &echo_main;
 493   vl_api_reset_session_reply_t *rmp;
 494   uword *p;
 495   int rv = 0;
 496
 497   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
 498
 499   if (p)
 500     {
 501       clib_warning ("got reset");
 502       /* Cleanup later */
 503       em->time_to_stop = 1;
 504     }
 505   else
 506     {
 507       clib_warning ("couldn't find session key %llx", mp->handle);
 508       rv = -11;
 509     }
 510
 511   rmp = vl_msg_api_alloc (sizeof (*rmp));
 512   memset (rmp, 0, sizeof (*rmp));
 513   rmp->_vl_msg_id = ntohs (VL_API_RESET_SESSION_REPLY);
 514   rmp->retval = rv;
 515   rmp->handle = mp->handle;
 516   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
 517 }
 518
 519 void
 520 client_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
 521 {
 522   svm_fifo_t *rx_fifo;
 523   int n_read, bytes, i;
 524
 525   rx_fifo = e->fifo;
 526
 527   bytes = svm_fifo_max_dequeue (rx_fifo);
 528   /* Allow enqueuing of new event */
 529   svm_fifo_unset_event (rx_fifo);
 530
 531   /* Read the bytes */
 532   do
 533     {
 534       n_read = svm_fifo_dequeue_nowait (rx_fifo,
 535                                         clib_min (vec_len (em->rx_buf),
 536                                                   bytes), em->rx_buf);
 537       if (n_read > 0)
 538         {
 539           bytes -= n_read;
 540           if (em->test_return_packets)
 541             {
 542               for (i = 0; i < n_read; i++)
 543                 {
 544                   if (em->rx_buf[i]
 545                       != ((em->client_bytes_received + i) & 0xff))
 546                     {
 547                       clib_warning ("error at byte %lld, 0x%x not 0x%x",
 548                                     em->client_bytes_received + i,
 549                                     em->rx_buf[i],
 550                                     ((em->client_bytes_received + i) & 0xff));
 551                     }
 552                 }
 553             }
 554           em->client_bytes_received += n_read;
 555         }
 556       else
 557         {
 558           if (n_read == -2)
 559             {
 560 //            clib_warning ("weird!");
 561               break;
 562             }
 563         }
 564
 565     }
 566   while (bytes > 0);
 567 }
 568
 569 void
 570 client_handle_event_queue (echo_main_t * em)
 571 {
 572   session_fifo_event_t _e, *e = &_e;;
 573
 574   svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
 575   switch (e->event_type)
 576     {
 577     case FIFO_EVENT_APP_RX:
 578       client_handle_fifo_event_rx (em, e);
 579       break;
 580
 581     case FIFO_EVENT_DISCONNECT:
 582       return;
 583
 584     default:
 585       clib_warning ("unknown event type %d", e->event_type);
 586       break;
 587     }
 588 }
 589
 590 static void *
 591 client_rx_thread_fn (void *arg)
 592 {
 593   session_fifo_event_t _e, *e = &_e;
 594   echo_main_t *em = &echo_main;
 595
 596   em->client_bytes_received = 0;
 597   while (1)
 598     {
 599       svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
 600       switch (e->event_type)
 601         {
 602         case FIFO_EVENT_APP_RX:
 603           client_handle_fifo_event_rx (em, e);
 604           break;
 605
 606         case FIFO_EVENT_DISCONNECT:
 607           return 0;
 608         default:
 609           clib_warning ("unknown event type %d", e->event_type);
 610           break;
 611         }
 612
 613       if (PREDICT_FALSE (em->time_to_stop == 1))
 614         break;
 615     }
 616   pthread_exit (0);
 617 }
 618
 619
 620 static void
 621 vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
 622 {
 623   echo_main_t *em = &echo_main;
 624   session_t *session;
 625   u32 session_index;
 626   svm_fifo_t *rx_fifo, *tx_fifo;
 627   int rv;
 628
 629   if (mp->retval)
 630     {
 631       clib_warning ("connection failed with code: %U", format_api_error,
 632                     clib_net_to_host_u32 (mp->retval));
 633       em->state = STATE_FAILED;
 634       return;
 635     }
 636   else
 637     {
 638       clib_warning ("connected with local ip %U port %d", format_ip46_address,
 639                     mp->lcl_ip, mp->is_ip4,
 640                     clib_net_to_host_u16 (mp->lcl_port));
 641     }
 642
 643   em->vpp_event_queue =
 644     uword_to_pointer (mp->vpp_event_queue_address, svm_queue_t *);
 645
 646   /*
 647    * Setup session
 648    */
 649
 650   pool_get (em->sessions, session);
 651   session_index = session - em->sessions;
 652
 653   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 654   rx_fifo->client_session_index = session_index;
 655   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 656   tx_fifo->client_session_index = session_index;
 657
 658   session->server_rx_fifo = rx_fifo;
 659   session->server_tx_fifo = tx_fifo;
 660   session->vpp_session_handle = mp->handle;
 661   session->start = clib_time_now (&em->clib_time);
 662
 663   /* Save handle */
 664   em->connected_session_index = session_index;
 665   em->state = STATE_READY;
 666
 667   /* Add it to lookup table */
 668   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 669
 670   /* Start RX thread */
 671   rv = pthread_create (&em->client_rx_thread_handle,
 672                        NULL /*attr */ , client_rx_thread_fn, 0);
 673   if (rv)
 674     {
 675       clib_warning ("pthread_create returned %d", rv);
 676       rv = VNET_API_ERROR_SYSCALL_ERROR_1;
 677     }
 678 }
 679
 680 static void
 681 send_test_chunk (echo_main_t * em, svm_fifo_t * tx_fifo, int mypid, u32 bytes)
 682 {
 683   u8 *test_data = em->connect_test_data;
 684   u64 bytes_sent = 0;
 685   int test_buf_offset = 0;
 686   u32 bytes_to_snd;
 687   u32 queue_max_chunk = 128 << 10, actual_write;
 688   session_fifo_event_t evt;
 689   int rv;
 690
 691   bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes;
 692   if (bytes_to_snd > vec_len (test_data))
 693     bytes_to_snd = vec_len (test_data);
 694
 695   while (bytes_to_snd > 0 && !em->time_to_stop)
 696     {
 697       actual_write = (bytes_to_snd > queue_max_chunk) ?
 698         queue_max_chunk : bytes_to_snd;
 699       rv = svm_fifo_enqueue_nowait (tx_fifo, actual_write,
 700                                     test_data + test_buf_offset);
 701
 702       if (rv > 0)
 703         {
 704           bytes_to_snd -= rv;
 705           test_buf_offset += rv;
 706           bytes_sent += rv;
 707
 708           if (svm_fifo_set_event (tx_fifo))
 709             {
 710               /* Fabricate TX event, send to vpp */
 711               evt.fifo = tx_fifo;
 712               evt.event_type = FIFO_EVENT_APP_TX;
 713
 714               svm_queue_add (em->vpp_event_queue,
 715                              (u8 *) & evt, 0 /* do wait for mutex */ );
 716             }
 717         }
 718     }
 719 }
 720
 721 void
 722 client_send_data (echo_main_t * em)
 723 {
 724   u8 *test_data = em->connect_test_data;
 725   int mypid = getpid ();
 726   session_t *session;
 727   svm_fifo_t *tx_fifo;
 728   u32 n_iterations, leftover;
 729   int i;
 730
 731   session = pool_elt_at_index (em->sessions, em->connected_session_index);
 732   tx_fifo = session->server_tx_fifo;
 733
 734   ASSERT (vec_len (test_data) > 0);
 735
 736   vec_validate (em->rx_buf, vec_len (test_data) - 1);
 737   n_iterations = em->bytes_to_send / vec_len (test_data);
 738
 739   for (i = 0; i < n_iterations; i++)
 740     {
 741       send_test_chunk (em, tx_fifo, mypid, 0);
 742       if (em->time_to_stop)
 743         break;
 744     }
 745
 746   leftover = em->bytes_to_send % vec_len (test_data);
 747   if (leftover)
 748     send_test_chunk (em, tx_fifo, mypid, leftover);
 749
 750   if (!em->drop_packets)
 751     {
 752       f64 timeout = clib_time_now (&em->clib_time) + 10;
 753
 754       /* Wait for the outstanding packets */
 755       while (em->client_bytes_received <
 756              vec_len (test_data) * n_iterations + leftover)
 757         {
 758           if (clib_time_now (&em->clib_time) > timeout)
 759             {
 760               clib_warning ("timed out waiting for the missing packets");
 761               break;
 762             }
 763         }
 764     }
 765   em->time_to_stop = 1;
 766 }
 767
 768 void
 769 client_send_connect (echo_main_t * em)
 770 {
 771   vl_api_connect_uri_t *cmp;
 772   cmp = vl_msg_api_alloc (sizeof (*cmp));
 773   memset (cmp, 0, sizeof (*cmp));
 774
 775   cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
 776   cmp->client_index = em->my_client_index;
 777   cmp->context = ntohl (0xfeedface);
 778   memcpy (cmp->uri, em->connect_uri, vec_len (em->connect_uri));
 779   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cmp);
 780 }
 781
 782 int
 783 client_connect (echo_main_t * em)
 784 {
 785   client_send_connect (em);
 786   if (wait_for_state_change (em, STATE_READY))
 787     {
 788       clib_warning ("Connect failed");
 789       return -1;
 790     }
 791   return 0;
 792 }
 793
 794 void
 795 client_send_disconnect (echo_main_t * em)
 796 {
 797   session_t *connected_session;
 798   vl_api_disconnect_session_t *dmp;
 799   connected_session = pool_elt_at_index (em->sessions,
 800                                          em->connected_session_index);
 801   dmp = vl_msg_api_alloc (sizeof (*dmp));
 802   memset (dmp, 0, sizeof (*dmp));
 803   dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
 804   dmp->client_index = em->my_client_index;
 805   dmp->handle = connected_session->vpp_session_handle;
 806   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & dmp);
 807 }
 808
 809 int
 810 client_disconnect (echo_main_t * em)
 811 {
 812   client_send_disconnect (em);
 813   clib_warning ("Sent disconnect");
 814   if (wait_for_state_change (em, STATE_START))
 815     {
 816       clib_warning ("Disconnect failed");
 817       return -1;
 818     }
 819   return 0;
 820 }
 821
 822 static void
 823 client_run (echo_main_t * em)
 824 {
 825   int i;
 826
 827   if (application_attach (em))
 828     return;
 829
 830   if (client_connect (em))
 831     {
 832       application_detach (em);
 833       return;
 834     }
 835
 836   /* Init test data */
 837   vec_validate (em->connect_test_data, 128 * 1024 - 1);
 838   for (i = 0; i < vec_len (em->connect_test_data); i++)
 839     em->connect_test_data[i] = i & 0xff;
 840
 841   /* Start send */
 842   client_send_data (em);
 843
 844   /* Disconnect and detach */
 845   client_disconnect (em);
 846   application_detach (em);
 847 }
 848
 849 static void
 850 vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
 851 {
 852   echo_main_t *em = &echo_main;
 853
 854   if (mp->retval)
 855     {
 856       clib_warning ("bind failed: %U", format_api_error,
 857                     clib_net_to_host_u32 (mp->retval));
 858       em->state = STATE_FAILED;
 859       return;
 860     }
 861
 862   em->state = STATE_READY;
 863 }
 864
 865 static void
 866 vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp)
 867 {
 868   echo_main_t *em = &echo_main;
 869
 870   if (mp->retval != 0)
 871     clib_warning ("returned %d", ntohl (mp->retval));
 872
 873   em->state = STATE_START;
 874 }
 875
 876 u8 *
 877 format_ip4_address (u8 * s, va_list * args)
 878 {
 879   u8 *a = va_arg (*args, u8 *);
 880   return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]);
 881 }
 882
 883 u8 *
 884 format_ip6_address (u8 * s, va_list * args)
 885 {
 886   ip6_address_t *a = va_arg (*args, ip6_address_t *);
 887   u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon;
 888
 889   i_max_n_zero = ARRAY_LEN (a->as_u16);
 890   max_n_zeros = 0;
 891   i_first_zero = i_max_n_zero;
 892   n_zeros = 0;
 893   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
 894     {
 895       u32 is_zero = a->as_u16[i] == 0;
 896       if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16))
 897         {
 898           i_first_zero = i;
 899           n_zeros = 0;
 900         }
 901       n_zeros += is_zero;
 902       if ((!is_zero && n_zeros > max_n_zeros)
 903           || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros))
 904         {
 905           i_max_n_zero = i_first_zero;
 906           max_n_zeros = n_zeros;
 907           i_first_zero = ARRAY_LEN (a->as_u16);
 908           n_zeros = 0;
 909         }
 910     }
 911
 912   last_double_colon = 0;
 913   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
 914     {
 915       if (i == i_max_n_zero && max_n_zeros > 1)
 916         {
 917           s = format (s, "::");
 918           i += max_n_zeros - 1;
 919           last_double_colon = 1;
 920         }
 921       else
 922         {
 923           s = format (s, "%s%x",
 924                       (last_double_colon || i == 0) ? "" : ":",
 925                       clib_net_to_host_u16 (a->as_u16[i]));
 926           last_double_colon = 0;
 927         }
 928     }
 929
 930   return s;
 931 }
 932
 933 /* Format an IP46 address. */
 934 u8 *
 935 format_ip46_address (u8 * s, va_list * args)
 936 {
 937   ip46_address_t *ip46 = va_arg (*args, ip46_address_t *);
 938   ip46_type_t type = va_arg (*args, ip46_type_t);
 939   int is_ip4 = 1;
 940
 941   switch (type)
 942     {
 943     case IP46_TYPE_ANY:
 944       is_ip4 = ip46_address_is_ip4 (ip46);
 945       break;
 946     case IP46_TYPE_IP4:
 947       is_ip4 = 1;
 948       break;
 949     case IP46_TYPE_IP6:
 950       is_ip4 = 0;
 951       break;
 952     }
 953
 954   return is_ip4 ?
 955     format (s, "%U", format_ip4_address, &ip46->ip4) :
 956     format (s, "%U", format_ip6_address, &ip46->ip6);
 957 }
 958
 959 static void
 960 vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
 961 {
 962   echo_main_t *em = &echo_main;
 963   vl_api_accept_session_reply_t *rmp;
 964   svm_fifo_t *rx_fifo, *tx_fifo;
 965   session_t *session;
 966   static f64 start_time;
 967   u32 session_index;
 968   u8 *ip_str;
 969
 970   if (start_time == 0.0)
 971     start_time = clib_time_now (&em->clib_time);
 972
 973   ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4);
 974   clib_warning ("Accepted session from: %s:%d", ip_str,
 975                 clib_net_to_host_u16 (mp->port));
 976   em->vpp_event_queue =
 977     uword_to_pointer (mp->vpp_event_queue_address, svm_queue_t *);
 978
 979   /* Allocate local session and set it up */
 980   pool_get (em->sessions, session);
 981   session_index = session - em->sessions;
 982
 983   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 984   rx_fifo->client_session_index = session_index;
 985   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 986   tx_fifo->client_session_index = session_index;
 987
 988   session->server_rx_fifo = rx_fifo;
 989   session->server_tx_fifo = tx_fifo;
 990
 991   /* Add it to lookup table */
 992   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 993
 994   em->state = STATE_READY;
 995
 996   /* Stats printing */
 997   if (pool_elts (em->sessions) && (pool_elts (em->sessions) % 20000) == 0)
 998     {
 999       f64 now = clib_time_now (&em->clib_time);
1000       fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
1001                pool_elts (em->sessions), now - start_time,
1002                (f64) pool_elts (em->sessions) / (now - start_time));
1003     }
1004
1005   /*
1006    * Send accept reply to vpp
1007    */
1008   rmp = vl_msg_api_alloc (sizeof (*rmp));
1009   memset (rmp, 0, sizeof (*rmp));
1010   rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY);
1011   rmp->handle = mp->handle;
1012   rmp->context = mp->context;
1013   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
1014
1015   session->bytes_received = 0;
1016   session->start = clib_time_now (&em->clib_time);
1017 }
1018
1019 void
1020 server_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
1021 {
1022   svm_fifo_t *rx_fifo, *tx_fifo;
1023   int n_read;
1024   session_fifo_event_t evt;
1025   svm_queue_t *q;
1026   session_t *session;
1027   int rv;
1028   u32 max_dequeue, offset, max_transfer, rx_buf_len;
1029
1030   rx_buf_len = vec_len (em->rx_buf);
1031   rx_fifo = e->fifo;
1032   session = &em->sessions[rx_fifo->client_session_index];
1033   tx_fifo = session->server_tx_fifo;
1034
1035   max_dequeue = svm_fifo_max_dequeue (rx_fifo);
1036   /* Allow enqueuing of a new event */
1037   svm_fifo_unset_event (rx_fifo);
1038
1039   if (PREDICT_FALSE (max_dequeue == 0))
1040     {
1041       return;
1042     }
1043
1044   /* Read the max_dequeue */
1045   do
1046     {
1047       max_transfer = clib_min (rx_buf_len, max_dequeue);
1048       n_read = svm_fifo_dequeue_nowait (rx_fifo, max_transfer, em->rx_buf);
1049       if (n_read > 0)
1050         {
1051           max_dequeue -= n_read;
1052           session->bytes_received += n_read;
1053         }
1054
1055       /* Reflect if a non-drop session */
1056       if (!em->drop_packets && n_read > 0)
1057         {
1058           offset = 0;
1059           do
1060             {
1061               rv = svm_fifo_enqueue_nowait (tx_fifo, n_read,
1062                                             &em->rx_buf[offset]);
1063               if (rv > 0)
1064                 {
1065                   n_read -= rv;
1066                   offset += rv;
1067                 }
1068             }
1069           while ((rv <= 0 || n_read > 0) && !em->time_to_stop);
1070
1071           /* If event wasn't set, add one */
1072           if (svm_fifo_set_event (tx_fifo))
1073             {
1074               /* Fabricate TX event, send to vpp */
1075               evt.fifo = tx_fifo;
1076               evt.event_type = FIFO_EVENT_APP_TX;
1077
1078               q = em->vpp_event_queue;
1079               svm_queue_add (q, (u8 *) & evt, 1 /* do wait for mutex */ );
1080             }
1081         }
1082     }
1083   while ((n_read < 0 || max_dequeue > 0) && !em->time_to_stop);
1084 }
1085
1086 void
1087 server_handle_event_queue (echo_main_t * em)
1088 {
1089   session_fifo_event_t _e, *e = &_e;
1090
1091   while (1)
1092     {
1093       svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
1094       switch (e->event_type)
1095         {
1096         case FIFO_EVENT_APP_RX:
1097           server_handle_fifo_event_rx (em, e);
1098           break;
1099
1100         case FIFO_EVENT_DISCONNECT:
1101           return;
1102
1103         default:
1104           clib_warning ("unknown event type %d", e->event_type);
1105           break;
1106         }
1107       if (PREDICT_FALSE (em->time_to_stop == 1))
1108         break;
1109       if (PREDICT_FALSE (em->time_to_print_stats == 1))
1110         {
1111           em->time_to_print_stats = 0;
1112           fformat (stdout, "%d connections\n", pool_elts (em->sessions));
1113         }
1114     }
1115 }
1116
1117 void
1118 server_send_listen (echo_main_t * em)
1119 {
1120   vl_api_bind_uri_t *bmp;
1121   bmp = vl_msg_api_alloc (sizeof (*bmp));
1122   memset (bmp, 0, sizeof (*bmp));
1123
1124   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
1125   bmp->client_index = em->my_client_index;
1126   bmp->context = ntohl (0xfeedface);
1127   memcpy (bmp->uri, em->uri, vec_len (em->uri));
1128   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
1129 }
1130
1131 int
1132 server_listen (echo_main_t * em)
1133 {
1134   server_send_listen (em);
1135   if (wait_for_state_change (em, STATE_READY))
1136     {
1137       clib_warning ("timeout waiting for STATE_READY");
1138       return -1;
1139     }
1140   return 0;
1141 }
1142
1143 void
1144 server_send_unbind (echo_main_t * em)
1145 {
1146   vl_api_unbind_uri_t *ump;
1147
1148   ump = vl_msg_api_alloc (sizeof (*ump));
1149   memset (ump, 0, sizeof (*ump));
1150
1151   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
1152   ump->client_index = em->my_client_index;
1153   memcpy (ump->uri, em->uri, vec_len (em->uri));
1154   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & ump);
1155 }
1156
1157 int
1158 server_unbind (echo_main_t * em)
1159 {
1160   server_send_unbind (em);
1161   if (wait_for_state_change (em, STATE_START))
1162     {
1163       clib_warning ("timeout waiting for STATE_START");
1164       return -1;
1165     }
1166   return 0;
1167 }
1168
1169 void
1170 server_run (echo_main_t * em)
1171 {
1172   session_t *session;
1173   int i;
1174
1175   /* $$$$ hack preallocation */
1176   for (i = 0; i < 200000; i++)
1177     {
1178       pool_get (em->sessions, session);
1179       memset (session, 0, sizeof (*session));
1180     }
1181   for (i = 0; i < 200000; i++)
1182     pool_put_index (em->sessions, i);
1183
1184   if (application_attach (em))
1185     return;
1186
1187   /* Bind to uri */
1188   if (server_listen (em))
1189     return;
1190
1191   /* Enter handle event loop */
1192   server_handle_event_queue (em);
1193
1194   /* Cleanup */
1195   server_send_unbind (em);
1196
1197   application_detach (em);
1198
1199   fformat (stdout, "Test complete...\n");
1200 }
1201
1202 static void
1203 vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t *
1204                                            mp)
1205 {
1206   echo_main_t *em = &echo_main;
1207   session_t *session;
1208
1209   if (mp->retval)
1210     {
1211       clib_warning ("vpp complained about disconnect: %d",
1212                     ntohl (mp->retval));
1213     }
1214
1215   em->state = STATE_START;
1216   session = pool_elt_at_index (em->sessions, em->connected_session_index);
1217   if (session)
1218     session_print_stats (em, session);
1219 }
1220
1221 #define foreach_tcp_echo_msg                                    \
1222 _(BIND_URI_REPLY, bind_uri_reply)                               \
1223 _(UNBIND_URI_REPLY, unbind_uri_reply)                           \
1224 _(ACCEPT_SESSION, accept_session)                               \
1225 _(CONNECT_SESSION_REPLY, connect_session_reply)                 \
1226 _(DISCONNECT_SESSION, disconnect_session)                       \
1227 _(DISCONNECT_SESSION_REPLY, disconnect_session_reply)           \
1228 _(RESET_SESSION, reset_session)                                 \
1229 _(APPLICATION_ATTACH_REPLY, application_attach_reply)           \
1230 _(APPLICATION_DETACH_REPLY, application_detach_reply)           \
1231 _(MAP_ANOTHER_SEGMENT, map_another_segment)                     \
1232
1233 void
1234 tcp_echo_api_hookup (echo_main_t * em)
1235 {
1236 #define _(N,n)                                                  \
1237     vl_msg_api_set_handlers(VL_API_##N, #n,                     \
1238                            vl_api_##n##_t_handler,              \
1239                            vl_noop_handler,                     \
1240                            vl_api_##n##_t_endian,               \
1241                            vl_api_##n##_t_print,                \
1242                            sizeof(vl_api_##n##_t), 1);
1243   foreach_tcp_echo_msg;
1244 #undef _
1245 }
1246
1247 int
1248 main (int argc, char **argv)
1249 {
1250   int i_am_master = 1, drop_packets = 0, test_return_packets = 0;
1251   echo_main_t *em = &echo_main;
1252   unformat_input_t _argv, *a = &_argv;
1253   u8 *chroot_prefix;
1254   u8 *heap, *uri = 0;
1255   u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234";
1256   u8 *connect_uri = (u8 *) "tcp://6.0.1.2/1234";
1257   u64 bytes_to_send = 64 << 10, mbytes;
1258   char *app_name;
1259   u32 tmp;
1260   mheap_t *h;
1261
1262   clib_mem_init (0, 256 << 20);
1263
1264   heap = clib_mem_get_per_cpu_heap ();
1265   h = mheap_header (heap);
1266
1267   /* make the main heap thread-safe */
1268   h->flags |= MHEAP_FLAG_THREAD_SAFE;
1269
1270   vec_validate (em->rx_buf, 128 << 10);
1271
1272   em->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
1273
1274   em->my_pid = getpid ();
1275   em->configured_segment_size = 1 << 20;
1276   em->socket_name = 0;
1277   em->use_sock_api = 1;
1278
1279   clib_time_init (&em->clib_time);
1280   init_error_string_table (em);
1281   svm_fifo_segment_main_init (0x200000000ULL, 20);
1282   unformat_init_command_line (a, argv);
1283
1284   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
1285     {
1286       if (unformat (a, "chroot prefix %s", &chroot_prefix))
1287         {
1288           vl_set_memory_root_path ((char *) chroot_prefix);
1289         }
1290       else if (unformat (a, "uri %s", &uri))
1291         ;
1292       else if (unformat (a, "segment-size %dM", &tmp))
1293         em->configured_segment_size = tmp << 20;
1294       else if (unformat (a, "segment-size %dG", &tmp))
1295         em->configured_segment_size = tmp << 30;
1296       else if (unformat (a, "master"))
1297         i_am_master = 1;
1298       else if (unformat (a, "slave"))
1299         i_am_master = 0;
1300       else if (unformat (a, "drop"))
1301         drop_packets = 1;
1302       else if (unformat (a, "test"))
1303         test_return_packets = 1;
1304       else if (unformat (a, "mbytes %lld", &mbytes))
1305         {
1306           bytes_to_send = mbytes << 20;
1307         }
1308       else if (unformat (a, "gbytes %lld", &mbytes))
1309         {
1310           bytes_to_send = mbytes << 30;
1311         }
1312       else if (unformat (a, "socket-name %s", &em->socket_name))
1313         ;
1314       else if (unformat (a, "use-svm-api"))
1315         em->use_sock_api = 0;
1316       else
1317         {
1318           fformat (stderr, "%s: usage [master|slave]\n");
1319           exit (1);
1320         }
1321     }
1322
1323   if (!em->socket_name)
1324     em->socket_name = format (0, "%s%c", API_SOCKET_FILE, 0);
1325
1326   if (uri)
1327     {
1328       em->uri = format (0, "%s%c", uri, 0);
1329       em->connect_uri = format (0, "%s%c", uri, 0);
1330     }
1331   else
1332     {
1333       em->uri = format (0, "%s%c", bind_uri, 0);
1334       em->connect_uri = format (0, "%s%c", connect_uri, 0);
1335     }
1336
1337   em->i_am_master = i_am_master;
1338   em->segment_main = &svm_fifo_segment_main;
1339   em->drop_packets = drop_packets;
1340   em->test_return_packets = test_return_packets;
1341   em->bytes_to_send = bytes_to_send;
1342   em->time_to_stop = 0;
1343
1344   setup_signal_handlers ();
1345   tcp_echo_api_hookup (em);
1346
1347   app_name = i_am_master ? "tcp_echo_server" : "tcp_echo_client";
1348   if (connect_to_vpp (app_name) < 0)
1349     {
1350       svm_region_exit ();
1351       fformat (stderr, "Couldn't connect to vpe, exiting...\n");
1352       exit (1);
1353     }
1354
1355   if (i_am_master == 0)
1356     client_run (em);
1357   else
1358     server_run (em);
1359
1360   disconnect_from_vpp (em);
1361   exit (0);
1362 }
1363
1364 /*
1365  * fd.io coding-style-patch-verification: ON
1366  *
1367  * Local Variables:
1368  * eval: (c-set-style "gnu")
1369  * End:
1370  */