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 #define TCP_ECHO_DBG 0
  42 #define DBG(_fmt,_args...)                      \
  43     if (TCP_ECHO_DBG)                           \
  44       clib_warning (_fmt, _args)
  45
  46 typedef struct
  47 {
  48   svm_fifo_t *server_rx_fifo;
  49   svm_fifo_t *server_tx_fifo;
  50
  51   svm_msg_q_t *vpp_evt_q;
  52
  53   u64 vpp_session_handle;
  54   u64 bytes_sent;
  55   u64 bytes_to_send;
  56   volatile u64 bytes_received;
  57   volatile u64 bytes_to_receive;
  58   f64 start;
  59 } session_t;
  60
  61 typedef enum
  62 {
  63   STATE_START,
  64   STATE_ATTACHED,
  65   STATE_LISTEN,
  66   STATE_READY,
  67   STATE_DISCONNECTING,
  68   STATE_FAILED,
  69   STATE_DETACHED
  70 } connection_state_t;
  71
  72 typedef struct
  73 {
  74   /* vpe input queue */
  75   svm_queue_t *vl_input_queue;
  76
  77   /* API client handle */
  78   u32 my_client_index;
  79
  80   /* The URI we're playing with */
  81   u8 *uri;
  82
  83   /* Session pool */
  84   session_t *sessions;
  85
  86   /* Hash table for disconnect processing */
  87   uword *session_index_by_vpp_handles;
  88
  89   /* intermediate rx buffer */
  90   u8 *rx_buf;
  91
  92   /* URI for slave's connect */
  93   u8 *connect_uri;
  94
  95   u32 connected_session_index;
  96
  97   int i_am_master;
  98
  99   /* drop all packets */
 100   int no_return;
 101
 102   /* Our event queue */
 103   svm_msg_q_t *our_event_queue;
 104
 105   u8 *socket_name;
 106
 107   pid_t my_pid;
 108
 109   /* For deadman timers */
 110   clib_time_t clib_time;
 111
 112   /* State of the connection, shared between msg RX thread and main thread */
 113   volatile connection_state_t state;
 114
 115   /* Signal variables */
 116   volatile int time_to_stop;
 117   volatile int time_to_print_stats;
 118
 119   u32 configured_segment_size;
 120
 121   /* VNET_API_ERROR_FOO -> "Foo" hash table */
 122   uword *error_string_by_error_number;
 123
 124   u8 *connect_test_data;
 125   pthread_t *client_thread_handles;
 126   u32 *thread_args;
 127   u32 client_bytes_received;
 128   u8 test_return_packets;
 129   u64 bytes_to_send;
 130   u32 fifo_size;
 131
 132   u32 n_clients;
 133   u64 tx_total;
 134   u64 rx_total;
 135
 136   volatile u32 n_clients_connected;
 137   volatile u32 n_active_clients;
 138
 139
 140   /** Flag that decides if socket, instead of svm, api is used to connect to
 141    * vpp. If sock api is used, shm binary api is subsequently bootstrapped
 142    * and all other messages are exchanged using shm IPC. */
 143   u8 use_sock_api;
 144
 145   svm_fifo_segment_main_t segment_main;
 146 } echo_main_t;
 147
 148 echo_main_t echo_main;
 149
 150 #if CLIB_DEBUG > 0
 151 #define NITER 10000
 152 #else
 153 #define NITER 4000000
 154 #endif
 155
 156 const char test_srv_crt_rsa[] =
 157   "-----BEGIN CERTIFICATE-----\r\n"
 158   "MIID5zCCAs+gAwIBAgIJALeMYCEHrTtJMA0GCSqGSIb3DQEBCwUAMIGJMQswCQYD\r\n"
 159   "VQQGEwJVUzELMAkGA1UECAwCQ0ExETAPBgNVBAcMCFNhbiBKb3NlMQ4wDAYDVQQK\r\n"
 160   "DAVDaXNjbzEOMAwGA1UECwwFZmQuaW8xFjAUBgNVBAMMDXRlc3R0bHMuZmQuaW8x\r\n"
 161   "IjAgBgkqhkiG9w0BCQEWE3ZwcC1kZXZAbGlzdHMuZmQuaW8wHhcNMTgwMzA1MjEx\r\n"
 162   "NTEyWhcNMjgwMzAyMjExNTEyWjCBiTELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNB\r\n"
 163   "MREwDwYDVQQHDAhTYW4gSm9zZTEOMAwGA1UECgwFQ2lzY28xDjAMBgNVBAsMBWZk\r\n"
 164   "LmlvMRYwFAYDVQQDDA10ZXN0dGxzLmZkLmlvMSIwIAYJKoZIhvcNAQkBFhN2cHAt\r\n"
 165   "ZGV2QGxpc3RzLmZkLmlvMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA\r\n"
 166   "4C1k8a1DuStgggqT4o09fP9sJ2dC54bxhS/Xk2VEfaIZ222WSo4X/syRVfVy9Yah\r\n"
 167   "cpI1zJ/RDxaZSFhgA+nPZBrFMsrULkrdAOpOVj8eDEp9JuWdO2ODSoFnCvLxcYWB\r\n"
 168   "Yc5kHryJpEaGJl1sFQSesnzMFty/59ta0stk0Fp8r5NhIjWvSovGzPo6Bhz+VS2c\r\n"
 169   "ebIZh4x1t2hHaFcgm0qJoJ6DceReWCW8w+yOVovTolGGq+bpb2Hn7MnRSZ2K2NdL\r\n"
 170   "+aLXpkZbS/AODP1FF2vTO1mYL290LO7/51vJmPXNKSDYMy5EvILr5/VqtjsFCwRL\r\n"
 171   "Q4jcM/+GeHSAFWx4qIv0BwIDAQABo1AwTjAdBgNVHQ4EFgQUWa1SOB37xmT53tZQ\r\n"
 172   "aXuLLhRI7U8wHwYDVR0jBBgwFoAUWa1SOB37xmT53tZQaXuLLhRI7U8wDAYDVR0T\r\n"
 173   "BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAoUht13W4ya27NVzQuCMvqPWL3VM4\r\n"
 174   "3xbPFk02FaGz/WupPu276zGlzJAZrbuDcQowwwU1Ni1Yygxl96s1c2M5rHDTrOKG\r\n"
 175   "rK0hbkSFBo+i6I8u4HiiQ4rYmG0Hv6+sXn3of0HsbtDPGgWZoipPWDljPYEURu3e\r\n"
 176   "3HRe/Dtsj9CakBoSDzs8ndWaBR+f4sM9Tk1cjD46Gq2T/qpSPXqKxEUXlzhdCAn4\r\n"
 177   "twub17Bq2kykHpppCwPg5M+v30tHG/R2Go15MeFWbEJthFk3TZMjKL7UFs7fH+x2\r\n"
 178   "wSonXb++jY+KmCb93C+soABBizE57g/KmiR2IxQ/LMjDik01RSUIaM0lLA==\r\n"
 179   "-----END CERTIFICATE-----\r\n";
 180 const u32 test_srv_crt_rsa_len = sizeof (test_srv_crt_rsa);
 181
 182 const char test_srv_key_rsa[] =
 183   "-----BEGIN PRIVATE KEY-----\r\n"
 184   "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDgLWTxrUO5K2CC\r\n"
 185   "CpPijT18/2wnZ0LnhvGFL9eTZUR9ohnbbZZKjhf+zJFV9XL1hqFykjXMn9EPFplI\r\n"
 186   "WGAD6c9kGsUyytQuSt0A6k5WPx4MSn0m5Z07Y4NKgWcK8vFxhYFhzmQevImkRoYm\r\n"
 187   "XWwVBJ6yfMwW3L/n21rSy2TQWnyvk2EiNa9Ki8bM+joGHP5VLZx5shmHjHW3aEdo\r\n"
 188   "VyCbSomgnoNx5F5YJbzD7I5Wi9OiUYar5ulvYefsydFJnYrY10v5otemRltL8A4M\r\n"
 189   "/UUXa9M7WZgvb3Qs7v/nW8mY9c0pINgzLkS8guvn9Wq2OwULBEtDiNwz/4Z4dIAV\r\n"
 190   "bHioi/QHAgMBAAECggEBAMzGipP8+oT166U+NlJXRFifFVN1DvdhG9PWnOxGL+c3\r\n"
 191   "ILmBBC08WQzmHshPemBvR6DZkA1H23cV5JTiLWrFtC00CvhXsLRMrE5+uWotI6yE\r\n"
 192   "iofybMroHvD6/X5R510UX9hQ6MHu5ShLR5VZ9zXHz5MpTmB/60jG5dLx+jgcwBK8\r\n"
 193   "LuGv2YB/WCUwT9QJ3YU2eaingnXtz/MrFbkbltrqlnBdlD+kTtw6Yac9y1XuuQXc\r\n"
 194   "BPeulLNDuPolJVWbUvDBZrpt2dXTgz8ws1sv+wCNE0xwQJsqW4Nx3QkpibUL9RUr\r\n"
 195   "CVbKlNfa9lopT6nGKlgX69R/uH35yh9AOsfasro6w0ECgYEA82UJ8u/+ORah+0sF\r\n"
 196   "Q0FfW5MTdi7OAUHOz16pUsGlaEv0ERrjZxmAkHA/VRwpvDBpx4alCv0Hc39PFLIk\r\n"
 197   "nhSsM2BEuBkTAs6/GaoNAiBtQVE/hN7awNRWVmlieS0go3Y3dzaE9IUMyj8sPOFT\r\n"
 198   "5JdJ6BM69PHKCkY3dKdnnfpFEuECgYEA68mRpteunF1mdZgXs+WrN+uLlRrQR20F\r\n"
 199   "ZyMYiUCH2Dtn26EzA2moy7FipIIrQcX/j+KhYNGM3e7MU4LymIO29E18mn8JODnH\r\n"
 200   "sQOXzBTsf8A4yIVMkcuQD3bfb0JiUGYUPOidTp2N7IJA7+6Yc3vQOyb74lnKnJoO\r\n"
 201   "gougPT2wS+cCgYAn7muzb6xFsXDhyW0Tm6YJYBfRS9yAWEuVufINobeBZPSl2cN1\r\n"
 202   "Jrnw+HlrfTNbrJWuJmjtZJXUXQ6cVp2rUbjutNyRV4vG6iRwEXYQ40EJdkr1gZpi\r\n"
 203   "CHQhuShuuPih2MNAy7EEbM+sXrDjTBR3bFqzuHPzu7dp+BshCFX3lRfAAQKBgGQt\r\n"
 204   "K5i7IhCFDjb/+3IPLgOAK7mZvsvZ4eXD33TQ2eZgtut1PXtBtNl17/b85uv293Fm\r\n"
 205   "VDISVcsk3eLNS8zIiT6afUoWlxAwXEs0v5WRfjl4radkGvgGiJpJYvyeM67877RB\r\n"
 206   "EDSKc/X8ESLfOB44iGvZUEMG6zJFscx9DgN25iQZAoGAbyd+JEWwdVH9/K3IH1t2\r\n"
 207   "PBkZX17kNWv+iVM1WyFjbe++vfKZCrOJiyiqhDeEqgrP3AuNMlaaduC3VRC3G5oV\r\n"
 208   "Mj1tlhDWQ/qhvKdCKNdIVQYDE75nw+FRWV8yYkHAnXYW3tNoweDIwixE0hkPR1bc\r\n"
 209   "oEjPLVNtx8SOj/M4rhaPT3I=\r\n" "-----END PRIVATE KEY-----\r\n";
 210 const u32 test_srv_key_rsa_len = sizeof (test_srv_key_rsa);
 211
 212 static u8 *
 213 format_api_error (u8 * s, va_list * args)
 214 {
 215   echo_main_t *em = &echo_main;
 216   i32 error = va_arg (*args, u32);
 217   uword *p;
 218
 219   p = hash_get (em->error_string_by_error_number, -error);
 220
 221   if (p)
 222     s = format (s, "%s", p[0]);
 223   else
 224     s = format (s, "%d", error);
 225   return s;
 226 }
 227
 228 static void
 229 init_error_string_table (echo_main_t * em)
 230 {
 231   em->error_string_by_error_number = hash_create (0, sizeof (uword));
 232
 233 #define _(n,v,s) hash_set (em->error_string_by_error_number, -v, s);
 234   foreach_vnet_api_error;
 235 #undef _
 236
 237   hash_set (em->error_string_by_error_number, 99, "Misc");
 238 }
 239
 240 static void handle_mq_event (session_event_t * e);
 241
 242 static int
 243 wait_for_state_change (echo_main_t * em, connection_state_t state)
 244 {
 245   svm_msg_q_msg_t msg;
 246   session_event_t *e;
 247   f64 timeout;
 248
 249 #if CLIB_DEBUG > 0
 250 #define TIMEOUT 600.0
 251 #else
 252 #define TIMEOUT 600.0
 253 #endif
 254
 255   timeout = clib_time_now (&em->clib_time) + TIMEOUT;
 256
 257   while (clib_time_now (&em->clib_time) < timeout)
 258     {
 259       if (em->state == state)
 260         return 0;
 261       if (em->state == STATE_FAILED)
 262         return -1;
 263       if (em->time_to_stop == 1)
 264         return 0;
 265       if (!em->our_event_queue || em->state < STATE_ATTACHED)
 266         continue;
 267
 268       if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_NOWAIT, 0))
 269         continue;
 270       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
 271       handle_mq_event (e);
 272       svm_msg_q_free_msg (em->our_event_queue, &msg);
 273     }
 274   clib_warning ("timeout waiting for state %d", state);
 275   return -1;
 276 }
 277
 278 void
 279 application_send_attach (echo_main_t * em)
 280 {
 281   vl_api_application_attach_t *bmp;
 282   vl_api_application_tls_cert_add_t *cert_mp;
 283   vl_api_application_tls_key_add_t *key_mp;
 284
 285   bmp = vl_msg_api_alloc (sizeof (*bmp));
 286   clib_memset (bmp, 0, sizeof (*bmp));
 287
 288   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH);
 289   bmp->client_index = em->my_client_index;
 290   bmp->context = ntohl (0xfeedface);
 291   bmp->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_ACCEPT_REDIRECT;
 292   bmp->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_ADD_SEGMENT;
 293   bmp->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_USE_MQ_FOR_CTRL_MSGS;
 294   bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 16;
 295   bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = em->fifo_size;
 296   bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = em->fifo_size;
 297   bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
 298   bmp->options[APP_OPTIONS_SEGMENT_SIZE] = 256 << 20;
 299   bmp->options[APP_OPTIONS_EVT_QUEUE_SIZE] = 256;
 300   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 301
 302   cert_mp = vl_msg_api_alloc (sizeof (*cert_mp) + test_srv_crt_rsa_len);
 303   clib_memset (cert_mp, 0, sizeof (*cert_mp));
 304   cert_mp->_vl_msg_id = ntohs (VL_API_APPLICATION_TLS_CERT_ADD);
 305   cert_mp->client_index = em->my_client_index;
 306   cert_mp->context = ntohl (0xfeedface);
 307   cert_mp->cert_len = clib_host_to_net_u16 (test_srv_crt_rsa_len);
 308   clib_memcpy_fast (cert_mp->cert, test_srv_crt_rsa, test_srv_crt_rsa_len);
 309   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cert_mp);
 310
 311   key_mp = vl_msg_api_alloc (sizeof (*key_mp) + test_srv_key_rsa_len);
 312   clib_memset (key_mp, 0, sizeof (*key_mp) + test_srv_key_rsa_len);
 313   key_mp->_vl_msg_id = ntohs (VL_API_APPLICATION_TLS_KEY_ADD);
 314   key_mp->client_index = em->my_client_index;
 315   key_mp->context = ntohl (0xfeedface);
 316   key_mp->key_len = clib_host_to_net_u16 (test_srv_key_rsa_len);
 317   clib_memcpy_fast (key_mp->key, test_srv_key_rsa, test_srv_key_rsa_len);
 318   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & key_mp);
 319 }
 320
 321 static int
 322 application_attach (echo_main_t * em)
 323 {
 324   application_send_attach (em);
 325   if (wait_for_state_change (em, STATE_ATTACHED))
 326     {
 327       clib_warning ("timeout waiting for STATE_ATTACHED");
 328       return -1;
 329     }
 330   return 0;
 331 }
 332
 333 void
 334 application_detach (echo_main_t * em)
 335 {
 336   vl_api_application_detach_t *bmp;
 337   bmp = vl_msg_api_alloc (sizeof (*bmp));
 338   clib_memset (bmp, 0, sizeof (*bmp));
 339
 340   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH);
 341   bmp->client_index = em->my_client_index;
 342   bmp->context = ntohl (0xfeedface);
 343   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 344
 345   DBG ("%s", "Sent detach");
 346 }
 347
 348 static int
 349 ssvm_segment_attach (char *name, ssvm_segment_type_t type, int fd)
 350 {
 351   svm_fifo_segment_create_args_t _a, *a = &_a;
 352   svm_fifo_segment_main_t *sm = &echo_main.segment_main;
 353   int rv;
 354
 355   clib_memset (a, 0, sizeof (*a));
 356   a->segment_name = (char *) name;
 357   a->segment_type = type;
 358
 359   if (type == SSVM_SEGMENT_MEMFD)
 360     a->memfd_fd = fd;
 361
 362   if ((rv = svm_fifo_segment_attach (sm, a)))
 363     {
 364       clib_warning ("svm_fifo_segment_attach ('%s') failed", name);
 365       return rv;
 366     }
 367
 368   vec_reset_length (a->new_segment_indices);
 369   return 0;
 370 }
 371
 372 static void
 373 vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
 374                                            mp)
 375 {
 376   echo_main_t *em = &echo_main;
 377   int *fds = 0;
 378   u32 n_fds = 0;
 379
 380   if (mp->retval)
 381     {
 382       clib_warning ("attach failed: %U", format_api_error,
 383                     clib_net_to_host_u32 (mp->retval));
 384       goto failed;
 385     }
 386
 387   if (mp->segment_name_length == 0)
 388     {
 389       clib_warning ("segment_name_length zero");
 390       goto failed;
 391     }
 392
 393   ASSERT (mp->app_event_queue_address);
 394   em->our_event_queue = uword_to_pointer (mp->app_event_queue_address,
 395                                           svm_msg_q_t *);
 396
 397   if (mp->n_fds)
 398     {
 399       vec_validate (fds, mp->n_fds);
 400       vl_socket_client_recv_fd_msg (fds, mp->n_fds, 5);
 401
 402       if (mp->fd_flags & SESSION_FD_F_VPP_MQ_SEGMENT)
 403         if (ssvm_segment_attach (0, SSVM_SEGMENT_MEMFD, fds[n_fds++]))
 404           goto failed;
 405
 406       if (mp->fd_flags & SESSION_FD_F_MEMFD_SEGMENT)
 407         if (ssvm_segment_attach ((char *) mp->segment_name,
 408                                  SSVM_SEGMENT_MEMFD, fds[n_fds++]))
 409           goto failed;
 410
 411       if (mp->fd_flags & SESSION_FD_F_MQ_EVENTFD)
 412         svm_msg_q_set_consumer_eventfd (em->our_event_queue, fds[n_fds++]);
 413
 414       vec_free (fds);
 415     }
 416   else
 417     {
 418       if (ssvm_segment_attach ((char *) mp->segment_name, SSVM_SEGMENT_SHM,
 419                                -1))
 420         goto failed;
 421     }
 422
 423   em->state = STATE_ATTACHED;
 424   return;
 425 failed:
 426   em->state = STATE_FAILED;
 427   return;
 428 }
 429
 430 static void
 431 vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t *
 432                                            mp)
 433 {
 434   if (mp->retval)
 435     clib_warning ("detach returned with err: %d", mp->retval);
 436   echo_main.state = STATE_DETACHED;
 437 }
 438
 439 static void
 440 stop_signal (int signum)
 441 {
 442   echo_main_t *um = &echo_main;
 443
 444   um->time_to_stop = 1;
 445 }
 446
 447 static void
 448 stats_signal (int signum)
 449 {
 450   echo_main_t *um = &echo_main;
 451
 452   um->time_to_print_stats = 1;
 453 }
 454
 455 static clib_error_t *
 456 setup_signal_handlers (void)
 457 {
 458   signal (SIGINT, stats_signal);
 459   signal (SIGQUIT, stop_signal);
 460   signal (SIGTERM, stop_signal);
 461
 462   return 0;
 463 }
 464
 465 void
 466 vlib_cli_output (struct vlib_main_t *vm, char *fmt, ...)
 467 {
 468   clib_warning ("BUG");
 469 }
 470
 471 int
 472 connect_to_vpp (char *name)
 473 {
 474   echo_main_t *em = &echo_main;
 475   api_main_t *am = &api_main;
 476
 477   if (em->use_sock_api)
 478     {
 479       if (vl_socket_client_connect ((char *) em->socket_name, name,
 480                                     0 /* default rx, tx buffer */ ))
 481         {
 482           clib_warning ("socket connect failed");
 483           return -1;
 484         }
 485
 486       if (vl_socket_client_init_shm (0))
 487         {
 488           clib_warning ("init shm api failed");
 489           return -1;
 490         }
 491     }
 492   else
 493     {
 494       if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0)
 495         {
 496           clib_warning ("shmem connect failed");
 497           return -1;
 498         }
 499     }
 500   em->vl_input_queue = am->shmem_hdr->vl_input_queue;
 501   em->my_client_index = am->my_client_index;
 502   return 0;
 503 }
 504
 505 void
 506 disconnect_from_vpp (echo_main_t * em)
 507 {
 508   if (em->use_sock_api)
 509     vl_socket_client_disconnect ();
 510   else
 511     vl_client_disconnect_from_vlib ();
 512 }
 513
 514 static void
 515 vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
 516 {
 517   svm_fifo_segment_main_t *sm = &echo_main.segment_main;
 518   svm_fifo_segment_create_args_t _a, *a = &_a;
 519   int rv;
 520
 521   clib_memset (a, 0, sizeof (*a));
 522   a->segment_name = (char *) mp->segment_name;
 523   a->segment_size = mp->segment_size;
 524   /* Attach to the segment vpp created */
 525   rv = svm_fifo_segment_attach (sm, a);
 526   if (rv)
 527     {
 528       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 529                     mp->segment_name);
 530       return;
 531     }
 532   clib_warning ("Mapped new segment '%s' size %d", mp->segment_name,
 533                 mp->segment_size);
 534 }
 535
 536 static void
 537 session_print_stats (echo_main_t * em, session_t * session)
 538 {
 539   f64 deltat;
 540   u64 bytes;
 541
 542   deltat = clib_time_now (&em->clib_time) - session->start;
 543   bytes = em->i_am_master ? session->bytes_received : em->bytes_to_send;
 544   fformat (stdout, "Finished in %.6f\n", deltat);
 545   fformat (stdout, "%.4f Gbit/second\n", (bytes * 8.0) / deltat / 1e9);
 546 }
 547
 548 static void
 549 test_recv_bytes (session_t * s, u8 * rx_buf, u32 n_read)
 550 {
 551   int i;
 552   for (i = 0; i < n_read; i++)
 553     {
 554       if (rx_buf[i] != ((s->bytes_received + i) & 0xff))
 555         {
 556           clib_warning ("error at byte %lld, 0x%x not 0x%x",
 557                         s->bytes_received + i, rx_buf[i],
 558                         ((s->bytes_received + i) & 0xff));
 559         }
 560     }
 561 }
 562
 563 static void
 564 recv_test_chunk (echo_main_t * em, session_t * s, u8 * rx_buf)
 565 {
 566   svm_fifo_t *rx_fifo = s->server_rx_fifo;
 567   u32 n_read_now, n_to_read;
 568   int n_read;
 569
 570   n_to_read = svm_fifo_max_dequeue (rx_fifo);
 571   svm_fifo_unset_event (rx_fifo);
 572
 573   do
 574     {
 575       n_read_now = clib_min (vec_len (rx_buf), n_to_read);
 576       n_read = svm_fifo_dequeue_nowait (rx_fifo, n_read_now, rx_buf);
 577       if (n_read <= 0)
 578         break;
 579
 580       if (n_read_now != n_read)
 581         clib_warning ("huh?");
 582
 583       if (em->test_return_packets)
 584         test_recv_bytes (s, rx_buf, n_read);
 585
 586       n_to_read -= n_read;
 587       s->bytes_received += n_read;
 588       s->bytes_to_receive -= n_read;
 589     }
 590   while (n_to_read > 0);
 591 }
 592
 593 void
 594 client_handle_fifo_event_rx (echo_main_t * em, session_event_t * e,
 595                              u8 * rx_buf)
 596 {
 597   session_t *s;
 598
 599   s = pool_elt_at_index (em->sessions, e->fifo->client_session_index);
 600   recv_test_chunk (em, s, rx_buf);
 601 }
 602
 603 static void
 604 send_test_chunk (echo_main_t * em, session_t * s)
 605 {
 606   u64 test_buf_len, bytes_this_chunk, test_buf_offset;
 607   svm_fifo_t *tx_fifo = s->server_tx_fifo;
 608   u8 *test_data = em->connect_test_data;
 609   u32 enq_space = 16 << 10;
 610   int written;
 611
 612   test_buf_len = vec_len (test_data);
 613   test_buf_offset = s->bytes_sent % test_buf_len;
 614   bytes_this_chunk = clib_min (test_buf_len - test_buf_offset,
 615                                s->bytes_to_send);
 616   enq_space = svm_fifo_max_enqueue (tx_fifo);
 617
 618   bytes_this_chunk = clib_min (bytes_this_chunk, enq_space);
 619   written = svm_fifo_enqueue_nowait (tx_fifo, bytes_this_chunk,
 620                                      test_data + test_buf_offset);
 621
 622   if (written > 0)
 623     {
 624       s->bytes_to_send -= written;
 625       s->bytes_sent += written;
 626
 627       if (svm_fifo_set_event (tx_fifo))
 628         app_send_io_evt_to_vpp (s->vpp_evt_q, tx_fifo, FIFO_EVENT_APP_TX,
 629                                 0 /* do wait for mutex */ );
 630     }
 631 }
 632
 633 /*
 634  * Rx/Tx polling thread per connection
 635  */
 636 static void *
 637 client_thread_fn (void *arg)
 638 {
 639   echo_main_t *em = &echo_main;
 640   static u8 *rx_buf = 0;
 641   u32 session_index = *(u32 *) arg;
 642   session_t *s;
 643
 644   vec_validate (rx_buf, 1 << 20);
 645
 646   while (!em->time_to_stop && em->state != STATE_READY)
 647     ;
 648
 649   s = pool_elt_at_index (em->sessions, session_index);
 650   while (!em->time_to_stop)
 651     {
 652       send_test_chunk (em, s);
 653       recv_test_chunk (em, s, rx_buf);
 654       if (!s->bytes_to_send && !s->bytes_to_receive)
 655         break;
 656     }
 657
 658   DBG ("session %d done", session_index);
 659   em->tx_total += s->bytes_sent;
 660   em->rx_total += s->bytes_received;
 661   em->n_active_clients--;
 662
 663   pthread_exit (0);
 664 }
 665
 666 /*
 667  * Rx thread that handles all connections.
 668  *
 669  * Not used.
 670  */
 671 void *
 672 client_rx_thread_fn (void *arg)
 673 {
 674   session_event_t _e, *e = &_e;
 675   echo_main_t *em = &echo_main;
 676   static u8 *rx_buf = 0;
 677   svm_msg_q_msg_t msg;
 678
 679   vec_validate (rx_buf, 1 << 20);
 680
 681   while (!em->time_to_stop && em->state != STATE_READY)
 682     ;
 683
 684   while (!em->time_to_stop)
 685     {
 686       svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
 687       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
 688       switch (e->event_type)
 689         {
 690         case FIFO_EVENT_APP_RX:
 691           client_handle_fifo_event_rx (em, e, rx_buf);
 692           break;
 693         default:
 694           clib_warning ("unknown event type %d", e->event_type);
 695           break;
 696         }
 697       svm_msg_q_free_msg (em->our_event_queue, &msg);
 698     }
 699   pthread_exit (0);
 700 }
 701
 702 void
 703 client_send_connect (echo_main_t * em)
 704 {
 705   vl_api_connect_uri_t *cmp;
 706   cmp = vl_msg_api_alloc (sizeof (*cmp));
 707   clib_memset (cmp, 0, sizeof (*cmp));
 708
 709   cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
 710   cmp->client_index = em->my_client_index;
 711   cmp->context = ntohl (0xfeedface);
 712   memcpy (cmp->uri, em->connect_uri, vec_len (em->connect_uri));
 713   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cmp);
 714 }
 715
 716 void
 717 client_send_disconnect (echo_main_t * em, session_t * s)
 718 {
 719   vl_api_disconnect_session_t *dmp;
 720   dmp = vl_msg_api_alloc (sizeof (*dmp));
 721   clib_memset (dmp, 0, sizeof (*dmp));
 722   dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
 723   dmp->client_index = em->my_client_index;
 724   dmp->handle = s->vpp_session_handle;
 725   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & dmp);
 726 }
 727
 728 int
 729 client_disconnect (echo_main_t * em, session_t * s)
 730 {
 731   client_send_disconnect (em, s);
 732   pool_put (em->sessions, s);
 733   clib_memset (s, 0xfe, sizeof (*s));
 734   return 0;
 735 }
 736
 737 static void
 738 session_accepted_handler (session_accepted_msg_t * mp)
 739 {
 740   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 741   session_accepted_reply_msg_t *rmp;
 742   svm_fifo_t *rx_fifo, *tx_fifo;
 743   echo_main_t *em = &echo_main;
 744   session_t *session;
 745   static f64 start_time;
 746   u32 session_index;
 747   u8 *ip_str;
 748
 749   if (start_time == 0.0)
 750     start_time = clib_time_now (&em->clib_time);
 751
 752   ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4);
 753   clib_warning ("Accepted session from: %s:%d", ip_str,
 754                 clib_net_to_host_u16 (mp->port));
 755
 756   /* Allocate local session and set it up */
 757   pool_get (em->sessions, session);
 758   session_index = session - em->sessions;
 759
 760   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 761   rx_fifo->client_session_index = session_index;
 762   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 763   tx_fifo->client_session_index = session_index;
 764
 765   session->server_rx_fifo = rx_fifo;
 766   session->server_tx_fifo = tx_fifo;
 767   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
 768                                          svm_msg_q_t *);
 769
 770   /* Add it to lookup table */
 771   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 772
 773   em->state = STATE_READY;
 774
 775   /* Stats printing */
 776   if (pool_elts (em->sessions) && (pool_elts (em->sessions) % 20000) == 0)
 777     {
 778       f64 now = clib_time_now (&em->clib_time);
 779       fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
 780                pool_elts (em->sessions), now - start_time,
 781                (f64) pool_elts (em->sessions) / (now - start_time));
 782     }
 783
 784   /*
 785    * Send accept reply to vpp
 786    */
 787   app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
 788                              SESSION_CTRL_EVT_ACCEPTED_REPLY);
 789   rmp = (session_accepted_reply_msg_t *) app_evt->evt->data;
 790   rmp->handle = mp->handle;
 791   rmp->context = mp->context;
 792   app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
 793
 794   session->bytes_received = 0;
 795   session->start = clib_time_now (&em->clib_time);
 796 }
 797
 798 static void
 799 session_connected_handler (session_connected_msg_t * mp)
 800 {
 801   echo_main_t *em = &echo_main;
 802   session_t *session;
 803   u32 session_index;
 804   svm_fifo_t *rx_fifo, *tx_fifo;
 805   int rv;
 806
 807   if (mp->retval)
 808     {
 809       clib_warning ("connection failed with code: %U", format_api_error,
 810                     clib_net_to_host_u32 (mp->retval));
 811       em->state = STATE_FAILED;
 812       return;
 813     }
 814
 815   /*
 816    * Setup session
 817    */
 818
 819   pool_get (em->sessions, session);
 820   clib_memset (session, 0, sizeof (*session));
 821   session_index = session - em->sessions;
 822
 823   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 824   rx_fifo->client_session_index = session_index;
 825   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 826   tx_fifo->client_session_index = session_index;
 827
 828   session->server_rx_fifo = rx_fifo;
 829   session->server_tx_fifo = tx_fifo;
 830   session->vpp_session_handle = mp->handle;
 831   session->start = clib_time_now (&em->clib_time);
 832   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
 833                                          svm_msg_q_t *);
 834
 835   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 836
 837   /*
 838    * Start RX thread
 839    */
 840   em->thread_args[em->n_clients_connected] = session_index;
 841   rv = pthread_create (&em->client_thread_handles[em->n_clients_connected],
 842                        NULL /*attr */ , client_thread_fn,
 843                        (void *) &em->thread_args[em->n_clients_connected]);
 844   if (rv)
 845     {
 846       clib_warning ("pthread_create returned %d", rv);
 847       return;
 848     }
 849
 850   em->n_clients_connected += 1;
 851   clib_warning ("session %u (0x%llx) connected with local ip %U port %d",
 852                 session_index, mp->handle, format_ip46_address, mp->lcl_ip,
 853                 mp->is_ip4, clib_net_to_host_u16 (mp->lcl_port));
 854 }
 855
 856 static void
 857 session_disconnected_handler (session_disconnected_msg_t * mp)
 858 {
 859   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 860   session_disconnected_reply_msg_t *rmp;
 861   echo_main_t *em = &echo_main;
 862   session_t *session = 0;
 863   uword *p;
 864   int rv = 0;
 865
 866   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
 867   if (!p)
 868     {
 869       clib_warning ("couldn't find session key %llx", mp->handle);
 870       return;
 871     }
 872
 873   session = pool_elt_at_index (em->sessions, p[0]);
 874   hash_unset (em->session_index_by_vpp_handles, mp->handle);
 875   pool_put (em->sessions, session);
 876
 877   app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
 878                              SESSION_CTRL_EVT_DISCONNECTED_REPLY);
 879   rmp = (session_disconnected_reply_msg_t *) app_evt->evt->data;
 880   rmp->retval = rv;
 881   rmp->handle = mp->handle;
 882   rmp->context = mp->context;
 883   app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
 884
 885   session_print_stats (em, session);
 886 }
 887
 888 static void
 889 session_reset_handler (session_reset_msg_t * mp)
 890 {
 891   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 892   echo_main_t *em = &echo_main;
 893   session_reset_reply_msg_t *rmp;
 894   session_t *session = 0;
 895   uword *p;
 896   int rv = 0;
 897
 898   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
 899
 900   if (p)
 901     {
 902       session = pool_elt_at_index (em->sessions, p[0]);
 903       clib_warning ("got reset");
 904       /* Cleanup later */
 905       em->time_to_stop = 1;
 906     }
 907   else
 908     {
 909       clib_warning ("couldn't find session key %llx", mp->handle);
 910       return;
 911     }
 912
 913   app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
 914                              SESSION_CTRL_EVT_RESET_REPLY);
 915   rmp = (session_reset_reply_msg_t *) app_evt->evt->data;
 916   rmp->retval = rv;
 917   rmp->handle = mp->handle;
 918   app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
 919 }
 920
 921 static void
 922 handle_mq_event (session_event_t * e)
 923 {
 924   switch (e->event_type)
 925     {
 926     case SESSION_CTRL_EVT_ACCEPTED:
 927       session_accepted_handler ((session_accepted_msg_t *) e->data);
 928       break;
 929     case SESSION_CTRL_EVT_CONNECTED:
 930       session_connected_handler ((session_connected_msg_t *) e->data);
 931       break;
 932     case SESSION_CTRL_EVT_DISCONNECTED:
 933       session_disconnected_handler ((session_disconnected_msg_t *) e->data);
 934       break;
 935     case SESSION_CTRL_EVT_RESET:
 936       session_reset_handler ((session_reset_msg_t *) e->data);
 937       break;
 938     default:
 939       clib_warning ("unhandled %u", e->event_type);
 940     }
 941 }
 942
 943 static void
 944 clients_run (echo_main_t * em)
 945 {
 946   f64 start_time, deltat, timeout = 100.0;
 947   svm_msg_q_msg_t msg;
 948   session_event_t *e;
 949   session_t *s;
 950   int i;
 951
 952   /* Init test data */
 953   vec_validate (em->connect_test_data, 1024 * 1024 - 1);
 954   for (i = 0; i < vec_len (em->connect_test_data); i++)
 955     em->connect_test_data[i] = i & 0xff;
 956
 957   /*
 958    * Attach and connect the clients
 959    */
 960   if (application_attach (em))
 961     return;
 962
 963   for (i = 0; i < em->n_clients; i++)
 964     client_send_connect (em);
 965
 966   start_time = clib_time_now (&em->clib_time);
 967   while (em->n_clients_connected < em->n_clients
 968          && (clib_time_now (&em->clib_time) - start_time < timeout)
 969          && em->state != STATE_FAILED)
 970
 971     {
 972       svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
 973       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
 974       handle_mq_event (e);
 975       svm_msg_q_free_msg (em->our_event_queue, &msg);
 976     }
 977
 978   if (em->n_clients_connected != em->n_clients)
 979     {
 980       clib_warning ("failed to initialize all connections");
 981       return;
 982     }
 983
 984   /*
 985    * Initialize connections
 986    */
 987   for (i = 0; i < em->n_clients; i++)
 988     {
 989       s = pool_elt_at_index (em->sessions, i);
 990       s->bytes_to_send = em->bytes_to_send;
 991       if (!em->no_return)
 992         s->bytes_to_receive = em->bytes_to_send;
 993     }
 994   em->n_active_clients = em->n_clients_connected;
 995
 996   /*
 997    * Wait for client threads to send the data
 998    */
 999   start_time = clib_time_now (&em->clib_time);
1000   em->state = STATE_READY;
1001   while (em->n_active_clients)
1002     if (!svm_msg_q_is_empty (em->our_event_queue))
1003       {
1004         if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0))
1005           {
1006             clib_warning ("svm msg q returned");
1007             continue;
1008           }
1009         e = svm_msg_q_msg_data (em->our_event_queue, &msg);
1010         if (e->event_type != FIFO_EVENT_APP_RX)
1011           handle_mq_event (e);
1012         svm_msg_q_free_msg (em->our_event_queue, &msg);
1013       }
1014
1015   for (i = 0; i < em->n_clients; i++)
1016     {
1017       s = pool_elt_at_index (em->sessions, i);
1018       client_disconnect (em, s);
1019     }
1020
1021   /*
1022    * Stats and detach
1023    */
1024   deltat = clib_time_now (&em->clib_time) - start_time;
1025   fformat (stdout, "%lld bytes (%lld mbytes, %lld gbytes) in %.2f seconds\n",
1026            em->tx_total, em->tx_total / (1ULL << 20),
1027            em->tx_total / (1ULL << 30), deltat);
1028   fformat (stdout, "%.4f Gbit/second\n", (em->tx_total * 8.0) / deltat / 1e9);
1029
1030   application_detach (em);
1031 }
1032
1033 static void
1034 vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
1035 {
1036   echo_main_t *em = &echo_main;
1037
1038   if (mp->retval)
1039     {
1040       clib_warning ("bind failed: %U", format_api_error,
1041                     clib_net_to_host_u32 (mp->retval));
1042       em->state = STATE_FAILED;
1043       return;
1044     }
1045
1046   em->state = STATE_READY;
1047 }
1048
1049 static void
1050 vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp)
1051 {
1052   echo_main_t *em = &echo_main;
1053
1054   if (mp->retval != 0)
1055     clib_warning ("returned %d", ntohl (mp->retval));
1056
1057   em->state = STATE_START;
1058 }
1059
1060 u8 *
1061 format_ip4_address (u8 * s, va_list * args)
1062 {
1063   u8 *a = va_arg (*args, u8 *);
1064   return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]);
1065 }
1066
1067 u8 *
1068 format_ip6_address (u8 * s, va_list * args)
1069 {
1070   ip6_address_t *a = va_arg (*args, ip6_address_t *);
1071   u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon;
1072
1073   i_max_n_zero = ARRAY_LEN (a->as_u16);
1074   max_n_zeros = 0;
1075   i_first_zero = i_max_n_zero;
1076   n_zeros = 0;
1077   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
1078     {
1079       u32 is_zero = a->as_u16[i] == 0;
1080       if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16))
1081         {
1082           i_first_zero = i;
1083           n_zeros = 0;
1084         }
1085       n_zeros += is_zero;
1086       if ((!is_zero && n_zeros > max_n_zeros)
1087           || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros))
1088         {
1089           i_max_n_zero = i_first_zero;
1090           max_n_zeros = n_zeros;
1091           i_first_zero = ARRAY_LEN (a->as_u16);
1092           n_zeros = 0;
1093         }
1094     }
1095
1096   last_double_colon = 0;
1097   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
1098     {
1099       if (i == i_max_n_zero && max_n_zeros > 1)
1100         {
1101           s = format (s, "::");
1102           i += max_n_zeros - 1;
1103           last_double_colon = 1;
1104         }
1105       else
1106         {
1107           s = format (s, "%s%x",
1108                       (last_double_colon || i == 0) ? "" : ":",
1109                       clib_net_to_host_u16 (a->as_u16[i]));
1110           last_double_colon = 0;
1111         }
1112     }
1113
1114   return s;
1115 }
1116
1117 /* Format an IP46 address. */
1118 u8 *
1119 format_ip46_address (u8 * s, va_list * args)
1120 {
1121   ip46_address_t *ip46 = va_arg (*args, ip46_address_t *);
1122   ip46_type_t type = va_arg (*args, ip46_type_t);
1123   int is_ip4 = 1;
1124
1125   switch (type)
1126     {
1127     case IP46_TYPE_ANY:
1128       is_ip4 = ip46_address_is_ip4 (ip46);
1129       break;
1130     case IP46_TYPE_IP4:
1131       is_ip4 = 1;
1132       break;
1133     case IP46_TYPE_IP6:
1134       is_ip4 = 0;
1135       break;
1136     }
1137
1138   return is_ip4 ?
1139     format (s, "%U", format_ip4_address, &ip46->ip4) :
1140     format (s, "%U", format_ip6_address, &ip46->ip6);
1141 }
1142
1143 static void
1144 server_handle_fifo_event_rx (echo_main_t * em, session_event_t * e)
1145 {
1146   svm_fifo_t *rx_fifo, *tx_fifo;
1147   int n_read;
1148   session_t *session;
1149   int rv;
1150   u32 max_dequeue, offset, max_transfer, rx_buf_len;
1151
1152   rx_buf_len = vec_len (em->rx_buf);
1153   rx_fifo = e->fifo;
1154   session = pool_elt_at_index (em->sessions, rx_fifo->client_session_index);
1155   tx_fifo = session->server_tx_fifo;
1156
1157   max_dequeue = svm_fifo_max_dequeue (rx_fifo);
1158   /* Allow enqueuing of a new event */
1159   svm_fifo_unset_event (rx_fifo);
1160
1161   if (PREDICT_FALSE (!max_dequeue))
1162     return;
1163
1164   /* Read the max_dequeue */
1165   do
1166     {
1167       max_transfer = clib_min (rx_buf_len, max_dequeue);
1168       n_read = svm_fifo_dequeue_nowait (rx_fifo, max_transfer, em->rx_buf);
1169       if (n_read > 0)
1170         {
1171           max_dequeue -= n_read;
1172           session->bytes_received += n_read;
1173           session->bytes_to_receive -= n_read;
1174         }
1175
1176       /* Reflect if a non-drop session */
1177       if (!em->no_return && n_read > 0)
1178         {
1179           offset = 0;
1180           do
1181             {
1182               rv = svm_fifo_enqueue_nowait (tx_fifo, n_read,
1183                                             &em->rx_buf[offset]);
1184               if (rv > 0)
1185                 {
1186                   n_read -= rv;
1187                   offset += rv;
1188                 }
1189             }
1190           while ((rv <= 0 || n_read > 0) && !em->time_to_stop);
1191
1192           /* If event wasn't set, add one */
1193           if (svm_fifo_set_event (tx_fifo))
1194             app_send_io_evt_to_vpp (session->vpp_evt_q, tx_fifo,
1195                                     FIFO_EVENT_APP_TX, SVM_Q_WAIT);
1196         }
1197     }
1198   while ((n_read < 0 || max_dequeue > 0) && !em->time_to_stop);
1199 }
1200
1201 static void
1202 server_handle_event_queue (echo_main_t * em)
1203 {
1204   svm_msg_q_msg_t msg;
1205   session_event_t *e;
1206
1207   while (1)
1208     {
1209       svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
1210       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
1211       switch (e->event_type)
1212         {
1213         case FIFO_EVENT_APP_RX:
1214           server_handle_fifo_event_rx (em, e);
1215           break;
1216         default:
1217           handle_mq_event (e);
1218           break;
1219         }
1220       if (PREDICT_FALSE (em->time_to_stop == 1))
1221         break;
1222       if (PREDICT_FALSE (em->time_to_print_stats == 1))
1223         {
1224           em->time_to_print_stats = 0;
1225           fformat (stdout, "%d connections\n", pool_elts (em->sessions));
1226         }
1227       svm_msg_q_free_msg (em->our_event_queue, &msg);
1228     }
1229 }
1230
1231 void
1232 server_send_listen (echo_main_t * em)
1233 {
1234   vl_api_bind_uri_t *bmp;
1235   bmp = vl_msg_api_alloc (sizeof (*bmp));
1236   clib_memset (bmp, 0, sizeof (*bmp));
1237
1238   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
1239   bmp->client_index = em->my_client_index;
1240   bmp->context = ntohl (0xfeedface);
1241   memcpy (bmp->uri, em->uri, vec_len (em->uri));
1242   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
1243 }
1244
1245 int
1246 server_listen (echo_main_t * em)
1247 {
1248   server_send_listen (em);
1249   if (wait_for_state_change (em, STATE_READY))
1250     {
1251       clib_warning ("timeout waiting for STATE_READY");
1252       return -1;
1253     }
1254   return 0;
1255 }
1256
1257 void
1258 server_send_unbind (echo_main_t * em)
1259 {
1260   vl_api_unbind_uri_t *ump;
1261
1262   ump = vl_msg_api_alloc (sizeof (*ump));
1263   clib_memset (ump, 0, sizeof (*ump));
1264
1265   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
1266   ump->client_index = em->my_client_index;
1267   memcpy (ump->uri, em->uri, vec_len (em->uri));
1268   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & ump);
1269 }
1270
1271 int
1272 server_unbind (echo_main_t * em)
1273 {
1274   server_send_unbind (em);
1275   if (wait_for_state_change (em, STATE_START))
1276     {
1277       clib_warning ("timeout waiting for STATE_START");
1278       return -1;
1279     }
1280   return 0;
1281 }
1282
1283 void
1284 server_run (echo_main_t * em)
1285 {
1286   session_t *session;
1287   int i;
1288
1289   /* $$$$ hack preallocation */
1290   for (i = 0; i < 200000; i++)
1291     {
1292       pool_get (em->sessions, session);
1293       clib_memset (session, 0, sizeof (*session));
1294     }
1295   for (i = 0; i < 200000; i++)
1296     pool_put_index (em->sessions, i);
1297
1298   if (application_attach (em))
1299     return;
1300
1301   /* Bind to uri */
1302   if (server_listen (em))
1303     return;
1304
1305   /* Enter handle event loop */
1306   server_handle_event_queue (em);
1307
1308   /* Cleanup */
1309   server_send_unbind (em);
1310
1311   application_detach (em);
1312
1313   fformat (stdout, "Test complete...\n");
1314 }
1315
1316 static void
1317 vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t *
1318                                            mp)
1319 {
1320   echo_main_t *em = &echo_main;
1321   uword *p;
1322
1323   if (mp->retval)
1324     {
1325       clib_warning ("vpp complained about disconnect: %d",
1326                     ntohl (mp->retval));
1327       return;
1328     }
1329
1330   em->state = STATE_START;
1331
1332   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
1333   if (p)
1334     {
1335       hash_unset (em->session_index_by_vpp_handles, mp->handle);
1336     }
1337   else
1338     {
1339       clib_warning ("couldn't find session key %llx", mp->handle);
1340     }
1341 }
1342
1343 static void
1344   vl_api_application_tls_cert_add_reply_t_handler
1345   (vl_api_application_tls_cert_add_reply_t * mp)
1346 {
1347   if (mp->retval)
1348     clib_warning ("failed to add tls cert");
1349 }
1350
1351 static void
1352   vl_api_application_tls_key_add_reply_t_handler
1353   (vl_api_application_tls_key_add_reply_t * mp)
1354 {
1355   if (mp->retval)
1356     clib_warning ("failed to add tls key");
1357 }
1358
1359 #define foreach_tcp_echo_msg                                            \
1360 _(BIND_URI_REPLY, bind_uri_reply)                                       \
1361 _(UNBIND_URI_REPLY, unbind_uri_reply)                                   \
1362 _(DISCONNECT_SESSION_REPLY, disconnect_session_reply)                   \
1363 _(APPLICATION_ATTACH_REPLY, application_attach_reply)                   \
1364 _(APPLICATION_DETACH_REPLY, application_detach_reply)                   \
1365 _(MAP_ANOTHER_SEGMENT, map_another_segment)                             \
1366 _(APPLICATION_TLS_CERT_ADD_REPLY, application_tls_cert_add_reply)       \
1367 _(APPLICATION_TLS_KEY_ADD_REPLY, application_tls_key_add_reply)         \
1368
1369 void
1370 tcp_echo_api_hookup (echo_main_t * em)
1371 {
1372 #define _(N,n)                                                  \
1373     vl_msg_api_set_handlers(VL_API_##N, #n,                     \
1374                            vl_api_##n##_t_handler,              \
1375                            vl_noop_handler,                     \
1376                            vl_api_##n##_t_endian,               \
1377                            vl_api_##n##_t_print,                \
1378                            sizeof(vl_api_##n##_t), 1);
1379   foreach_tcp_echo_msg;
1380 #undef _
1381 }
1382
1383 int
1384 main (int argc, char **argv)
1385 {
1386   int i_am_server = 1, test_return_packets = 0;
1387   echo_main_t *em = &echo_main;
1388   svm_fifo_segment_main_t *sm = &em->segment_main;
1389   unformat_input_t _argv, *a = &_argv;
1390   u8 *chroot_prefix;
1391   u8 *uri = 0;
1392   u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234";
1393   u8 *connect_uri = (u8 *) "tcp://6.0.1.1/1234";
1394   u64 bytes_to_send = 64 << 10, mbytes;
1395   char *app_name;
1396   u32 tmp;
1397
1398   clib_mem_init_thread_safe (0, 256 << 20);
1399
1400   clib_memset (em, 0, sizeof (*em));
1401   em->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
1402   em->my_pid = getpid ();
1403   em->configured_segment_size = 1 << 20;
1404   em->socket_name = 0;
1405   em->use_sock_api = 1;
1406   em->fifo_size = 64 << 10;
1407   em->n_clients = 1;
1408
1409   clib_time_init (&em->clib_time);
1410   init_error_string_table (em);
1411   svm_fifo_segment_main_init (sm, HIGH_SEGMENT_BASEVA, 20);
1412   unformat_init_command_line (a, argv);
1413
1414   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
1415     {
1416       if (unformat (a, "chroot prefix %s", &chroot_prefix))
1417         {
1418           vl_set_memory_root_path ((char *) chroot_prefix);
1419         }
1420       else if (unformat (a, "uri %s", &uri))
1421         ;
1422       else if (unformat (a, "segment-size %dM", &tmp))
1423         em->configured_segment_size = tmp << 20;
1424       else if (unformat (a, "segment-size %dG", &tmp))
1425         em->configured_segment_size = tmp << 30;
1426       else if (unformat (a, "server"))
1427         i_am_server = 1;
1428       else if (unformat (a, "client"))
1429         i_am_server = 0;
1430       else if (unformat (a, "no-return"))
1431         em->no_return = 1;
1432       else if (unformat (a, "test"))
1433         test_return_packets = 1;
1434       else if (unformat (a, "mbytes %lld", &mbytes))
1435         {
1436           bytes_to_send = mbytes << 20;
1437         }
1438       else if (unformat (a, "gbytes %lld", &mbytes))
1439         {
1440           bytes_to_send = mbytes << 30;
1441         }
1442       else if (unformat (a, "socket-name %s", &em->socket_name))
1443         ;
1444       else if (unformat (a, "use-svm-api"))
1445         em->use_sock_api = 0;
1446       else if (unformat (a, "fifo-size %d", &tmp))
1447         em->fifo_size = tmp << 10;
1448       else if (unformat (a, "nclients %d", &em->n_clients))
1449         ;
1450       else
1451         {
1452           fformat (stderr, "%s: usage [master|slave]\n", argv[0]);
1453           exit (1);
1454         }
1455     }
1456
1457   if (!em->socket_name)
1458     em->socket_name = format (0, "%s%c", API_SOCKET_FILE, 0);
1459
1460   if (uri)
1461     {
1462       em->uri = format (0, "%s%c", uri, 0);
1463       em->connect_uri = format (0, "%s%c", uri, 0);
1464     }
1465   else
1466     {
1467       em->uri = format (0, "%s%c", bind_uri, 0);
1468       em->connect_uri = format (0, "%s%c", connect_uri, 0);
1469     }
1470
1471   em->i_am_master = i_am_server;
1472   em->test_return_packets = test_return_packets;
1473   em->bytes_to_send = bytes_to_send;
1474   em->time_to_stop = 0;
1475   vec_validate (em->rx_buf, 128 << 10);
1476   vec_validate (em->client_thread_handles, em->n_clients - 1);
1477   vec_validate (em->thread_args, em->n_clients - 1);
1478
1479   setup_signal_handlers ();
1480   tcp_echo_api_hookup (em);
1481
1482   app_name = i_am_server ? "tcp_echo_server" : "tcp_echo_client";
1483   if (connect_to_vpp (app_name) < 0)
1484     {
1485       svm_region_exit ();
1486       fformat (stderr, "Couldn't connect to vpe, exiting...\n");
1487       exit (1);
1488     }
1489
1490   if (i_am_server == 0)
1491     clients_run (em);
1492   else
1493     server_run (em);
1494
1495   /* Make sure detach finishes */
1496   wait_for_state_change (em, STATE_DETACHED);
1497
1498   disconnect_from_vpp (em);
1499   exit (0);
1500 }
1501
1502 /*
1503  * fd.io coding-style-patch-verification: ON
1504  *
1505  * Local Variables:
1506  * eval: (c-set-style "gnu")
1507  * End:
1508  */