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