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 <vlib/vlib.h>
  19 #include <vnet/vnet.h>
  20 #include <svm/svm_fifo_segment.h>
  21 #include <vlibmemory/api.h>
  22 #include <vpp/api/vpe_msg_enum.h>
  23 #include <vnet/session/application_interface.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 /* Satisfy external references when not linking with -lvlib */
  42 vlib_main_t vlib_global_main;
  43 vlib_main_t **vlib_mains;
  44
  45 typedef struct
  46 {
  47   svm_fifo_t *server_rx_fifo;
  48   svm_fifo_t *server_tx_fifo;
  49
  50   u32 vpp_session_index;
  51   u32 vpp_session_thread;
  52 } session_t;
  53
  54 typedef enum
  55 {
  56   STATE_START,
  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   u32 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 int
 134 wait_for_state_change (uri_tcp_test_main_t * utm, connection_state_t state)
 135 {
 136 #if CLIB_DEBUG > 0
 137 #define TIMEOUT 600.0
 138 #else
 139 #define TIMEOUT 600.0
 140 #endif
 141
 142   f64 timeout = clib_time_now (&utm->clib_time) + TIMEOUT;
 143
 144   while (clib_time_now (&utm->clib_time) < timeout)
 145     {
 146       if (utm->state == state)
 147         return 0;
 148       if (utm->state == STATE_FAILED)
 149         return -1;
 150     }
 151   clib_warning ("timeout waiting for STATE_READY");
 152   return -1;
 153 }
 154
 155 static void
 156 init_error_string_table (uri_tcp_test_main_t * utm)
 157 {
 158   utm->error_string_by_error_number = hash_create (0, sizeof (uword));
 159
 160 #define _(n,v,s) hash_set (utm->error_string_by_error_number, -v, s);
 161   foreach_vnet_api_error;
 162 #undef _
 163
 164   hash_set (utm->error_string_by_error_number, 99, "Misc");
 165 }
 166
 167 static void
 168 stop_signal (int signum)
 169 {
 170   uri_tcp_test_main_t *um = &uri_tcp_test_main;
 171
 172   um->time_to_stop = 1;
 173 }
 174
 175 static void
 176 stats_signal (int signum)
 177 {
 178   uri_tcp_test_main_t *um = &uri_tcp_test_main;
 179
 180   um->time_to_print_stats = 1;
 181 }
 182
 183 static clib_error_t *
 184 setup_signal_handlers (void)
 185 {
 186   signal (SIGINT, stats_signal);
 187   signal (SIGQUIT, stop_signal);
 188   signal (SIGTERM, stop_signal);
 189
 190   return 0;
 191 }
 192
 193 void
 194 vlib_cli_output (struct vlib_main_t *vm, char *fmt, ...)
 195 {
 196   clib_warning ("BUG");
 197 }
 198
 199 int
 200 connect_to_vpp (char *name)
 201 {
 202   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 203   api_main_t *am = &api_main;
 204
 205   if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0)
 206     return -1;
 207
 208   utm->vl_input_queue = am->shmem_hdr->vl_input_queue;
 209   utm->my_client_index = am->my_client_index;
 210
 211   return 0;
 212 }
 213
 214 static void
 215 vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
 216 {
 217   svm_fifo_segment_create_args_t _a, *a = &_a;
 218   int rv;
 219
 220   a->segment_name = (char *) mp->segment_name;
 221   a->segment_size = mp->segment_size;
 222   /* Attach to the segment vpp created */
 223   rv = svm_fifo_segment_attach (a);
 224   if (rv)
 225     {
 226       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 227                     mp->segment_name);
 228       return;
 229     }
 230   clib_warning ("Mapped new segment '%s' size %d", mp->segment_name,
 231                 mp->segment_size);
 232 }
 233
 234 static void
 235 vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp)
 236 {
 237   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 238   session_t *session;
 239   vl_api_disconnect_session_reply_t *rmp;
 240   uword *p;
 241   int rv = 0;
 242   u64 key;
 243
 244   key = (((u64) mp->session_thread_index) << 32) | (u64) mp->session_index;
 245
 246   p = hash_get (utm->session_index_by_vpp_handles, key);
 247
 248   if (p)
 249     {
 250       session = pool_elt_at_index (utm->sessions, p[0]);
 251       hash_unset (utm->session_index_by_vpp_handles, key);
 252       pool_put (utm->sessions, session);
 253     }
 254   else
 255     {
 256       clib_warning ("couldn't find session key %llx", key);
 257       rv = -11;
 258     }
 259
 260   utm->time_to_stop = 1;
 261
 262   rmp = vl_msg_api_alloc (sizeof (*rmp));
 263   memset (rmp, 0, sizeof (*rmp));
 264
 265   rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY);
 266   rmp->retval = rv;
 267   rmp->session_index = mp->session_index;
 268   rmp->session_thread_index = mp->session_thread_index;
 269   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
 270 }
 271
 272 static void
 273 vl_api_reset_session_t_handler (vl_api_reset_session_t * mp)
 274 {
 275   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 276   session_t *session;
 277   vl_api_reset_session_reply_t *rmp;
 278   uword *p;
 279   int rv = 0;
 280   u64 key;
 281
 282   key = (((u64) mp->session_thread_index) << 32) | (u64) mp->session_index;
 283
 284   p = hash_get (utm->session_index_by_vpp_handles, key);
 285
 286   if (p)
 287     {
 288       session = pool_elt_at_index (utm->sessions, p[0]);
 289       hash_unset (utm->session_index_by_vpp_handles, key);
 290       pool_put (utm->sessions, session);
 291       utm->time_to_stop = 1;
 292     }
 293   else
 294     {
 295       clib_warning ("couldn't find session key %llx", key);
 296       rv = -11;
 297     }
 298
 299   rmp = vl_msg_api_alloc (sizeof (*rmp));
 300   memset (rmp, 0, sizeof (*rmp));
 301   rmp->_vl_msg_id = ntohs (VL_API_RESET_SESSION_REPLY);
 302   rmp->retval = rv;
 303   rmp->session_index = mp->session_index;
 304   rmp->session_thread_index = mp->session_thread_index;
 305   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
 306 }
 307
 308 void
 309 client_handle_fifo_event_rx (uri_tcp_test_main_t * utm,
 310                              session_fifo_event_t * e)
 311 {
 312   svm_fifo_t *rx_fifo;
 313   int n_read, bytes, i;
 314
 315   rx_fifo = e->fifo;
 316
 317   bytes = svm_fifo_max_dequeue (rx_fifo);
 318   /* Allow enqueuing of new event */
 319   svm_fifo_unset_event (rx_fifo);
 320
 321   /* Read the bytes */
 322   do
 323     {
 324       n_read = svm_fifo_dequeue_nowait (rx_fifo, 0,
 325                                         clib_min (vec_len (utm->rx_buf),
 326                                                   bytes), utm->rx_buf);
 327       if (n_read > 0)
 328         {
 329           bytes -= n_read;
 330           for (i = 0; i < n_read; i++)
 331             {
 332               if (utm->rx_buf[i] != ((utm->client_bytes_received + i) & 0xff))
 333                 {
 334                   clib_warning ("error at byte %lld, 0x%x not 0x%x",
 335                                 utm->client_bytes_received + i,
 336                                 utm->rx_buf[i],
 337                                 ((utm->client_bytes_received + i) & 0xff));
 338                 }
 339             }
 340           utm->client_bytes_received += n_read;
 341         }
 342       else
 343         {
 344           if (n_read == -2)
 345             {
 346               clib_warning ("weird!");
 347               break;
 348             }
 349         }
 350
 351     }
 352   while (bytes > 0);
 353 }
 354
 355 void
 356 client_handle_event_queue (uri_tcp_test_main_t * utm)
 357 {
 358   session_fifo_event_t _e, *e = &_e;;
 359
 360   unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e,
 361                                 0 /* nowait */ );
 362   switch (e->event_type)
 363     {
 364     case FIFO_EVENT_SERVER_RX:
 365       client_handle_fifo_event_rx (utm, e);
 366       break;
 367
 368     case FIFO_EVENT_SERVER_EXIT:
 369       return;
 370
 371     default:
 372       clib_warning ("unknown event type %d", e->event_type);
 373       break;
 374     }
 375 }
 376
 377 static void *
 378 client_rx_thread_fn (void *arg)
 379 {
 380   session_fifo_event_t _e, *e = &_e;
 381   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 382
 383   utm->client_bytes_received = 0;
 384   while (1)
 385     {
 386       unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e,
 387                                     0 /* nowait */ );
 388       switch (e->event_type)
 389         {
 390         case FIFO_EVENT_SERVER_RX:
 391           client_handle_fifo_event_rx (utm, e);
 392           break;
 393
 394         case FIFO_EVENT_SERVER_EXIT:
 395           return 0;
 396         default:
 397           clib_warning ("unknown event type %d", e->event_type);
 398           break;
 399         }
 400
 401       if (PREDICT_FALSE (utm->time_to_stop == 1))
 402         break;
 403     }
 404   pthread_exit (0);
 405 }
 406
 407
 408 static void
 409 vl_api_connect_uri_reply_t_handler (vl_api_connect_uri_reply_t * mp)
 410 {
 411   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 412   svm_fifo_segment_create_args_t _a, *a = &_a;
 413   session_t *session;
 414   u32 session_index;
 415   svm_fifo_t *rx_fifo, *tx_fifo;
 416   int rv;
 417   u64 key;
 418
 419   if (mp->retval)
 420     {
 421       clib_warning ("connection failed with code: %d", mp->retval);
 422       utm->state = STATE_FAILED;
 423       return;
 424     }
 425
 426   /*
 427    * Attatch to segment
 428    */
 429
 430   if (mp->segment_name_length == 0)
 431     {
 432       clib_warning ("segment_name_length zero");
 433       utm->state = STATE_FAILED;
 434       return;
 435     }
 436
 437   a->segment_name = (char *) mp->segment_name;
 438   a->segment_size = mp->segment_size;
 439
 440   ASSERT (mp->client_event_queue_address);
 441
 442   /* Attach to the segment vpp created */
 443   rv = svm_fifo_segment_attach (a);
 444   if (rv)
 445     {
 446       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 447                     mp->segment_name);
 448       return;
 449     }
 450
 451   /*
 452    * Save the queues
 453    */
 454
 455   utm->our_event_queue = (unix_shared_memory_queue_t *)
 456     mp->client_event_queue_address;
 457
 458   utm->vpp_event_queue = (unix_shared_memory_queue_t *)
 459     mp->vpp_event_queue_address;
 460
 461   /*
 462    * Setup session
 463    */
 464
 465   pool_get (utm->sessions, session);
 466   session_index = session - utm->sessions;
 467
 468   rx_fifo = (svm_fifo_t *) mp->server_rx_fifo;
 469   rx_fifo->client_session_index = session_index;
 470   tx_fifo = (svm_fifo_t *) mp->server_tx_fifo;
 471   tx_fifo->client_session_index = session_index;
 472
 473   session->server_rx_fifo = rx_fifo;
 474   session->server_tx_fifo = tx_fifo;
 475   session->vpp_session_index = mp->session_index;
 476   session->vpp_session_thread = mp->session_thread_index;
 477
 478   /* Save handle */
 479   utm->connected_session_index = session_index;
 480   utm->state = STATE_READY;
 481
 482   /* Add it to lookup table */
 483   key = (((u64) mp->session_thread_index) << 32) | (u64) mp->session_index;
 484   hash_set (utm->session_index_by_vpp_handles, key, session_index);
 485
 486   /* Start RX thread */
 487   rv = pthread_create (&utm->client_rx_thread_handle,
 488                        NULL /*attr */ , client_rx_thread_fn, 0);
 489   if (rv)
 490     {
 491       clib_warning ("pthread_create returned %d", rv);
 492       rv = VNET_API_ERROR_SYSCALL_ERROR_1;
 493     }
 494 }
 495
 496 static void
 497 send_test_chunk (uri_tcp_test_main_t * utm, svm_fifo_t * tx_fifo, int mypid,
 498                  u32 bytes)
 499 {
 500   u8 *test_data = utm->connect_test_data;
 501   u64 bytes_sent = 0;
 502   int test_buf_offset = 0;
 503   u32 bytes_to_snd;
 504   u32 queue_max_chunk = 64 << 10, actual_write;
 505   session_fifo_event_t evt;
 506   static int serial_number = 0;
 507   int rv;
 508
 509   bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes;
 510   if (bytes_to_snd > vec_len (test_data))
 511     bytes_to_snd = vec_len (test_data);
 512
 513   while (bytes_to_snd > 0)
 514     {
 515       actual_write =
 516         bytes_to_snd > queue_max_chunk ? queue_max_chunk : bytes_to_snd;
 517       rv = svm_fifo_enqueue_nowait (tx_fifo, mypid, actual_write,
 518                                     test_data + test_buf_offset);
 519
 520       if (rv > 0)
 521         {
 522           bytes_to_snd -= rv;
 523           test_buf_offset += rv;
 524           bytes_sent += rv;
 525
 526           if (svm_fifo_set_event (tx_fifo))
 527             {
 528               /* Fabricate TX event, send to vpp */
 529               evt.fifo = tx_fifo;
 530               evt.event_type = FIFO_EVENT_SERVER_TX;
 531               evt.event_id = serial_number++;
 532
 533               unix_shared_memory_queue_add (utm->vpp_event_queue,
 534                                             (u8 *) & evt,
 535                                             0 /* do wait for mutex */ );
 536             }
 537         }
 538     }
 539 }
 540
 541 void
 542 client_send_data (uri_tcp_test_main_t * utm)
 543 {
 544   u8 *test_data = utm->connect_test_data;
 545   int mypid = getpid ();
 546   session_t *session;
 547   svm_fifo_t *tx_fifo;
 548   u32 n_iterations, leftover;
 549   int i;
 550
 551   session = pool_elt_at_index (utm->sessions, utm->connected_session_index);
 552   tx_fifo = session->server_tx_fifo;
 553
 554   vec_validate (utm->rx_buf, vec_len (test_data) - 1);
 555   n_iterations = utm->bytes_to_send / vec_len (test_data);
 556
 557   for (i = 0; i < n_iterations; i++)
 558     {
 559       send_test_chunk (utm, tx_fifo, mypid, 0);
 560     }
 561
 562   leftover = utm->bytes_to_send % vec_len (test_data);
 563   if (leftover)
 564     send_test_chunk (utm, tx_fifo, mypid, leftover);
 565
 566   if (utm->test_return_packets)
 567     {
 568       f64 timeout = clib_time_now (&utm->clib_time) + 2;
 569
 570       /* Wait for the outstanding packets */
 571       while (utm->client_bytes_received <
 572              vec_len (test_data) * n_iterations + leftover)
 573         {
 574           if (clib_time_now (&utm->clib_time) > timeout)
 575             {
 576               clib_warning ("timed out waiting for the missing packets");
 577               break;
 578             }
 579         }
 580     }
 581   utm->time_to_stop = 1;
 582 }
 583
 584 void
 585 client_connect (uri_tcp_test_main_t * utm)
 586 {
 587   vl_api_connect_uri_t *cmp;
 588   cmp = vl_msg_api_alloc (sizeof (*cmp));
 589   memset (cmp, 0, sizeof (*cmp));
 590
 591   cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
 592   cmp->client_index = utm->my_client_index;
 593   cmp->context = ntohl (0xfeedface);
 594   memcpy (cmp->uri, utm->connect_uri, vec_len (utm->connect_uri));
 595   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & cmp);
 596 }
 597
 598 void
 599 client_disconnect (uri_tcp_test_main_t * utm)
 600 {
 601   session_t *connected_session;
 602   vl_api_disconnect_session_t *dmp;
 603   connected_session = pool_elt_at_index (utm->sessions,
 604                                          utm->connected_session_index);
 605   dmp = vl_msg_api_alloc (sizeof (*dmp));
 606   memset (dmp, 0, sizeof (*dmp));
 607   dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
 608   dmp->client_index = utm->my_client_index;
 609   dmp->session_index = connected_session->vpp_session_index;
 610   dmp->session_thread_index = connected_session->vpp_session_thread;
 611   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & dmp);
 612 }
 613
 614 static void
 615 client_test (uri_tcp_test_main_t * utm)
 616 {
 617   int i;
 618
 619   client_connect (utm);
 620
 621   if (wait_for_state_change (utm, STATE_READY))
 622     {
 623       return;
 624     }
 625
 626   /* Init test data */
 627   vec_validate (utm->connect_test_data, 64 * 1024 - 1);
 628   for (i = 0; i < vec_len (utm->connect_test_data); i++)
 629     utm->connect_test_data[i] = i & 0xff;
 630
 631   /* Start send */
 632   client_send_data (utm);
 633
 634   /* Disconnect */
 635   client_disconnect (utm);
 636
 637   if (wait_for_state_change (utm, STATE_START))
 638     {
 639       return;
 640     }
 641 }
 642
 643 static void
 644 vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
 645 {
 646   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 647   svm_fifo_segment_create_args_t _a, *a = &_a;
 648   int rv;
 649
 650   if (mp->retval)
 651     {
 652       clib_warning ("bind failed: %d", mp->retval);
 653       utm->state = STATE_FAILED;
 654       return;
 655     }
 656
 657   if (mp->segment_name_length == 0)
 658     {
 659       clib_warning ("segment_name_length zero");
 660       return;
 661     }
 662
 663   a->segment_name = (char *) mp->segment_name;
 664   a->segment_size = mp->segment_size;
 665
 666   ASSERT (mp->server_event_queue_address);
 667
 668   /* Attach to the segment vpp created */
 669   rv = svm_fifo_segment_attach (a);
 670   if (rv)
 671     {
 672       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 673                     mp->segment_name);
 674       return;
 675     }
 676
 677   utm->our_event_queue =
 678     (unix_shared_memory_queue_t *) mp->server_event_queue_address;
 679
 680   utm->state = STATE_READY;
 681 }
 682
 683 static void
 684 vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp)
 685 {
 686   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 687
 688   if (mp->retval != 0)
 689     clib_warning ("returned %d", ntohl (mp->retval));
 690
 691   utm->state = STATE_START;
 692 }
 693
 694 static void
 695 vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
 696 {
 697   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 698   vl_api_accept_session_reply_t *rmp;
 699   svm_fifo_t *rx_fifo, *tx_fifo;
 700   session_t *session;
 701   static f64 start_time;
 702   u64 key;
 703   u32 session_index;
 704
 705   if (start_time == 0.0)
 706     start_time = clib_time_now (&utm->clib_time);
 707
 708   utm->vpp_event_queue = (unix_shared_memory_queue_t *)
 709     mp->vpp_event_queue_address;
 710
 711   /* Allocate local session and set it up */
 712   pool_get (utm->sessions, session);
 713   session_index = session - utm->sessions;
 714
 715   rx_fifo = (svm_fifo_t *) mp->server_rx_fifo;
 716   rx_fifo->client_session_index = session_index;
 717   tx_fifo = (svm_fifo_t *) mp->server_tx_fifo;
 718   tx_fifo->client_session_index = session_index;
 719
 720   session->server_rx_fifo = rx_fifo;
 721   session->server_tx_fifo = tx_fifo;
 722
 723   /* Add it to lookup table */
 724   key = (((u64) mp->session_thread_index) << 32) | (u64) mp->session_index;
 725   hash_set (utm->session_index_by_vpp_handles, key, session_index);
 726
 727   utm->state = STATE_READY;
 728
 729   /* Stats printing */
 730   if (pool_elts (utm->sessions) && (pool_elts (utm->sessions) % 20000) == 0)
 731     {
 732       f64 now = clib_time_now (&utm->clib_time);
 733       fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
 734                pool_elts (utm->sessions), now - start_time,
 735                (f64) pool_elts (utm->sessions) / (now - start_time));
 736     }
 737
 738   /*
 739    * Send accept reply to vpp
 740    */
 741   rmp = vl_msg_api_alloc (sizeof (*rmp));
 742   memset (rmp, 0, sizeof (*rmp));
 743   rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY);
 744   rmp->session_type = mp->session_type;
 745   rmp->session_index = mp->session_index;
 746   rmp->session_thread_index = mp->session_thread_index;
 747   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & rmp);
 748 }
 749
 750 void
 751 server_handle_fifo_event_rx (uri_tcp_test_main_t * utm,
 752                              session_fifo_event_t * e)
 753 {
 754   svm_fifo_t *rx_fifo, *tx_fifo;
 755   int n_read;
 756   session_fifo_event_t evt;
 757   unix_shared_memory_queue_t *q;
 758   int rv, bytes;
 759
 760   rx_fifo = e->fifo;
 761   tx_fifo = utm->sessions[rx_fifo->client_session_index].server_tx_fifo;
 762
 763   bytes = svm_fifo_max_dequeue (rx_fifo);
 764   /* Allow enqueuing of a new event */
 765   svm_fifo_unset_event (rx_fifo);
 766
 767   if (bytes == 0)
 768     return;
 769
 770   /* Read the bytes */
 771   do
 772     {
 773       n_read = svm_fifo_dequeue_nowait (rx_fifo, 0, vec_len (utm->rx_buf),
 774                                         utm->rx_buf);
 775       if (n_read > 0)
 776         bytes -= n_read;
 777
 778       if (utm->drop_packets)
 779         continue;
 780
 781       /* Reflect if a non-drop session */
 782       if (n_read > 0)
 783         {
 784           do
 785             {
 786               rv = svm_fifo_enqueue_nowait (tx_fifo, 0, n_read, utm->rx_buf);
 787             }
 788           while (rv <= 0 && !utm->time_to_stop);
 789
 790           /* If event wasn't set, add one */
 791           if (svm_fifo_set_event (tx_fifo))
 792             {
 793               /* Fabricate TX event, send to vpp */
 794               evt.fifo = tx_fifo;
 795               evt.event_type = FIFO_EVENT_SERVER_TX;
 796               evt.event_id = e->event_id;
 797
 798               q = utm->vpp_event_queue;
 799               unix_shared_memory_queue_add (q, (u8 *) & evt,
 800                                             0 /* do wait for mutex */ );
 801             }
 802         }
 803     }
 804   while ((n_read < 0 || bytes > 0) && !utm->time_to_stop);
 805 }
 806
 807 void
 808 server_handle_event_queue (uri_tcp_test_main_t * utm)
 809 {
 810   session_fifo_event_t _e, *e = &_e;;
 811
 812   while (1)
 813     {
 814       unix_shared_memory_queue_sub (utm->our_event_queue, (u8 *) e,
 815                                     0 /* nowait */ );
 816       switch (e->event_type)
 817         {
 818         case FIFO_EVENT_SERVER_RX:
 819           server_handle_fifo_event_rx (utm, e);
 820           break;
 821
 822         case FIFO_EVENT_SERVER_EXIT:
 823           return;
 824
 825         default:
 826           clib_warning ("unknown event type %d", e->event_type);
 827           break;
 828         }
 829       if (PREDICT_FALSE (utm->time_to_stop == 1))
 830         break;
 831       if (PREDICT_FALSE (utm->time_to_print_stats == 1))
 832         {
 833           utm->time_to_print_stats = 0;
 834           fformat (stdout, "%d connections\n", pool_elts (utm->sessions));
 835         }
 836     }
 837 }
 838
 839 void
 840 server_bind (uri_tcp_test_main_t * utm)
 841 {
 842   vl_api_bind_uri_t *bmp;
 843   u32 fifo_size = 3 << 20;
 844   bmp = vl_msg_api_alloc (sizeof (*bmp));
 845   memset (bmp, 0, sizeof (*bmp));
 846
 847   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
 848   bmp->client_index = utm->my_client_index;
 849   bmp->context = ntohl (0xfeedface);
 850   bmp->initial_segment_size = 256 << 20;        /* size of initial segment */
 851   bmp->options[SESSION_OPTIONS_FLAGS] =
 852     SESSION_OPTIONS_FLAGS_USE_FIFO | SESSION_OPTIONS_FLAGS_ADD_SEGMENT;
 853   bmp->options[SESSION_OPTIONS_RX_FIFO_SIZE] = fifo_size;
 854   bmp->options[SESSION_OPTIONS_TX_FIFO_SIZE] = fifo_size;
 855   bmp->options[SESSION_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
 856   memcpy (bmp->uri, utm->uri, vec_len (utm->uri));
 857   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & bmp);
 858 }
 859
 860 void
 861 server_unbind (uri_tcp_test_main_t * utm)
 862 {
 863   vl_api_unbind_uri_t *ump;
 864
 865   ump = vl_msg_api_alloc (sizeof (*ump));
 866   memset (ump, 0, sizeof (*ump));
 867
 868   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
 869   ump->client_index = utm->my_client_index;
 870   memcpy (ump->uri, utm->uri, vec_len (utm->uri));
 871   vl_msg_api_send_shmem (utm->vl_input_queue, (u8 *) & ump);
 872 }
 873
 874 void
 875 server_test (uri_tcp_test_main_t * utm)
 876 {
 877   /* Bind to uri */
 878   server_bind (utm);
 879
 880   if (wait_for_state_change (utm, STATE_READY))
 881     {
 882       clib_warning ("timeout waiting for STATE_READY");
 883       return;
 884     }
 885
 886   /* Enter handle event loop */
 887   server_handle_event_queue (utm);
 888
 889   /* Cleanup */
 890   server_unbind (utm);
 891
 892   if (wait_for_state_change (utm, STATE_START))
 893     {
 894       clib_warning ("timeout waiting for STATE_START");
 895       return;
 896     }
 897
 898   fformat (stdout, "Test complete...\n");
 899 }
 900
 901 static void
 902 vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t *
 903                                            mp)
 904 {
 905   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 906
 907   clib_warning ("retval %d", ntohl (mp->retval));
 908   utm->state = STATE_START;
 909 }
 910
 911 #define foreach_uri_msg                                 \
 912 _(BIND_URI_REPLY, bind_uri_reply)                       \
 913 _(UNBIND_URI_REPLY, unbind_uri_reply)                   \
 914 _(ACCEPT_SESSION, accept_session)                       \
 915 _(CONNECT_URI_REPLY, connect_uri_reply)                 \
 916 _(DISCONNECT_SESSION, disconnect_session)               \
 917 _(DISCONNECT_SESSION_REPLY, disconnect_session_reply)   \
 918 _(RESET_SESSION, reset_session)                         \
 919 _(MAP_ANOTHER_SEGMENT, map_another_segment)
 920
 921 void
 922 uri_api_hookup (uri_tcp_test_main_t * utm)
 923 {
 924 #define _(N,n)                                                  \
 925     vl_msg_api_set_handlers(VL_API_##N, #n,                     \
 926                            vl_api_##n##_t_handler,              \
 927                            vl_noop_handler,                     \
 928                            vl_api_##n##_t_endian,               \
 929                            vl_api_##n##_t_print,                \
 930                            sizeof(vl_api_##n##_t), 1);
 931   foreach_uri_msg;
 932 #undef _
 933 }
 934
 935 int
 936 main (int argc, char **argv)
 937 {
 938   uri_tcp_test_main_t *utm = &uri_tcp_test_main;
 939   unformat_input_t _argv, *a = &_argv;
 940   u8 *chroot_prefix;
 941   u8 *heap, *uri = 0;
 942   u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234";
 943   u8 *connect_uri = (u8 *) "tcp://6.0.1.2/1234";
 944   u32 bytes_to_send = 64 << 10, mbytes;
 945   u32 tmp;
 946   mheap_t *h;
 947   session_t *session;
 948   int i;
 949   int i_am_master = 1, drop_packets = 0, test_return_packets = 0;
 950
 951   clib_mem_init (0, 256 << 20);
 952
 953   heap = clib_mem_get_per_cpu_heap ();
 954   h = mheap_header (heap);
 955
 956   /* make the main heap thread-safe */
 957   h->flags |= MHEAP_FLAG_THREAD_SAFE;
 958
 959   vec_validate (utm->rx_buf, 65536);
 960
 961   utm->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
 962
 963   utm->my_pid = getpid ();
 964   utm->configured_segment_size = 1 << 20;
 965
 966   clib_time_init (&utm->clib_time);
 967   init_error_string_table (utm);
 968   svm_fifo_segment_init (0x200000000ULL, 20);
 969   unformat_init_command_line (a, argv);
 970
 971   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
 972     {
 973       if (unformat (a, "chroot prefix %s", &chroot_prefix))
 974         {
 975           vl_set_memory_root_path ((char *) chroot_prefix);
 976         }
 977       else if (unformat (a, "uri %s", &uri))
 978         ;
 979       else if (unformat (a, "segment-size %dM", &tmp))
 980         utm->configured_segment_size = tmp << 20;
 981       else if (unformat (a, "segment-size %dG", &tmp))
 982         utm->configured_segment_size = tmp << 30;
 983       else if (unformat (a, "master"))
 984         i_am_master = 1;
 985       else if (unformat (a, "slave"))
 986         i_am_master = 0;
 987       else if (unformat (a, "drop"))
 988         drop_packets = 1;
 989       else if (unformat (a, "test"))
 990         test_return_packets = 1;
 991       else if (unformat (a, "mbytes %d", &mbytes))
 992         {
 993           bytes_to_send = mbytes << 20;
 994         }
 995       else
 996         {
 997           fformat (stderr, "%s: usage [master|slave]\n");
 998           exit (1);
 999         }
1000     }
1001
1002   if (uri)
1003     {
1004       utm->uri = format (0, "%s%c", uri, 0);
1005       utm->connect_uri = format (0, "%s%c", uri, 0);
1006     }
1007   else
1008     {
1009       utm->uri = format (0, "%s%c", bind_uri, 0);
1010       utm->connect_uri = format (0, "%s%c", connect_uri, 0);
1011     }
1012
1013   utm->i_am_master = i_am_master;
1014   utm->segment_main = &svm_fifo_segment_main;
1015   utm->drop_packets = drop_packets;
1016   utm->test_return_packets = test_return_packets;
1017   utm->bytes_to_send = bytes_to_send;
1018
1019   setup_signal_handlers ();
1020   uri_api_hookup (utm);
1021
1022   if (connect_to_vpp (i_am_master ? "uri_tcp_server" : "uri_tcp_client") < 0)
1023     {
1024       svm_region_exit ();
1025       fformat (stderr, "Couldn't connect to vpe, exiting...\n");
1026       exit (1);
1027     }
1028
1029   if (i_am_master == 0)
1030     {
1031       client_test (utm);
1032       vl_client_disconnect_from_vlib ();
1033       exit (0);
1034     }
1035
1036   /* $$$$ hack preallocation */
1037   for (i = 0; i < 200000; i++)
1038     {
1039       pool_get (utm->sessions, session);
1040       memset (session, 0, sizeof (*session));
1041     }
1042   for (i = 0; i < 200000; i++)
1043     pool_put_index (utm->sessions, i);
1044
1045   server_test (utm);
1046
1047   vl_client_disconnect_from_vlib ();
1048   exit (0);
1049 }
1050
1051 /*
1052  * fd.io coding-style-patch-verification: ON
1053  *
1054  * Local Variables:
1055  * eval: (c-set-style "gnu")
1056  * End:
1057  */