src/plugins/hs_apps/sapi/quic_echo.c

   1 /*
   2  * Copyright (c) 2019 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 <vlibmemory/api.h>
  21
  22 #include <vpp/api/vpe_msg_enum.h>
  23 #include <svm/fifo_segment.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 QUIC_ECHO_DBG 0
  42 #define DBG(_fmt, _args...)                     \
  43     if (QUIC_ECHO_DBG)                          \
  44       clib_warning (_fmt, ##_args)
  45
  46 typedef struct
  47 {
  48   CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
  49 #define _(type, name) type name;
  50   foreach_app_session_field
  51 #undef _
  52   u64 vpp_session_handle;
  53   u64 bytes_sent;
  54   u64 bytes_to_send;
  55   volatile u64 bytes_received;
  56   volatile u64 bytes_to_receive;
  57   f64 start;
  58 } echo_session_t;
  59
  60 typedef enum
  61 {
  62   STATE_START,
  63   STATE_ATTACHED,
  64   STATE_LISTEN,
  65   STATE_READY,
  66   STATE_DISCONNECTING,
  67   STATE_FAILED,
  68   STATE_DETACHED
  69 } connection_state_t;
  70
  71 typedef enum
  72 {
  73   ECHO_EVT_START,               /* app starts */
  74   ECHO_EVT_FIRST_QCONNECT,      /* First connect Quic session sent */
  75   ECHO_EVT_LAST_QCONNECTED,     /* All Quic session are connected */
  76   ECHO_EVT_FIRST_SCONNECT,      /* First connect Stream session sent */
  77   ECHO_EVT_LAST_SCONNECTED,     /* All Stream session are connected */
  78   ECHO_EVT_LAST_BYTE,           /* Last byte received */
  79   ECHO_EVT_EXIT,                /* app exits */
  80 } echo_test_evt_t;
  81
  82 enum quic_session_type_t
  83 {
  84   QUIC_SESSION_TYPE_QUIC = 0,
  85   QUIC_SESSION_TYPE_STREAM = 1,
  86   QUIC_SESSION_TYPE_LISTEN = INT32_MAX,
  87 };
  88
  89 typedef struct _quic_echo_cb_vft
  90 {
  91   void (*quic_connected_cb) (session_connected_msg_t * mp, u32 session_index);
  92   void (*client_stream_connected_cb) (session_connected_msg_t * mp,
  93                                       u32 session_index);
  94   void (*server_stream_connected_cb) (session_connected_msg_t * mp,
  95                                       u32 session_index);
  96   void (*quic_accepted_cb) (session_accepted_msg_t * mp, u32 session_index);
  97   void (*client_stream_accepted_cb) (session_accepted_msg_t * mp,
  98                                      u32 session_index);
  99   void (*server_stream_accepted_cb) (session_accepted_msg_t * mp,
 100                                      u32 session_index);
 101 } quic_echo_cb_vft_t;
 102
 103
 104 typedef enum
 105 {
 106   RETURN_PACKETS_NOTEST,
 107   RETURN_PACKETS_LOG_WRONG,
 108   RETURN_PACKETS_ASSERT,
 109 } test_return_packets_t;
 110
 111 typedef struct
 112 {
 113   /* vpe input queue */
 114   svm_queue_t *vl_input_queue;
 115
 116   /* API client handle */
 117   u32 my_client_index;
 118
 119   /* The URI we're playing with */
 120   u8 *uri;
 121
 122   /* Session pool */
 123   echo_session_t *sessions;
 124
 125   /* Hash table for disconnect processing */
 126   uword *session_index_by_vpp_handles;
 127   /* Handle of vpp listener session */
 128   u64 listener_handle;
 129
 130   /* Hash table for shared segment_names */
 131   uword *shared_segment_handles;
 132   clib_spinlock_t segment_handles_lock;
 133
 134   /* intermediate rx buffer */
 135   u8 *rx_buf;
 136
 137   int i_am_master;
 138
 139   /* drop all packets */
 140   int no_return;
 141
 142   /* Our event queue */
 143   svm_msg_q_t *our_event_queue;
 144
 145   u8 *socket_name;
 146
 147   pid_t my_pid;
 148
 149   /* For deadman timers */
 150   clib_time_t clib_time;
 151
 152   /* State of the connection, shared between msg RX thread and main thread */
 153   volatile connection_state_t state;
 154
 155   /* Signal variables */
 156   volatile int time_to_stop;
 157
 158   /* VNET_API_ERROR_FOO -> "Foo" hash table */
 159   uword *error_string_by_error_number;
 160
 161   u8 *connect_test_data;
 162   pthread_t *client_thread_handles;
 163   u32 *thread_args;
 164   u8 test_return_packets;
 165   u64 bytes_to_send;
 166   u64 bytes_to_receive;
 167   u32 fifo_size;
 168
 169   u8 *appns_id;
 170   u64 appns_flags;
 171   u64 appns_secret;
 172
 173   u32 n_clients;                /* Target number of QUIC sessions */
 174   u32 n_stream_clients;         /* Target Number of STREAM sessions per QUIC session */
 175   volatile u32 n_quic_clients_connected;        /* Number of connected QUIC sessions */
 176   volatile u32 n_clients_connected;     /* Number of STREAM sessions connected */
 177
 178   u64 tx_total;
 179   u64 rx_total;
 180
 181   /* Event based timing : start & end depend on CLI specified events */
 182   u8 first_sconnect_sent;       /* Sent the first Stream session connect ? */
 183   f64 start_time;
 184   f64 end_time;
 185   u8 timing_start_event;
 186   u8 timing_end_event;
 187
 188   /* cb vft for QUIC scenarios */
 189   quic_echo_cb_vft_t cb_vft;
 190
 191   /** Flag that decides if socket, instead of svm, api is used to connect to
 192    * vpp. If sock api is used, shm binary api is subsequently bootstrapped
 193    * and all other messages are exchanged using shm IPC. */
 194   u8 use_sock_api;
 195
 196   /* Limit the number of incorrect data messages */
 197   int max_test_msg;
 198
 199   fifo_segment_main_t segment_main;
 200 } echo_main_t;
 201
 202 echo_main_t echo_main;
 203
 204 #if CLIB_DEBUG > 0
 205 #define NITER 10000
 206 #else
 207 #define NITER 4000000
 208 #endif
 209
 210 #if CLIB_DEBUG > 0
 211 #define TIMEOUT 10.0
 212 #else
 213 #define TIMEOUT 10.0
 214 #endif
 215
 216 u8 *
 217 format_quic_echo_state (u8 * s, va_list * args)
 218 {
 219   u32 state = va_arg (*args, u32);
 220   if (state == STATE_START)
 221     return format (s, "STATE_START");
 222   if (state == STATE_ATTACHED)
 223     return format (s, "STATE_ATTACHED");
 224   if (state == STATE_LISTEN)
 225     return format (s, "STATE_LISTEN");
 226   if (state == STATE_READY)
 227     return format (s, "STATE_READY");
 228   if (state == STATE_DISCONNECTING)
 229     return format (s, "STATE_DISCONNECTING");
 230   if (state == STATE_FAILED)
 231     return format (s, "STATE_FAILED");
 232   if (state == STATE_DETACHED)
 233     return format (s, "STATE_DETACHED");
 234   else
 235     return format (s, "unknown state");
 236 }
 237
 238 static u8 *
 239 format_api_error (u8 * s, va_list * args)
 240 {
 241   echo_main_t *em = &echo_main;
 242   i32 error = va_arg (*args, u32);
 243   uword *p;
 244
 245   p = hash_get (em->error_string_by_error_number, -error);
 246
 247   if (p)
 248     s = format (s, "%s", p[0]);
 249   else
 250     s = format (s, "%d", error);
 251   return s;
 252 }
 253
 254 static void
 255 quic_echo_notify_event (echo_main_t * em, echo_test_evt_t e)
 256 {
 257   if (em->timing_start_event == e)
 258     em->start_time = clib_time_now (&em->clib_time);
 259   else if (em->timing_end_event == e)
 260     em->end_time = clib_time_now (&em->clib_time);
 261 }
 262
 263 static uword
 264 echo_unformat_timing_event (unformat_input_t * input, va_list * args)
 265 {
 266   echo_test_evt_t *a = va_arg (*args, echo_test_evt_t *);
 267   if (unformat (input, "start"))
 268     *a = ECHO_EVT_START;
 269   else if (unformat (input, "qconnect"))
 270     *a = ECHO_EVT_FIRST_QCONNECT;
 271   else if (unformat (input, "qconnected"))
 272     *a = ECHO_EVT_LAST_QCONNECTED;
 273   else if (unformat (input, "sconnect"))
 274     *a = ECHO_EVT_FIRST_SCONNECT;
 275   else if (unformat (input, "sconnected"))
 276     *a = ECHO_EVT_LAST_SCONNECTED;
 277   else if (unformat (input, "lastbyte"))
 278     *a = ECHO_EVT_LAST_BYTE;
 279   else if (unformat (input, "exit"))
 280     *a = ECHO_EVT_EXIT;
 281   else
 282     return 0;
 283   return 1;
 284 }
 285
 286 u8 *
 287 echo_format_timing_event (u8 * s, va_list * args)
 288 {
 289   u32 timing_event = va_arg (*args, u32);
 290   if (timing_event == ECHO_EVT_START)
 291     return format (s, "start");
 292   if (timing_event == ECHO_EVT_FIRST_QCONNECT)
 293     return format (s, "qconnect");
 294   if (timing_event == ECHO_EVT_LAST_QCONNECTED)
 295     return format (s, "qconnected");
 296   if (timing_event == ECHO_EVT_FIRST_SCONNECT)
 297     return format (s, "sconnect");
 298   if (timing_event == ECHO_EVT_LAST_SCONNECTED)
 299     return format (s, "sconnected");
 300   if (timing_event == ECHO_EVT_LAST_BYTE)
 301     return format (s, "lastbyte");
 302   if (timing_event == ECHO_EVT_EXIT)
 303     return format (s, "exit");
 304   else
 305     return format (s, "unknown timing event");
 306 }
 307
 308 static void
 309 init_error_string_table (echo_main_t * em)
 310 {
 311   em->error_string_by_error_number = hash_create (0, sizeof (uword));
 312
 313 #define _(n,v,s) hash_set (em->error_string_by_error_number, -v, s);
 314   foreach_vnet_api_error;
 315 #undef _
 316
 317   hash_set (em->error_string_by_error_number, 99, "Misc");
 318 }
 319
 320 static void handle_mq_event (echo_main_t * em, session_event_t * e,
 321                              int handle_rx);
 322 static void echo_handle_rx (echo_main_t * em, session_event_t * e);
 323
 324 static int
 325 wait_for_segment_allocation (u64 segment_handle)
 326 {
 327   echo_main_t *em = &echo_main;
 328   f64 timeout;
 329   timeout = clib_time_now (&em->clib_time) + TIMEOUT;
 330   uword *segment_present;
 331   DBG ("Waiting for segment %lx...", segment_handle);
 332   while (clib_time_now (&em->clib_time) < timeout)
 333     {
 334       clib_spinlock_lock (&em->segment_handles_lock);
 335       segment_present = hash_get (em->shared_segment_handles, segment_handle);
 336       clib_spinlock_unlock (&em->segment_handles_lock);
 337       if (segment_present != 0)
 338         return 0;
 339       if (em->time_to_stop == 1)
 340         return 0;
 341     }
 342   DBG ("timeout waiting for segment_allocation %lx", segment_handle);
 343   return -1;
 344 }
 345
 346 static int
 347 wait_for_disconnected_sessions (echo_main_t * em)
 348 {
 349   f64 timeout;
 350   timeout = clib_time_now (&em->clib_time) + TIMEOUT;
 351   while (clib_time_now (&em->clib_time) < timeout)
 352     {
 353       if (hash_elts (em->session_index_by_vpp_handles) == 0)
 354         return 0;
 355     }
 356   DBG ("timeout waiting for disconnected_sessions");
 357   return -1;
 358 }
 359
 360 static int
 361 wait_for_state_change (echo_main_t * em, connection_state_t state,
 362                        f64 timeout)
 363 {
 364   svm_msg_q_msg_t msg;
 365   session_event_t *e;
 366   f64 end_time = clib_time_now (&em->clib_time) + timeout;
 367
 368   while (!timeout || clib_time_now (&em->clib_time) < end_time)
 369     {
 370       if (em->state == state)
 371         return 0;
 372       if (em->state == STATE_FAILED)
 373         return -1;
 374       if (em->time_to_stop == 1)
 375         return 0;
 376       if (!em->our_event_queue || em->state < STATE_ATTACHED)
 377         continue;
 378
 379       if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_NOWAIT, 0))
 380         continue;
 381       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
 382       handle_mq_event (em, e, 0 /* handle_rx */ );
 383       svm_msg_q_free_msg (em->our_event_queue, &msg);
 384     }
 385   clib_warning ("timeout waiting for state %U", format_quic_echo_state,
 386                 state);
 387   return -1;
 388 }
 389
 390 static void
 391 notify_rx_data_to_vpp (echo_session_t * s)
 392 {
 393   svm_fifo_t *f = s->tx_fifo;
 394   return;                       /* FOR NOW */
 395   if (svm_fifo_set_event (f))
 396     {
 397       DBG ("did send event");
 398       app_send_io_evt_to_vpp (s->vpp_evt_q, f->master_session_index,
 399                               SESSION_IO_EVT_TX, 0 /* noblock */ );
 400     }
 401 }
 402
 403 void
 404 application_send_attach (echo_main_t * em)
 405 {
 406   vl_api_application_attach_t *bmp;
 407   vl_api_application_tls_cert_add_t *cert_mp;
 408   vl_api_application_tls_key_add_t *key_mp;
 409
 410   bmp = vl_msg_api_alloc (sizeof (*bmp));
 411   clib_memset (bmp, 0, sizeof (*bmp));
 412
 413   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH);
 414   bmp->client_index = em->my_client_index;
 415   bmp->context = ntohl (0xfeedface);
 416   bmp->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_ACCEPT_REDIRECT;
 417   bmp->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_ADD_SEGMENT;
 418   bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 16;
 419   bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = em->fifo_size;
 420   bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = em->fifo_size;
 421   bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
 422   bmp->options[APP_OPTIONS_SEGMENT_SIZE] = 256 << 20;
 423   bmp->options[APP_OPTIONS_EVT_QUEUE_SIZE] = 256;
 424   if (em->appns_id)
 425     {
 426       bmp->namespace_id_len = vec_len (em->appns_id);
 427       clib_memcpy_fast (bmp->namespace_id, em->appns_id,
 428                         bmp->namespace_id_len);
 429       bmp->options[APP_OPTIONS_FLAGS] |= em->appns_flags;
 430       bmp->options[APP_OPTIONS_NAMESPACE_SECRET] = em->appns_secret;
 431     }
 432   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 433
 434   cert_mp = vl_msg_api_alloc (sizeof (*cert_mp) + test_srv_crt_rsa_len);
 435   clib_memset (cert_mp, 0, sizeof (*cert_mp));
 436   cert_mp->_vl_msg_id = ntohs (VL_API_APPLICATION_TLS_CERT_ADD);
 437   cert_mp->client_index = em->my_client_index;
 438   cert_mp->context = ntohl (0xfeedface);
 439   cert_mp->cert_len = clib_host_to_net_u16 (test_srv_crt_rsa_len);
 440   clib_memcpy_fast (cert_mp->cert, test_srv_crt_rsa, test_srv_crt_rsa_len);
 441   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cert_mp);
 442
 443   key_mp = vl_msg_api_alloc (sizeof (*key_mp) + test_srv_key_rsa_len);
 444   clib_memset (key_mp, 0, sizeof (*key_mp) + test_srv_key_rsa_len);
 445   key_mp->_vl_msg_id = ntohs (VL_API_APPLICATION_TLS_KEY_ADD);
 446   key_mp->client_index = em->my_client_index;
 447   key_mp->context = ntohl (0xfeedface);
 448   key_mp->key_len = clib_host_to_net_u16 (test_srv_key_rsa_len);
 449   clib_memcpy_fast (key_mp->key, test_srv_key_rsa, test_srv_key_rsa_len);
 450   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & key_mp);
 451 }
 452
 453 static int
 454 application_attach (echo_main_t * em)
 455 {
 456   application_send_attach (em);
 457   return wait_for_state_change (em, STATE_ATTACHED, TIMEOUT);
 458 }
 459
 460 void
 461 application_detach (echo_main_t * em)
 462 {
 463   vl_api_application_detach_t *bmp;
 464   bmp = vl_msg_api_alloc (sizeof (*bmp));
 465   clib_memset (bmp, 0, sizeof (*bmp));
 466
 467   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH);
 468   bmp->client_index = em->my_client_index;
 469   bmp->context = ntohl (0xfeedface);
 470   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 471
 472   DBG ("%s", "Sent detach");
 473 }
 474
 475 static int
 476 ssvm_segment_attach (char *name, ssvm_segment_type_t type, int fd)
 477 {
 478   fifo_segment_create_args_t _a, *a = &_a;
 479   fifo_segment_main_t *sm = &echo_main.segment_main;
 480   int rv;
 481
 482   clib_memset (a, 0, sizeof (*a));
 483   a->segment_name = (char *) name;
 484   a->segment_type = type;
 485
 486   if (type == SSVM_SEGMENT_MEMFD)
 487     a->memfd_fd = fd;
 488
 489   if ((rv = fifo_segment_attach (sm, a)))
 490     {
 491       clib_warning ("svm_fifo_segment_attach ('%s') failed", name);
 492       return rv;
 493     }
 494   vec_reset_length (a->new_segment_indices);
 495   return 0;
 496 }
 497
 498 static void
 499 vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
 500                                            mp)
 501 {
 502   echo_main_t *em = &echo_main;
 503   int *fds = 0;
 504   u32 n_fds = 0;
 505   u64 segment_handle;
 506   segment_handle = clib_net_to_host_u64 (mp->segment_handle);
 507   DBG ("Attached returned app %u", htons (mp->app_index));
 508
 509   if (mp->retval)
 510     {
 511       clib_warning ("attach failed: %U", format_api_error,
 512                     clib_net_to_host_u32 (mp->retval));
 513       goto failed;
 514     }
 515
 516   if (mp->segment_name_length == 0)
 517     {
 518       clib_warning ("segment_name_length zero");
 519       goto failed;
 520     }
 521
 522   ASSERT (mp->app_event_queue_address);
 523   em->our_event_queue = uword_to_pointer (mp->app_event_queue_address,
 524                                           svm_msg_q_t *);
 525
 526   if (mp->n_fds)
 527     {
 528       vec_validate (fds, mp->n_fds);
 529       vl_socket_client_recv_fd_msg (fds, mp->n_fds, 5);
 530
 531       if (mp->fd_flags & SESSION_FD_F_VPP_MQ_SEGMENT)
 532         if (ssvm_segment_attach (0, SSVM_SEGMENT_MEMFD, fds[n_fds++]))
 533           goto failed;
 534
 535       if (mp->fd_flags & SESSION_FD_F_MEMFD_SEGMENT)
 536         if (ssvm_segment_attach ((char *) mp->segment_name,
 537                                  SSVM_SEGMENT_MEMFD, fds[n_fds++]))
 538           goto failed;
 539
 540       if (mp->fd_flags & SESSION_FD_F_MQ_EVENTFD)
 541         svm_msg_q_set_consumer_eventfd (em->our_event_queue, fds[n_fds++]);
 542
 543       vec_free (fds);
 544     }
 545   else
 546     {
 547       if (ssvm_segment_attach ((char *) mp->segment_name, SSVM_SEGMENT_SHM,
 548                                -1))
 549         goto failed;
 550     }
 551   clib_spinlock_lock (&em->segment_handles_lock);
 552   hash_set (em->shared_segment_handles, segment_handle, 1);
 553   clib_spinlock_unlock (&em->segment_handles_lock);
 554   DBG ("Mapped new segment %lx", segment_handle);
 555
 556   em->state = STATE_ATTACHED;
 557   return;
 558 failed:
 559   em->state = STATE_FAILED;
 560   return;
 561 }
 562
 563 static void
 564 vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t *
 565                                            mp)
 566 {
 567   if (mp->retval)
 568     clib_warning ("detach returned with err: %d", mp->retval);
 569   echo_main.state = STATE_DETACHED;
 570 }
 571
 572 static void
 573 stop_signal (int signum)
 574 {
 575   echo_main_t *um = &echo_main;
 576   um->time_to_stop = 1;
 577 }
 578
 579 static clib_error_t *
 580 setup_signal_handlers (void)
 581 {
 582   signal (SIGINT, stop_signal);
 583   signal (SIGQUIT, stop_signal);
 584   signal (SIGTERM, stop_signal);
 585   return 0;
 586 }
 587
 588 int
 589 connect_to_vpp (char *name)
 590 {
 591   echo_main_t *em = &echo_main;
 592   api_main_t *am = &api_main;
 593
 594   if (em->use_sock_api)
 595     {
 596       if (vl_socket_client_connect ((char *) em->socket_name, name,
 597                                     0 /* default rx, tx buffer */ ))
 598         {
 599           clib_warning ("socket connect failed");
 600           return -1;
 601         }
 602
 603       if (vl_socket_client_init_shm (0, 1 /* want_pthread */ ))
 604         {
 605           clib_warning ("init shm api failed");
 606           return -1;
 607         }
 608     }
 609   else
 610     {
 611       if (vl_client_connect_to_vlib ("/vpe-api", name, 32) < 0)
 612         {
 613           clib_warning ("shmem connect failed");
 614           return -1;
 615         }
 616     }
 617   em->vl_input_queue = am->shmem_hdr->vl_input_queue;
 618   em->my_client_index = am->my_client_index;
 619   return 0;
 620 }
 621
 622 void
 623 disconnect_from_vpp (echo_main_t * em)
 624 {
 625   if (em->use_sock_api)
 626     vl_socket_client_disconnect ();
 627   else
 628     vl_client_disconnect_from_vlib ();
 629 }
 630
 631 static void
 632 vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
 633 {
 634   fifo_segment_main_t *sm = &echo_main.segment_main;
 635   fifo_segment_create_args_t _a, *a = &_a;
 636   echo_main_t *em = &echo_main;
 637   int rv;
 638   int *fds = 0;
 639   u64 segment_handle;
 640   segment_handle = clib_net_to_host_u64 (mp->segment_handle);
 641
 642   if (mp->fd_flags & SESSION_FD_F_MEMFD_SEGMENT)
 643     {
 644       vec_validate (fds, 1);
 645       vl_socket_client_recv_fd_msg (fds, 1, 5);
 646       if (ssvm_segment_attach
 647           ((char *) mp->segment_name, SSVM_SEGMENT_MEMFD, fds[0]))
 648         clib_warning
 649           ("svm_fifo_segment_attach ('%s') failed on SSVM_SEGMENT_MEMFD",
 650            mp->segment_name);
 651       clib_spinlock_lock (&em->segment_handles_lock);
 652       hash_set (em->shared_segment_handles, segment_handle, 1);
 653       clib_spinlock_unlock (&em->segment_handles_lock);
 654       vec_free (fds);
 655       DBG ("Mapped new segment %lx", segment_handle);
 656       return;
 657     }
 658
 659   clib_memset (a, 0, sizeof (*a));
 660   a->segment_name = (char *) mp->segment_name;
 661   a->segment_size = mp->segment_size;
 662   /* Attach to the segment vpp created */
 663   rv = fifo_segment_attach (sm, a);
 664   if (rv)
 665     {
 666       clib_warning ("svm_fifo_segment_attach ('%s') failed",
 667                     mp->segment_name);
 668       return;
 669     }
 670   clib_spinlock_lock (&em->segment_handles_lock);
 671   hash_set (em->shared_segment_handles, mp->segment_name, 1);
 672   clib_spinlock_unlock (&em->segment_handles_lock);
 673   clib_warning ("Mapped new segment '%s' size %d", mp->segment_name,
 674                 mp->segment_size);
 675 }
 676
 677 static void
 678 session_print_stats (echo_main_t * em, echo_session_t * session)
 679 {
 680   f64 deltat = clib_time_now (&em->clib_time) - session->start;
 681   fformat (stdout, "Session %x done in %.6fs RX[%.4f] TX[%.4f] Gbit/s\n",
 682            session->session_index, deltat,
 683            (session->bytes_received * 8.0) / deltat / 1e9,
 684            (session->bytes_sent * 8.0) / deltat / 1e9);
 685 }
 686
 687 static void
 688 print_global_stats (echo_main_t * em)
 689 {
 690   f64 deltat = em->end_time - em->start_time;
 691   u8 *s = format (0, "%U:%U",
 692                   echo_format_timing_event, em->timing_start_event,
 693                   echo_format_timing_event, em->timing_end_event);
 694   fformat (stdout, "Timinig %s\n", s);
 695   fformat (stdout, "-------- TX --------\n");
 696   fformat (stdout, "%lld bytes (%lld mbytes, %lld gbytes) in %.2f seconds\n",
 697            em->tx_total, em->tx_total / (1ULL << 20),
 698            em->tx_total / (1ULL << 30), deltat);
 699   fformat (stdout, "%.4f Gbit/second\n", (em->tx_total * 8.0) / deltat / 1e9);
 700   fformat (stdout, "-------- RX --------\n");
 701   fformat (stdout, "%lld bytes (%lld mbytes, %lld gbytes) in %.2f seconds\n",
 702            em->rx_total, em->rx_total / (1ULL << 20),
 703            em->rx_total / (1ULL << 30), deltat);
 704   fformat (stdout, "%.4f Gbit/second\n", (em->rx_total * 8.0) / deltat / 1e9);
 705   fformat (stdout, "--------------------\n");
 706 }
 707
 708
 709 static void
 710 test_recv_bytes (echo_main_t * em, echo_session_t * s, u8 * rx_buf,
 711                  u32 n_read)
 712 {
 713   int i;
 714   u8 expected;
 715   for (i = 0; i < n_read; i++)
 716     {
 717       expected = (s->bytes_received + i) & 0xff;
 718       if (rx_buf[i] != expected && em->max_test_msg > 0)
 719         {
 720           clib_warning
 721             ("Session[%lx][0x%lx] byte[%lld], got 0x%x but expected 0x%x",
 722              s->session_index, s->vpp_session_handle, s->bytes_received + i,
 723              rx_buf[i], expected);
 724           em->max_test_msg--;
 725           if (em->max_test_msg == 0)
 726             clib_warning ("Too many errors, hiding next ones");
 727           if (em->test_return_packets == RETURN_PACKETS_ASSERT)
 728             ASSERT (0);
 729         }
 730     }
 731 }
 732
 733 static void
 734 recv_data_chunk (echo_main_t * em, echo_session_t * s, u8 * rx_buf)
 735 {
 736   int n_to_read, n_read;
 737
 738   n_to_read = svm_fifo_max_dequeue (s->rx_fifo);
 739   if (!n_to_read)
 740     return;
 741
 742   do
 743     {
 744       n_read =
 745         app_recv_stream ((app_session_t *) s, rx_buf, vec_len (rx_buf));
 746       if (n_read <= 0)
 747         break;
 748       notify_rx_data_to_vpp (s);
 749       if (em->test_return_packets)
 750         test_recv_bytes (em, s, rx_buf, n_read);
 751
 752       ASSERT (s->bytes_to_receive >= n_read);
 753       n_to_read -= n_read;
 754       s->bytes_received += n_read;
 755       s->bytes_to_receive -= n_read;
 756     }
 757   while (n_to_read > 0);
 758 }
 759
 760 static void
 761 send_data_chunk (echo_main_t * em, echo_session_t * s)
 762 {
 763   u64 test_buf_len, bytes_this_chunk, test_buf_offset;
 764   u8 *test_data = em->connect_test_data;
 765   int n_sent;
 766
 767   test_buf_len = vec_len (test_data);
 768   test_buf_offset = s->bytes_sent % test_buf_len;
 769   bytes_this_chunk = clib_min (test_buf_len - test_buf_offset,
 770                                s->bytes_to_send);
 771
 772   n_sent = app_send_stream ((app_session_t *) s, test_data + test_buf_offset,
 773                             bytes_this_chunk, 0);
 774
 775   if (n_sent > 0)
 776     {
 777       s->bytes_to_send -= n_sent;
 778       s->bytes_sent += n_sent;
 779     }
 780 }
 781
 782 /*
 783  * Rx/Tx polling thread per connection
 784  */
 785 static void *
 786 client_thread_fn (void *arg)
 787 {
 788   echo_main_t *em = &echo_main;
 789   u8 *rx_buf = 0;
 790   u32 session_index = *(u32 *) arg;
 791   echo_session_t *s;
 792
 793   vec_validate (rx_buf, 1 << 20);
 794
 795   while (!em->time_to_stop && em->state != STATE_READY)
 796     ;
 797
 798   s = pool_elt_at_index (em->sessions, session_index);
 799   while (!em->time_to_stop)
 800     {
 801       send_data_chunk (em, s);
 802       recv_data_chunk (em, s, rx_buf);
 803       if (!s->bytes_to_send && !s->bytes_to_receive)
 804         break;
 805     }
 806
 807   DBG ("[%lu/%lu] -> S(%x) -> [%lu/%lu]",
 808        s->bytes_received, s->bytes_received + s->bytes_to_receive,
 809        session_index, s->bytes_sent, s->bytes_sent + s->bytes_to_send);
 810   em->tx_total += s->bytes_sent;
 811   em->rx_total += s->bytes_received;
 812   em->n_clients_connected--;
 813
 814   if (em->n_clients_connected == 0)
 815     quic_echo_notify_event (em, ECHO_EVT_LAST_BYTE);
 816
 817   pthread_exit (0);
 818 }
 819
 820 static void
 821 echo_send_connect (echo_main_t * em, u8 * uri, u32 opaque)
 822 {
 823   vl_api_connect_uri_t *cmp;
 824   cmp = vl_msg_api_alloc (sizeof (*cmp));
 825   clib_memset (cmp, 0, sizeof (*cmp));
 826
 827   cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
 828   cmp->client_index = em->my_client_index;
 829   cmp->context = ntohl (opaque);
 830   memcpy (cmp->uri, uri, vec_len (uri));
 831   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cmp);
 832 }
 833
 834 static void
 835 client_disconnect_session (echo_main_t * em, echo_session_t * s)
 836 {
 837   vl_api_disconnect_session_t *dmp;
 838   dmp = vl_msg_api_alloc (sizeof (*dmp));
 839   clib_memset (dmp, 0, sizeof (*dmp));
 840   dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
 841   dmp->client_index = em->my_client_index;
 842   dmp->handle = s->vpp_session_handle;
 843   DBG ("Sending Session disonnect handle %lu", dmp->handle);
 844   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & dmp);
 845   pool_put (em->sessions, s);
 846   clib_memset (s, 0xfe, sizeof (*s));
 847 }
 848
 849 static void
 850 session_bound_handler (session_bound_msg_t * mp)
 851 {
 852   echo_main_t *em = &echo_main;
 853
 854   if (mp->retval)
 855     {
 856       clib_warning ("bind failed: %U", format_api_error,
 857                     clib_net_to_host_u32 (mp->retval));
 858       em->state = STATE_FAILED;
 859       return;
 860     }
 861
 862   clib_warning ("listening on %U:%u", format_ip46_address, mp->lcl_ip,
 863                 mp->lcl_is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6,
 864                 clib_net_to_host_u16 (mp->lcl_port));
 865   em->listener_handle = mp->handle;
 866   em->state = STATE_LISTEN;
 867 }
 868
 869 static void
 870 session_accepted_handler (session_accepted_msg_t * mp)
 871 {
 872   app_session_evt_t _app_evt, *app_evt = &_app_evt;
 873   session_accepted_reply_msg_t *rmp;
 874   svm_fifo_t *rx_fifo, *tx_fifo;
 875   echo_main_t *em = &echo_main;
 876   echo_session_t *session;
 877   u32 session_index;
 878
 879   /* Allocate local session and set it up */
 880   pool_get (em->sessions, session);
 881   session_index = session - em->sessions;
 882
 883   if (wait_for_segment_allocation (mp->segment_handle))
 884     return;
 885
 886   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 887   rx_fifo->client_session_index = session_index;
 888   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 889   tx_fifo->client_session_index = session_index;
 890
 891   session->rx_fifo = rx_fifo;
 892   session->tx_fifo = tx_fifo;
 893   session->session_index = session_index;
 894   session->vpp_session_handle = mp->handle;
 895   session->start = clib_time_now (&em->clib_time);
 896   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
 897                                          svm_msg_q_t *);
 898
 899   /* Add it to lookup table */
 900   DBG ("Accepted session handle %lx, Listener %lx idx %lu", mp->handle,
 901        mp->listener_handle, session_index);
 902   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 903
 904   app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
 905                              SESSION_CTRL_EVT_ACCEPTED_REPLY);
 906   rmp = (session_accepted_reply_msg_t *) app_evt->evt->data;
 907   rmp->handle = mp->handle;
 908   rmp->context = mp->context;
 909   app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
 910
 911   DBG ("SSession handle is %lu", mp->handle);
 912   if (mp->listener_handle == em->listener_handle)
 913     {
 914       if (em->cb_vft.quic_accepted_cb)
 915         em->cb_vft.quic_accepted_cb (mp, session_index);
 916       em->n_quic_clients_connected++;
 917     }
 918   else if (em->i_am_master)
 919     {
 920       if (em->cb_vft.server_stream_accepted_cb)
 921         em->cb_vft.server_stream_accepted_cb (mp, session_index);
 922       em->n_clients_connected++;
 923     }
 924   else
 925     {
 926       if (em->cb_vft.client_stream_accepted_cb)
 927         em->cb_vft.client_stream_accepted_cb (mp, session_index);
 928       em->n_clients_connected++;
 929     }
 930
 931   if (em->n_clients_connected == em->n_clients * em->n_stream_clients)
 932     {
 933       em->state = STATE_READY;
 934       quic_echo_notify_event (em, ECHO_EVT_LAST_SCONNECTED);
 935     }
 936   if (em->n_quic_clients_connected == em->n_clients)
 937     quic_echo_notify_event (em, ECHO_EVT_LAST_QCONNECTED);
 938 }
 939
 940 static void
 941 session_connected_handler (session_connected_msg_t * mp)
 942 {
 943   echo_main_t *em = &echo_main;
 944   echo_session_t *session;
 945   u32 session_index;
 946   svm_fifo_t *rx_fifo, *tx_fifo;
 947
 948   if (mp->retval)
 949     {
 950       clib_warning ("connection failed with code: %U", format_api_error,
 951                     clib_net_to_host_u32 (mp->retval));
 952       em->state = STATE_FAILED;
 953       return;
 954     }
 955
 956   pool_get (em->sessions, session);
 957   clib_memset (session, 0, sizeof (*session));
 958   session_index = session - em->sessions;
 959   DBG ("CONNECTED session[%lx][0x%lx]", session_index, mp->handle);
 960
 961   if (wait_for_segment_allocation (mp->segment_handle))
 962     return;
 963
 964   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
 965   rx_fifo->client_session_index = session_index;
 966   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 967   tx_fifo->client_session_index = session_index;
 968
 969   session->rx_fifo = rx_fifo;
 970   session->tx_fifo = tx_fifo;
 971   session->vpp_session_handle = mp->handle;
 972   session->session_index = session_index;
 973   session->start = clib_time_now (&em->clib_time);
 974   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
 975                                          svm_msg_q_t *);
 976
 977   DBG ("Connected session handle %lx, idx %lu RX[%lx] TX[%lx]", mp->handle,
 978        session_index, rx_fifo, tx_fifo);
 979   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 980
 981   if (mp->context == QUIC_SESSION_TYPE_QUIC)
 982     {
 983       if (em->cb_vft.quic_connected_cb)
 984         em->cb_vft.quic_connected_cb (mp, session_index);
 985       em->n_quic_clients_connected++;
 986     }
 987   else if (em->i_am_master)
 988     {
 989       if (em->cb_vft.server_stream_connected_cb)
 990         em->cb_vft.server_stream_connected_cb (mp, session_index);
 991       em->n_clients_connected++;
 992     }
 993   else
 994     {
 995       if (em->cb_vft.client_stream_connected_cb)
 996         em->cb_vft.client_stream_connected_cb (mp, session_index);
 997       em->n_clients_connected++;
 998     }
 999
1000   if (em->n_clients_connected == em->n_clients * em->n_stream_clients)
1001     {
1002       em->state = STATE_READY;
1003       quic_echo_notify_event (em, ECHO_EVT_LAST_SCONNECTED);
1004     }
1005   if (em->n_quic_clients_connected == em->n_clients)
1006     quic_echo_notify_event (em, ECHO_EVT_LAST_QCONNECTED);
1007 }
1008
1009 /*
1010  *
1011  *  ECHO Callback definitions
1012  *
1013  */
1014
1015
1016 static void
1017 echo_on_connected_connect (session_connected_msg_t * mp, u32 session_index)
1018 {
1019   echo_main_t *em = &echo_main;
1020   u8 *uri = format (0, "QUIC://session/%lu", mp->handle);
1021   int i;
1022
1023   if (!em->first_sconnect_sent)
1024     {
1025       em->first_sconnect_sent = 1;
1026       quic_echo_notify_event (em, ECHO_EVT_FIRST_SCONNECT);
1027     }
1028   for (i = 0; i < em->n_stream_clients; i++)
1029     {
1030       DBG ("CONNECT : new QUIC stream #%d: %s", i, uri);
1031       echo_send_connect (em, uri, QUIC_SESSION_TYPE_STREAM);
1032     }
1033
1034   clib_warning ("session %u (0x%llx) connected with local ip %U port %d",
1035                 session_index, mp->handle, format_ip46_address, &mp->lcl.ip,
1036                 mp->lcl.is_ip4, clib_net_to_host_u16 (mp->lcl.port));
1037 }
1038
1039 static void
1040 echo_on_connected_send (session_connected_msg_t * mp, u32 session_index)
1041 {
1042   echo_main_t *em = &echo_main;
1043   int rv;
1044   echo_session_t *session;
1045
1046   DBG ("Stream Session Connected");
1047
1048   session = pool_elt_at_index (em->sessions, session_index);
1049   session->bytes_to_send = em->bytes_to_send;
1050   session->bytes_to_receive = em->bytes_to_receive;
1051
1052   /*
1053    * Start RX thread
1054    */
1055   em->thread_args[em->n_clients_connected] = session_index;
1056   rv = pthread_create (&em->client_thread_handles[em->n_clients_connected],
1057                        NULL /*attr */ , client_thread_fn,
1058                        (void *) &em->thread_args[em->n_clients_connected]);
1059   if (rv)
1060     {
1061       clib_warning ("pthread_create returned %d", rv);
1062       return;
1063     }
1064 }
1065
1066 static void
1067 echo_on_connected_error (session_connected_msg_t * mp, u32 session_index)
1068 {
1069   clib_warning ("Got a wrong connected on session %u [%lx]", session_index,
1070                 mp->handle);
1071 }
1072
1073 static void
1074 echo_on_accept_recv (session_accepted_msg_t * mp, u32 session_index)
1075 {
1076   echo_main_t *em = &echo_main;
1077   int rv;
1078   echo_session_t *session;
1079
1080   session = pool_elt_at_index (em->sessions, session_index);
1081   session->bytes_to_send = em->bytes_to_send;
1082   session->bytes_to_receive = em->bytes_to_receive;
1083
1084   DBG ("Stream session accepted 0x%lx, expecting %lu bytes",
1085        session->vpp_session_handle, session->bytes_to_receive);
1086
1087   /*
1088    * Start RX thread
1089    */
1090   em->thread_args[em->n_clients_connected] = session_index;
1091   rv = pthread_create (&em->client_thread_handles[em->n_clients_connected],
1092                        NULL /*attr */ , client_thread_fn,
1093                        (void *) &em->thread_args[em->n_clients_connected]);
1094   if (rv)
1095     {
1096       clib_warning ("pthread_create returned %d", rv);
1097       return;
1098     }
1099
1100 }
1101
1102 static void
1103 echo_on_accept_connect (session_accepted_msg_t * mp, u32 session_index)
1104 {
1105   echo_main_t *em = &echo_main;
1106   DBG ("Accept on QSession index %u", mp->handle);
1107   u8 *uri = format (0, "QUIC://session/%lu", mp->handle);
1108   u32 i;
1109
1110   if (!em->first_sconnect_sent)
1111     {
1112       em->first_sconnect_sent = 1;
1113       quic_echo_notify_event (em, ECHO_EVT_FIRST_SCONNECT);
1114     }
1115   for (i = 0; i < em->n_stream_clients; i++)
1116     {
1117       DBG ("ACCEPT : new QUIC stream #%d: %s", i, uri);
1118       echo_send_connect (em, uri, QUIC_SESSION_TYPE_STREAM);
1119     }
1120 }
1121
1122 static void
1123 echo_on_accept_error (session_accepted_msg_t * mp, u32 session_index)
1124 {
1125   clib_warning ("Got a wrong accept on session %u [%lx]", session_index,
1126                 mp->handle);
1127 }
1128
1129 static void
1130 echo_on_accept_log_ip (session_accepted_msg_t * mp, u32 session_index)
1131 {
1132   u8 *ip_str;
1133   ip_str = format (0, "%U", format_ip46_address, &mp->rmt.ip, mp->rmt.is_ip4);
1134   clib_warning ("Accepted session from: %s:%d", ip_str,
1135                 clib_net_to_host_u16 (mp->rmt.port));
1136
1137 }
1138
1139 static const quic_echo_cb_vft_t default_cb_vft = {
1140   /* Qsessions */
1141   .quic_accepted_cb = &echo_on_accept_log_ip,
1142   .quic_connected_cb = &echo_on_connected_connect,
1143   /* client initiated streams */
1144   .server_stream_accepted_cb = NULL,
1145   .client_stream_connected_cb = &echo_on_connected_send,
1146   /* server initiated streams */
1147   .client_stream_accepted_cb = &echo_on_accept_error,
1148   .server_stream_connected_cb = &echo_on_connected_error,
1149 };
1150
1151 static const quic_echo_cb_vft_t server_stream_cb_vft = {
1152   /* Qsessions */
1153   .quic_accepted_cb = &echo_on_accept_connect,
1154   .quic_connected_cb = NULL,
1155   /* client initiated streams */
1156   .server_stream_accepted_cb = &echo_on_accept_error,
1157   .client_stream_connected_cb = &echo_on_connected_error,
1158   /* server initiated streams */
1159   .client_stream_accepted_cb = &echo_on_accept_recv,
1160   .server_stream_connected_cb = &echo_on_connected_send,
1161 };
1162
1163 static uword
1164 echo_unformat_quic_setup_vft (unformat_input_t * input, va_list * args)
1165 {
1166   echo_main_t *em = &echo_main;
1167   if (unformat (input, "serverstream"))
1168     {
1169       clib_warning ("Using QUIC server initiated streams");
1170       em->no_return = 1;
1171       em->cb_vft = server_stream_cb_vft;
1172       return 1;
1173     }
1174   else if (unformat (input, "default"))
1175     return 1;
1176   return 0;
1177 }
1178
1179 static uword
1180 echo_unformat_data (unformat_input_t * input, va_list * args)
1181 {
1182   u64 _a;
1183   u64 *a = va_arg (*args, u64 *);
1184   if (unformat (input, "%lluGb", &_a))
1185     {
1186       *a = _a << 30;
1187       return 1;
1188     }
1189   else if (unformat (input, "%lluMb", &_a))
1190     {
1191       *a = _a << 20;
1192       return 1;
1193     }
1194   else if (unformat (input, "%lluKb", &_a))
1195     {
1196       *a = _a << 10;
1197       return 1;
1198     }
1199   else if (unformat (input, "%llu", a))
1200     return 1;
1201   return 0;
1202 }
1203
1204 /*
1205  *
1206  *  End of ECHO callback definitions
1207  *
1208  */
1209
1210 static void
1211 session_disconnected_handler (session_disconnected_msg_t * mp)
1212 {
1213   app_session_evt_t _app_evt, *app_evt = &_app_evt;
1214   session_disconnected_reply_msg_t *rmp;
1215   echo_main_t *em = &echo_main;
1216   echo_session_t *session = 0;
1217   uword *p;
1218   int rv = 0;
1219   DBG ("Disonnected session handle %lx", mp->handle);
1220   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
1221   if (!p)
1222     {
1223       clib_warning ("couldn't find session key %llx", mp->handle);
1224       return;
1225     }
1226
1227   session = pool_elt_at_index (em->sessions, p[0]);
1228   hash_unset (em->session_index_by_vpp_handles, mp->handle);
1229
1230   pool_put (em->sessions, session);
1231
1232   app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
1233                              SESSION_CTRL_EVT_DISCONNECTED_REPLY);
1234   rmp = (session_disconnected_reply_msg_t *) app_evt->evt->data;
1235   rmp->retval = rv;
1236   rmp->handle = mp->handle;
1237   rmp->context = mp->context;
1238   app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
1239
1240   session_print_stats (em, session);
1241 }
1242
1243 static void
1244 session_reset_handler (session_reset_msg_t * mp)
1245 {
1246   app_session_evt_t _app_evt, *app_evt = &_app_evt;
1247   echo_main_t *em = &echo_main;
1248   session_reset_reply_msg_t *rmp;
1249   echo_session_t *session = 0;
1250   uword *p;
1251   int rv = 0;
1252
1253   DBG ("Reset session handle %lx", mp->handle);
1254   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
1255
1256   if (p)
1257     {
1258       session = pool_elt_at_index (em->sessions, p[0]);
1259       clib_warning ("got reset");
1260       /* Cleanup later */
1261       em->time_to_stop = 1;
1262     }
1263   else
1264     {
1265       clib_warning ("couldn't find session key %llx", mp->handle);
1266       return;
1267     }
1268
1269   app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
1270                              SESSION_CTRL_EVT_RESET_REPLY);
1271   rmp = (session_reset_reply_msg_t *) app_evt->evt->data;
1272   rmp->retval = rv;
1273   rmp->handle = mp->handle;
1274   app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
1275 }
1276
1277 static void
1278 handle_mq_event (echo_main_t * em, session_event_t * e, int handle_rx)
1279 {
1280   switch (e->event_type)
1281     {
1282     case SESSION_CTRL_EVT_BOUND:
1283       DBG ("SESSION_CTRL_EVT_BOUND");
1284       session_bound_handler ((session_bound_msg_t *) e->data);
1285       break;
1286     case SESSION_CTRL_EVT_ACCEPTED:
1287       DBG ("SESSION_CTRL_EVT_ACCEPTED");
1288       session_accepted_handler ((session_accepted_msg_t *) e->data);
1289       break;
1290     case SESSION_CTRL_EVT_CONNECTED:
1291       DBG ("SESSION_CTRL_EVT_CONNECTED");
1292       session_connected_handler ((session_connected_msg_t *) e->data);
1293       break;
1294     case SESSION_CTRL_EVT_DISCONNECTED:
1295       DBG ("SESSION_CTRL_EVT_DISCONNECTED");
1296       session_disconnected_handler ((session_disconnected_msg_t *) e->data);
1297       break;
1298     case SESSION_CTRL_EVT_RESET:
1299       DBG ("SESSION_CTRL_EVT_RESET");
1300       session_reset_handler ((session_reset_msg_t *) e->data);
1301       break;
1302     case SESSION_IO_EVT_RX:
1303       DBG ("SESSION_IO_EVT_RX");
1304       if (handle_rx)
1305         echo_handle_rx (em, e);
1306       break;
1307     default:
1308       clib_warning ("unhandled event %u", e->event_type);
1309     }
1310 }
1311
1312 static int
1313 clients_run (echo_main_t * em)
1314 {
1315   svm_msg_q_msg_t msg;
1316   session_event_t *e;
1317   echo_session_t *s;
1318   hash_pair_t *p;
1319   int i;
1320
1321   /*
1322    * Attach and connect the clients
1323    */
1324   if (application_attach (em))
1325     return -1;
1326
1327   quic_echo_notify_event (em, ECHO_EVT_FIRST_QCONNECT);
1328   for (i = 0; i < em->n_clients; i++)
1329     echo_send_connect (em, em->uri, QUIC_SESSION_TYPE_QUIC);
1330
1331   wait_for_state_change (em, STATE_READY, TIMEOUT);
1332
1333   /*
1334    * Wait for client threads to send the data
1335    */
1336   DBG ("Waiting for data on %u clients", em->n_clients_connected);
1337   while (em->n_clients_connected)
1338     {
1339       if (svm_msg_q_is_empty (em->our_event_queue))
1340         continue;
1341       if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_TIMEDWAIT, 1))
1342         continue;
1343       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
1344       handle_mq_event (em, e, 0 /* handle_rx */ );
1345       svm_msg_q_free_msg (em->our_event_queue, &msg);
1346     }
1347
1348   /* *INDENT-OFF* */
1349   hash_foreach_pair (p, em->session_index_by_vpp_handles,
1350     ({
1351       s = pool_elt_at_index (em->sessions, p->value[0]);
1352       DBG ("Sending disconnect on session %lu", p->key);
1353       client_disconnect_session (em, s);
1354     }));
1355   /* *INDENT-ON* */
1356
1357   wait_for_disconnected_sessions (em);
1358   application_detach (em);
1359   return 0;
1360 }
1361
1362 static void
1363 vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
1364 {
1365   echo_main_t *em = &echo_main;
1366   if (mp->retval)
1367     {
1368       clib_warning ("bind failed: %U", format_api_error,
1369                     clib_net_to_host_u32 (mp->retval));
1370       em->state = STATE_FAILED;
1371       return;
1372     }
1373
1374   em->state = STATE_LISTEN;
1375 }
1376
1377 static void
1378 vl_api_unbind_uri_reply_t_handler (vl_api_unbind_uri_reply_t * mp)
1379 {
1380   echo_main_t *em = &echo_main;
1381
1382   if (mp->retval != 0)
1383     clib_warning ("returned %d", ntohl (mp->retval));
1384
1385   em->state = STATE_START;
1386 }
1387
1388 u8 *
1389 format_ip4_address (u8 * s, va_list * args)
1390 {
1391   u8 *a = va_arg (*args, u8 *);
1392   return format (s, "%d.%d.%d.%d", a[0], a[1], a[2], a[3]);
1393 }
1394
1395 u8 *
1396 format_ip6_address (u8 * s, va_list * args)
1397 {
1398   ip6_address_t *a = va_arg (*args, ip6_address_t *);
1399   u32 i, i_max_n_zero, max_n_zeros, i_first_zero, n_zeros, last_double_colon;
1400
1401   i_max_n_zero = ARRAY_LEN (a->as_u16);
1402   max_n_zeros = 0;
1403   i_first_zero = i_max_n_zero;
1404   n_zeros = 0;
1405   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
1406     {
1407       u32 is_zero = a->as_u16[i] == 0;
1408       if (is_zero && i_first_zero >= ARRAY_LEN (a->as_u16))
1409         {
1410           i_first_zero = i;
1411           n_zeros = 0;
1412         }
1413       n_zeros += is_zero;
1414       if ((!is_zero && n_zeros > max_n_zeros)
1415           || (i + 1 >= ARRAY_LEN (a->as_u16) && n_zeros > max_n_zeros))
1416         {
1417           i_max_n_zero = i_first_zero;
1418           max_n_zeros = n_zeros;
1419           i_first_zero = ARRAY_LEN (a->as_u16);
1420           n_zeros = 0;
1421         }
1422     }
1423
1424   last_double_colon = 0;
1425   for (i = 0; i < ARRAY_LEN (a->as_u16); i++)
1426     {
1427       if (i == i_max_n_zero && max_n_zeros > 1)
1428         {
1429           s = format (s, "::");
1430           i += max_n_zeros - 1;
1431           last_double_colon = 1;
1432         }
1433       else
1434         {
1435           s = format (s, "%s%x",
1436                       (last_double_colon || i == 0) ? "" : ":",
1437                       clib_net_to_host_u16 (a->as_u16[i]));
1438           last_double_colon = 0;
1439         }
1440     }
1441
1442   return s;
1443 }
1444
1445 /* Format an IP46 address. */
1446 u8 *
1447 format_ip46_address (u8 * s, va_list * args)
1448 {
1449   ip46_address_t *ip46 = va_arg (*args, ip46_address_t *);
1450   ip46_type_t type = va_arg (*args, ip46_type_t);
1451   int is_ip4 = 1;
1452
1453   switch (type)
1454     {
1455     case IP46_TYPE_ANY:
1456       is_ip4 = ip46_address_is_ip4 (ip46);
1457       break;
1458     case IP46_TYPE_IP4:
1459       is_ip4 = 1;
1460       break;
1461     case IP46_TYPE_IP6:
1462       is_ip4 = 0;
1463       break;
1464     }
1465
1466   return is_ip4 ?
1467     format (s, "%U", format_ip4_address, &ip46->ip4) :
1468     format (s, "%U", format_ip6_address, &ip46->ip6);
1469 }
1470
1471 static void
1472 echo_handle_rx (echo_main_t * em, session_event_t * e)
1473 {
1474   int n_read, max_dequeue, n_sent;
1475   u32 offset, to_dequeue;
1476   echo_session_t *s;
1477   s = pool_elt_at_index (em->sessions, e->session_index);
1478
1479   /* Clear event only once. Otherwise, if we do it in the loop by calling
1480    * app_recv_stream, we may end up with a lot of unhandled rx events on the
1481    * message queue */
1482   svm_fifo_unset_event (s->rx_fifo);
1483   max_dequeue = svm_fifo_max_dequeue (s->rx_fifo);
1484   if (PREDICT_FALSE (!max_dequeue))
1485     return;
1486   do
1487     {
1488       /* The options here are to limit ourselves to max_dequeue or read
1489        * even the data that was enqueued while we were dequeueing and which
1490        * now has an rx event in the mq. Either of the two work. */
1491       to_dequeue = clib_min (max_dequeue, vec_len (em->rx_buf));
1492       n_read = app_recv_stream_raw (s->rx_fifo, em->rx_buf, to_dequeue,
1493                                     0 /* clear evt */ , 0 /* peek */ );
1494
1495       if (n_read <= 0)
1496         break;
1497       DBG ("Notify cause %u bytes", n_read);
1498       notify_rx_data_to_vpp (s);
1499       if (em->test_return_packets)
1500         test_recv_bytes (em, s, em->rx_buf, n_read);
1501
1502       max_dequeue -= n_read;
1503       s->bytes_received += n_read;
1504       s->bytes_to_receive -= n_read;
1505
1506       /* Reflect if a non-drop session */
1507       if (!em->no_return)
1508         {
1509           offset = 0;
1510           do
1511             {
1512               n_sent = app_send_stream ((app_session_t *) s,
1513                                         em->rx_buf + offset,
1514                                         n_read, SVM_Q_WAIT);
1515               if (n_sent <= 0)
1516                 continue;
1517               n_read -= n_sent;
1518               s->bytes_to_send -= n_sent;
1519               s->bytes_sent += n_sent;
1520               offset += n_sent;
1521             }
1522           while (n_read > 0);
1523         }
1524     }
1525   while (max_dequeue > 0 && !em->time_to_stop);
1526 }
1527
1528 static void
1529 server_handle_mq (echo_main_t * em)
1530 {
1531   svm_msg_q_msg_t msg;
1532   session_event_t *e;
1533
1534   while (1)
1535     {
1536       int rc = svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_TIMEDWAIT, 1);
1537       if (PREDICT_FALSE (rc == ETIMEDOUT && em->time_to_stop))
1538         break;
1539       if (rc == ETIMEDOUT)
1540         continue;
1541       e = svm_msg_q_msg_data (em->our_event_queue, &msg);
1542       handle_mq_event (em, e, em->state == STATE_READY /* handle_rx */ );
1543       svm_msg_q_free_msg (em->our_event_queue, &msg);
1544     }
1545 }
1546
1547 static void
1548 server_send_listen (echo_main_t * em)
1549 {
1550   vl_api_bind_uri_t *bmp;
1551   bmp = vl_msg_api_alloc (sizeof (*bmp));
1552   clib_memset (bmp, 0, sizeof (*bmp));
1553
1554   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
1555   bmp->client_index = em->my_client_index;
1556   bmp->context = ntohl (0xfeedface);
1557   memcpy (bmp->uri, em->uri, vec_len (em->uri));
1558   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
1559 }
1560
1561 static void
1562 server_send_unbind (echo_main_t * em)
1563 {
1564   vl_api_unbind_uri_t *ump;
1565
1566   ump = vl_msg_api_alloc (sizeof (*ump));
1567   clib_memset (ump, 0, sizeof (*ump));
1568
1569   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
1570   ump->client_index = em->my_client_index;
1571   memcpy (ump->uri, em->uri, vec_len (em->uri));
1572   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & ump);
1573 }
1574
1575 static int
1576 server_run (echo_main_t * em)
1577 {
1578   echo_session_t *session;
1579   int i;
1580
1581   /* $$$$ hack preallocation */
1582   for (i = 0; i < 200000; i++)
1583     {
1584       pool_get (em->sessions, session);
1585       clib_memset (session, 0, sizeof (*session));
1586     }
1587   for (i = 0; i < 200000; i++)
1588     pool_put_index (em->sessions, i);
1589
1590   if (application_attach (em))
1591     return -1;
1592
1593   /* Bind to uri */
1594   server_send_listen (em);
1595   if (wait_for_state_change (em, STATE_READY, 0))
1596     return -2;
1597
1598   /* Enter handle event loop */
1599   server_handle_mq (em);
1600
1601   /* Cleanup */
1602   server_send_unbind (em);
1603   application_detach (em);
1604   fformat (stdout, "Test complete...\n");
1605   return 0;
1606 }
1607
1608 static void
1609 vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t *
1610                                            mp)
1611 {
1612   echo_main_t *em = &echo_main;
1613   uword *p;
1614   DBG ("Got disonnected reply for session handle %lu", mp->handle);
1615   em->state = STATE_START;
1616
1617   p = hash_get (em->session_index_by_vpp_handles, mp->handle);
1618   if (p)
1619     hash_unset (em->session_index_by_vpp_handles, mp->handle);
1620   else
1621     clib_warning ("couldn't find session key %llx", mp->handle);
1622
1623   if (mp->retval)
1624     clib_warning ("vpp complained about disconnect: %d", ntohl (mp->retval));
1625 }
1626
1627 static void
1628   vl_api_application_tls_cert_add_reply_t_handler
1629   (vl_api_application_tls_cert_add_reply_t * mp)
1630 {
1631   if (mp->retval)
1632     clib_warning ("failed to add tls cert");
1633 }
1634
1635 static void
1636   vl_api_application_tls_key_add_reply_t_handler
1637   (vl_api_application_tls_key_add_reply_t * mp)
1638 {
1639   if (mp->retval)
1640     clib_warning ("failed to add tls key");
1641 }
1642
1643 #define foreach_quic_echo_msg                                           \
1644 _(BIND_URI_REPLY, bind_uri_reply)                                       \
1645 _(UNBIND_URI_REPLY, unbind_uri_reply)                                   \
1646 _(DISCONNECT_SESSION_REPLY, disconnect_session_reply)                   \
1647 _(APPLICATION_ATTACH_REPLY, application_attach_reply)                   \
1648 _(APPLICATION_DETACH_REPLY, application_detach_reply)                   \
1649 _(MAP_ANOTHER_SEGMENT, map_another_segment)                             \
1650 _(APPLICATION_TLS_CERT_ADD_REPLY, application_tls_cert_add_reply)       \
1651 _(APPLICATION_TLS_KEY_ADD_REPLY, application_tls_key_add_reply)         \
1652
1653 void
1654 quic_echo_api_hookup (echo_main_t * em)
1655 {
1656 #define _(N,n)                                                  \
1657     vl_msg_api_set_handlers(VL_API_##N, #n,                     \
1658                            vl_api_##n##_t_handler,              \
1659                            vl_noop_handler,                     \
1660                            vl_api_##n##_t_endian,               \
1661                            vl_api_##n##_t_print,                \
1662                            sizeof(vl_api_##n##_t), 1);
1663   foreach_quic_echo_msg;
1664 #undef _
1665 }
1666
1667 static void
1668 print_usage_and_exit (void)
1669 {
1670   fprintf (stderr,
1671            "quic_echo [socket-name SOCKET] [client|server] [uri URI] [OPTIONS]\n"
1672            "\n"
1673            "  socket-name PATH    Specify the binary socket path to connect to VPP\n"
1674            "  use-svm-api         Use SVM API to connect to VPP\n"
1675            "  test-bytes[:assert] Check data correctness when receiving (assert fails on first error)\n"
1676            "  fifo-size N         Use N Kb fifos\n"
1677            "  appns NAMESPACE     Use the namespace NAMESPACE\n"
1678            "  all-scope           all-scope option\n"
1679            "  local-scope         local-scope option\n"
1680            "  global-scope        global-scope option\n"
1681            "  secret SECRET       set namespace secret\n"
1682            "  chroot prefix PATH  Use PATH as memory root path\n"
1683            "  quic-setup OPT      OPT=serverstream : Client open N connections. On each one server opens M streams\n"
1684            "                            by default : Client open N connections. On each one client opens M streams\n"
1685            "\n"
1686            "  no-return            Drop the data when received, dont reply\n"
1687            "  nclients N[/M]       Open N QUIC connections, each one with M streams (M defaults to 1)\n"
1688            "  send N[Kb|Mb|GB]     Send N [K|M|G]bytes\n"
1689            "  recv N[Kb|Mb|GB]     Expect N [K|M|G]bytes\n"
1690            "  nclients N[/M]       Open N QUIC connections, each one with M streams (M defaults to 1)\n");
1691   exit (1);
1692 }
1693
1694
1695 void
1696 quic_echo_process_opts (int argc, char **argv)
1697 {
1698   echo_main_t *em = &echo_main;
1699   unformat_input_t _argv, *a = &_argv;
1700   u32 tmp;
1701   u8 *chroot_prefix;
1702   u8 *uri = 0;
1703
1704   unformat_init_command_line (a, argv);
1705   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
1706     {
1707       if (unformat (a, "chroot prefix %s", &chroot_prefix))
1708         {
1709           vl_set_memory_root_path ((char *) chroot_prefix);
1710         }
1711       else if (unformat (a, "uri %s", &uri))
1712         em->uri = format (0, "%s%c", uri, 0);
1713       else if (unformat (a, "server"))
1714         em->i_am_master = 1;
1715       else if (unformat (a, "client"))
1716         em->i_am_master = 0;
1717       else if (unformat (a, "no-return"))
1718         em->no_return = 1;
1719       else if (unformat (a, "test-bytes:assert"))
1720         em->test_return_packets = RETURN_PACKETS_ASSERT;
1721       else if (unformat (a, "test-bytes"))
1722         em->test_return_packets = RETURN_PACKETS_LOG_WRONG;
1723       else if (unformat (a, "socket-name %s", &em->socket_name))
1724         ;
1725       else if (unformat (a, "use-svm-api"))
1726         em->use_sock_api = 0;
1727       else if (unformat (a, "fifo-size %d", &tmp))
1728         em->fifo_size = tmp << 10;
1729       else
1730         if (unformat
1731             (a, "nclients %d/%d", &em->n_clients, &em->n_stream_clients))
1732         ;
1733       else if (unformat (a, "nclients %d", &em->n_clients))
1734         ;
1735       else if (unformat (a, "appns %_%v%_", &em->appns_id))
1736         ;
1737       else if (unformat (a, "all-scope"))
1738         em->appns_flags |= (APP_OPTIONS_FLAGS_USE_GLOBAL_SCOPE
1739                             | APP_OPTIONS_FLAGS_USE_LOCAL_SCOPE);
1740       else if (unformat (a, "local-scope"))
1741         em->appns_flags = APP_OPTIONS_FLAGS_USE_LOCAL_SCOPE;
1742       else if (unformat (a, "global-scope"))
1743         em->appns_flags = APP_OPTIONS_FLAGS_USE_GLOBAL_SCOPE;
1744       else if (unformat (a, "secret %lu", &em->appns_secret))
1745         ;
1746       else if (unformat (a, "quic-setup %U", echo_unformat_quic_setup_vft))
1747         ;
1748       else
1749         if (unformat (a, "send %U", echo_unformat_data, &em->bytes_to_send))
1750         ;
1751       else
1752         if (unformat
1753             (a, "recv %U", echo_unformat_data, &em->bytes_to_receive))
1754         ;
1755       else if (unformat (a, "time %U:%U",
1756                          echo_unformat_timing_event, &em->timing_start_event,
1757                          echo_unformat_timing_event, &em->timing_end_event))
1758         ;
1759       else
1760         print_usage_and_exit ();
1761     }
1762 }
1763
1764 int
1765 main (int argc, char **argv)
1766 {
1767   echo_main_t *em = &echo_main;
1768   fifo_segment_main_t *sm = &em->segment_main;
1769   char *app_name;
1770   int i, rv;
1771   u32 n_clients;
1772
1773   clib_mem_init_thread_safe (0, 256 << 20);
1774   clib_memset (em, 0, sizeof (*em));
1775   em->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
1776   em->shared_segment_handles = hash_create (0, sizeof (uword));
1777   em->my_pid = getpid ();
1778   em->socket_name = format (0, "%s%c", API_SOCKET_FILE, 0);
1779   em->use_sock_api = 1;
1780   em->fifo_size = 64 << 10;
1781   em->n_clients = 1;
1782   em->n_stream_clients = 1;
1783   em->max_test_msg = 50;
1784   em->time_to_stop = 0;
1785   em->i_am_master = 1;
1786   em->test_return_packets = RETURN_PACKETS_NOTEST;
1787   em->timing_start_event = ECHO_EVT_FIRST_QCONNECT;
1788   em->timing_end_event = ECHO_EVT_LAST_BYTE;
1789   em->bytes_to_receive = 64 << 10;
1790   em->bytes_to_send = 64 << 10;
1791   em->uri = format (0, "%s%c", "quic://0.0.0.0/1234", 0);
1792   em->cb_vft = default_cb_vft;
1793   quic_echo_process_opts (argc, argv);
1794
1795   n_clients = em->n_clients * em->n_stream_clients;
1796   vec_validate (em->client_thread_handles, n_clients - 1);
1797   vec_validate (em->thread_args, n_clients - 1);
1798   clib_time_init (&em->clib_time);
1799   init_error_string_table (em);
1800   fifo_segment_main_init (sm, HIGH_SEGMENT_BASEVA, 20);
1801   clib_spinlock_init (&em->segment_handles_lock);
1802   vec_validate (em->rx_buf, 4 << 20);
1803   vec_validate (em->connect_test_data, 1024 * 1024 - 1);
1804   for (i = 0; i < vec_len (em->connect_test_data); i++)
1805     em->connect_test_data[i] = i & 0xff;
1806
1807   setup_signal_handlers ();
1808   quic_echo_api_hookup (em);
1809
1810   app_name = em->i_am_master ? "quic_echo_server" : "quic_echo_client";
1811   if (connect_to_vpp (app_name) < 0)
1812     {
1813       svm_region_exit ();
1814       fformat (stderr, "Couldn't connect to vpe, exiting...\n");
1815       exit (1);
1816     }
1817
1818   quic_echo_notify_event (em, ECHO_EVT_START);
1819   if (em->i_am_master)
1820     rv = server_run (em);
1821   else
1822     rv = clients_run (em);
1823   if (rv)
1824     exit (rv);
1825   quic_echo_notify_event (em, ECHO_EVT_EXIT);
1826   print_global_stats (em);
1827
1828   /* Make sure detach finishes */
1829   if (wait_for_state_change (em, STATE_DETACHED, TIMEOUT))
1830     exit (-1);
1831   disconnect_from_vpp (em);
1832   exit (0);
1833 }
1834
1835 /*
1836  * fd.io coding-style-patch-verification: ON
1837  *
1838  * Local Variables:
1839  * eval: (c-set-style "gnu")
1840  * End:
1841  */