src/plugins/hs_apps/sapi/vpp_echo_proto_quic.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 <hs_apps/sapi/vpp_echo_common.h>
  20
  21 typedef struct _quic_echo_cb_vft
  22 {
  23   void (*quic_connected_cb) (session_connected_msg_t * mp, u32 session_index);
  24   void (*client_stream_connected_cb) (session_connected_msg_t * mp,
  25                                       u32 session_index);
  26   void (*server_stream_connected_cb) (session_connected_msg_t * mp,
  27                                       u32 session_index);
  28   void (*quic_accepted_cb) (session_accepted_msg_t * mp, u32 session_index);
  29   void (*client_stream_accepted_cb) (session_accepted_msg_t * mp,
  30                                      u32 session_index);
  31   void (*server_stream_accepted_cb) (session_accepted_msg_t * mp,
  32                                      u32 session_index);
  33 } quic_echo_cb_vft_t;
  34
  35 typedef struct
  36 {
  37   quic_echo_cb_vft_t cb_vft;    /* cb vft for QUIC scenarios */
  38   u8 send_quic_disconnects;     /* actively send disconnect */
  39   u32 n_stream_clients;         /* Target Number of STREAM sessions per QUIC session */
  40   volatile u32 n_quic_clients_connected;        /* Number of connected QUIC sessions */
  41 } quic_echo_proto_main_t;
  42
  43 quic_echo_proto_main_t quic_echo_proto_main;
  44
  45 /*
  46  *
  47  *  ECHO Callback definitions
  48  *
  49  */
  50
  51 static void
  52 quic_echo_on_connected_connect (session_connected_msg_t * mp,
  53                                 u32 session_index)
  54 {
  55   echo_main_t *em = &echo_main;
  56   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
  57   u8 *uri = format (0, "quic://session/%lu", mp->handle);
  58   u64 i;
  59
  60   echo_notify_event (em, ECHO_EVT_FIRST_SCONNECT);
  61   for (i = 0; i < eqm->n_stream_clients; i++)
  62     echo_send_rpc (em, echo_send_connect, (void *) uri, session_index);
  63
  64   ECHO_LOG (0, "Qsession 0x%llx connected to %U:%d",
  65             mp->handle, format_ip46_address, &mp->lcl.ip,
  66             mp->lcl.is_ip4, clib_net_to_host_u16 (mp->lcl.port));
  67 }
  68
  69 static void
  70 quic_echo_on_connected_send (session_connected_msg_t * mp, u32 session_index)
  71 {
  72   static u32 client_index = 0;
  73   echo_main_t *em = &echo_main;
  74   echo_session_t *session;
  75
  76   session = pool_elt_at_index (em->sessions, session_index);
  77   session->bytes_to_send = em->bytes_to_send;
  78   session->bytes_to_receive = em->bytes_to_receive;
  79   session->session_state = ECHO_SESSION_STATE_READY;
  80   em->data_thread_args[client_index++] = session->session_index;
  81 }
  82
  83 static void
  84 quic_echo_on_connected_error (session_connected_msg_t * mp, u32 session_index)
  85 {
  86   ECHO_FAIL (ECHO_FAIL_QUIC_WRONG_CONNECT,
  87              "Got a wrong connected on session %u [%lx]", session_index,
  88              mp->handle);
  89 }
  90
  91 static void
  92 quic_echo_on_accept_recv (session_accepted_msg_t * mp, u32 session_index)
  93 {
  94   static u32 client_index = 0;
  95   echo_main_t *em = &echo_main;
  96   echo_session_t *session;
  97
  98   session = pool_elt_at_index (em->sessions, session_index);
  99   session->bytes_to_send = em->bytes_to_send;
 100   session->bytes_to_receive = em->bytes_to_receive;
 101   em->data_thread_args[client_index++] = session->session_index;
 102   session->session_state = ECHO_SESSION_STATE_READY;
 103 }
 104
 105 static void
 106 quic_echo_on_accept_connect (session_accepted_msg_t * mp, u32 session_index)
 107 {
 108   echo_main_t *em = &echo_main;
 109   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 110   ECHO_LOG (1, "Accept on QSession 0x%lx %u", mp->handle);
 111   u8 *uri = format (0, "quic://session/%lu", mp->handle);
 112   u32 i;
 113
 114   echo_notify_event (em, ECHO_EVT_FIRST_SCONNECT);
 115   for (i = 0; i < eqm->n_stream_clients; i++)
 116     echo_send_rpc (em, echo_send_connect, (void *) uri, session_index);
 117 }
 118
 119 static void
 120 quic_echo_on_accept_error (session_accepted_msg_t * mp, u32 session_index)
 121 {
 122   ECHO_FAIL (ECHO_FAIL_QUIC_WRONG_ACCEPT,
 123              "Got a wrong accept on session %u [%lx]", session_index,
 124              mp->handle);
 125 }
 126
 127 static void
 128 quic_echo_on_accept_log_ip (session_accepted_msg_t * mp, u32 session_index)
 129 {
 130   u8 *ip_str;
 131   ip_str = format (0, "%U", format_ip46_address, &mp->rmt.ip, mp->rmt.is_ip4);
 132   ECHO_LOG (0, "Accepted session from: %s:%d", ip_str,
 133             clib_net_to_host_u16 (mp->rmt.port));
 134
 135 }
 136
 137 static const quic_echo_cb_vft_t default_cb_vft = {
 138   /* Qsessions */
 139   .quic_accepted_cb = quic_echo_on_accept_log_ip,
 140   .quic_connected_cb = quic_echo_on_connected_connect,
 141   /* client initiated streams */
 142   .server_stream_accepted_cb = quic_echo_on_accept_recv,
 143   .client_stream_connected_cb = quic_echo_on_connected_send,
 144   /* server initiated streams */
 145   .client_stream_accepted_cb = quic_echo_on_accept_error,
 146   .server_stream_connected_cb = quic_echo_on_connected_error,
 147 };
 148
 149 static const quic_echo_cb_vft_t server_stream_cb_vft = {
 150   /* Qsessions */
 151   .quic_accepted_cb = quic_echo_on_accept_connect,
 152   .quic_connected_cb = NULL,
 153   /* client initiated streams */
 154   .server_stream_accepted_cb = quic_echo_on_accept_error,
 155   .client_stream_connected_cb = quic_echo_on_connected_error,
 156   /* server initiated streams */
 157   .client_stream_accepted_cb = quic_echo_on_accept_recv,
 158   .server_stream_connected_cb = quic_echo_on_connected_send,
 159 };
 160
 161 static void quic_echo_cleanup_cb (echo_session_t * s, u8 parent_died);
 162
 163 static inline void
 164 quic_echo_cleanup_listener (u32 listener_index, echo_main_t * em,
 165                             quic_echo_proto_main_t * eqm)
 166 {
 167   echo_session_t *ls;
 168   ls = pool_elt_at_index (em->sessions, listener_index);
 169   ASSERT (ls->session_type == ECHO_SESSION_TYPE_QUIC);
 170   if (!clib_atomic_sub_fetch (&ls->accepted_session_count, 1))
 171     {
 172       if (eqm->send_quic_disconnects == ECHO_CLOSE_F_ACTIVE)
 173         {
 174           echo_send_rpc (em, echo_send_disconnect_session,
 175                          (void *) ls->vpp_session_handle, 0);
 176           clib_atomic_fetch_add (&em->stats.active_count.q, 1);
 177         }
 178       else if (eqm->send_quic_disconnects == ECHO_CLOSE_F_NONE)
 179         {
 180           quic_echo_cleanup_cb (ls, 0 /* parent_died */ );
 181           clib_atomic_fetch_add (&em->stats.clean_count.q, 1);
 182         }
 183     }
 184 }
 185
 186 static void
 187 quic_echo_cleanup_cb (echo_session_t * s, u8 parent_died)
 188 {
 189   echo_main_t *em = &echo_main;
 190   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 191   ASSERT (s->session_state < ECHO_SESSION_STATE_CLOSED);
 192   if (s->session_type == ECHO_SESSION_TYPE_QUIC)
 193     {
 194       if (parent_died)
 195         clib_atomic_fetch_add (&em->stats.clean_count.q, 1);
 196       /* Don't cleanup listener as it's handled by main() */
 197       clib_atomic_sub_fetch (&eqm->n_quic_clients_connected, 1);
 198     }
 199   else if (s->session_type == ECHO_SESSION_TYPE_STREAM)
 200     {
 201       if (parent_died)
 202         clib_atomic_fetch_add (&em->stats.clean_count.s, 1);
 203       else
 204         quic_echo_cleanup_listener (s->listener_index, em, eqm);
 205       clib_atomic_sub_fetch (&em->n_clients_connected, 1);
 206     }
 207
 208   ECHO_LOG (1, "Cleanup sessions (still %uQ %uS)",
 209             eqm->n_quic_clients_connected, em->n_clients_connected);
 210   s->session_state = ECHO_SESSION_STATE_CLOSED;
 211   if (!em->n_clients_connected && !eqm->n_quic_clients_connected)
 212     em->state = STATE_DATA_DONE;
 213 }
 214
 215 static void
 216 quic_echo_initiate_qsession_close_no_stream (echo_main_t * em)
 217 {
 218   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 219   ECHO_LOG (1, "Closing Qsessions");
 220   /* Close Quic session without streams */
 221   echo_session_t *s;
 222
 223   /* *INDENT-OFF* */
 224   pool_foreach (s, em->sessions,
 225   ({
 226     if (s->session_type == ECHO_SESSION_TYPE_QUIC)
 227       {
 228         if (eqm->send_quic_disconnects == ECHO_CLOSE_F_ACTIVE)
 229           {
 230             ECHO_LOG (1,"ACTIVE close 0x%lx", s->vpp_session_handle);
 231             echo_send_rpc (em, echo_send_disconnect_session, (void *) s->vpp_session_handle, 0);
 232             clib_atomic_fetch_add (&em->stats.active_count.q, 1);
 233           }
 234         else if (eqm->send_quic_disconnects == ECHO_CLOSE_F_NONE)
 235           {
 236             ECHO_LOG (1,"Discard close 0x%lx", s->vpp_session_handle);
 237             quic_echo_cleanup_cb (s, 0 /* parent_died */);
 238             clib_atomic_fetch_add (&em->stats.clean_count.q, 1);
 239           }
 240         else
 241           ECHO_LOG (1,"Passive close 0x%lx", s->vpp_session_handle);
 242       }
 243   }));
 244   /* *INDENT-ON* */
 245 }
 246
 247 static void
 248 quic_echo_on_connected (session_connected_msg_t * mp, u32 session_index)
 249 {
 250   echo_main_t *em = &echo_main;
 251   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 252   echo_session_t *listen_session;
 253   echo_session_t *session = pool_elt_at_index (em->sessions, session_index);
 254   if (session->listener_index == SESSION_INVALID_INDEX)
 255     {
 256       ECHO_LOG (1, "Connected session 0x%lx -> URI", mp->handle);
 257       session->session_type = ECHO_SESSION_TYPE_QUIC;
 258       session->accepted_session_count = 0;
 259       if (eqm->cb_vft.quic_connected_cb)
 260         eqm->cb_vft.quic_connected_cb (mp, session->session_index);
 261       clib_atomic_fetch_add (&eqm->n_quic_clients_connected, 1);
 262     }
 263   else
 264     {
 265       listen_session =
 266         pool_elt_at_index (em->sessions, session->listener_index);
 267       ECHO_LOG (1, "Connected session 0x%lx -> 0x%lx", mp->handle,
 268                 listen_session->vpp_session_handle);
 269       session->session_type = ECHO_SESSION_TYPE_STREAM;
 270       clib_atomic_fetch_add (&listen_session->accepted_session_count, 1);
 271       if (em->i_am_master && eqm->cb_vft.server_stream_connected_cb)
 272         eqm->cb_vft.server_stream_connected_cb (mp, session->session_index);
 273       if (!em->i_am_master && eqm->cb_vft.client_stream_connected_cb)
 274         eqm->cb_vft.client_stream_connected_cb (mp, session->session_index);
 275       clib_atomic_fetch_add (&em->n_clients_connected, 1);
 276     }
 277
 278   if (em->n_clients_connected == em->n_clients
 279       && em->n_clients_connected != 0)
 280     echo_notify_event (em, ECHO_EVT_LAST_SCONNECTED);
 281
 282   if (eqm->n_quic_clients_connected == em->n_connects
 283       && em->state < STATE_READY)
 284     {
 285       echo_notify_event (em, ECHO_EVT_LAST_QCONNECTED);
 286       em->state = STATE_READY;
 287       if (eqm->n_stream_clients == 0)
 288         quic_echo_initiate_qsession_close_no_stream (em);
 289     }
 290 }
 291
 292 static void
 293 quic_echo_retry_connect (u32 session_index)
 294 {
 295   /* retry connect */
 296   echo_session_t *session;
 297   echo_main_t *em = &echo_main;
 298   u8 *uri;
 299   if (session_index == SESSION_INVALID_INDEX)
 300     {
 301       ECHO_LOG (1, "Retrying connect %s", em->uri);
 302       echo_send_rpc (em, echo_send_connect, (void *) em->uri,
 303                      SESSION_INVALID_INDEX);
 304     }
 305   else
 306     {
 307       session = pool_elt_at_index (em->sessions, session_index);
 308       uri = format (0, "quic://session/%lu", session->vpp_session_handle);
 309       ECHO_LOG (1, "Retrying connect %s", uri);
 310       echo_send_rpc (em, echo_send_connect, (void *) uri, session_index);
 311     }
 312 }
 313
 314 static void
 315 quic_echo_connected_cb (session_connected_bundled_msg_t * mp,
 316                         u32 session_index, u8 is_failed)
 317 {
 318   if (is_failed)
 319     return quic_echo_retry_connect (session_index);
 320   return quic_echo_on_connected ((session_connected_msg_t *) mp,
 321                                  session_index);
 322 }
 323
 324 static void
 325 quic_echo_accepted_cb (session_accepted_msg_t * mp, echo_session_t * session)
 326 {
 327   echo_main_t *em = &echo_main;
 328   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 329   echo_session_t *ls;
 330   ls = pool_elt_at_index (em->sessions, session->listener_index);
 331   if (ls->session_type == ECHO_SESSION_TYPE_LISTEN)
 332     {
 333       echo_notify_event (em, ECHO_EVT_FIRST_QCONNECT);
 334       session->session_type = ECHO_SESSION_TYPE_QUIC;
 335       session->accepted_session_count = 0;
 336       if (eqm->cb_vft.quic_accepted_cb)
 337         eqm->cb_vft.quic_accepted_cb (mp, session->session_index);
 338       clib_atomic_fetch_add (&eqm->n_quic_clients_connected, 1);
 339     }
 340   else
 341     {
 342       session->session_type = ECHO_SESSION_TYPE_STREAM;
 343       echo_notify_event (em, ECHO_EVT_FIRST_SCONNECT);
 344       clib_atomic_fetch_add (&ls->accepted_session_count, 1);
 345       if (em->i_am_master && eqm->cb_vft.server_stream_accepted_cb)
 346         eqm->cb_vft.server_stream_accepted_cb (mp, session->session_index);
 347       if (!em->i_am_master && eqm->cb_vft.client_stream_accepted_cb)
 348         eqm->cb_vft.client_stream_accepted_cb (mp, session->session_index);
 349       clib_atomic_fetch_add (&em->n_clients_connected, 1);
 350     }
 351
 352   if (em->n_clients_connected == em->n_clients
 353       && em->n_clients_connected != 0)
 354     echo_notify_event (em, ECHO_EVT_LAST_SCONNECTED);
 355
 356   if (eqm->n_quic_clients_connected == em->n_connects
 357       && em->state < STATE_READY)
 358     {
 359       echo_notify_event (em, ECHO_EVT_LAST_QCONNECTED);
 360       em->state = STATE_READY;
 361       if (eqm->n_stream_clients == 0)
 362         quic_echo_initiate_qsession_close_no_stream (em);
 363     }
 364 }
 365
 366 static void
 367 quic_echo_disconnected_reply_cb (echo_session_t * s)
 368 {
 369   if (s->session_type == ECHO_SESSION_TYPE_STREAM)
 370     s->session_state = ECHO_SESSION_STATE_CLOSING;
 371   else
 372     quic_echo_cleanup_cb (s, 0 /* parent_died */ );     /* We can clean Q/Lsessions right away */
 373 }
 374
 375 static void
 376 quic_echo_disconnected_cb (session_disconnected_msg_t * mp,
 377                            echo_session_t * s)
 378 {
 379   echo_main_t *em = &echo_main;
 380   if (s->session_type == ECHO_SESSION_TYPE_STREAM)
 381     {
 382       echo_session_print_stats (em, s);
 383       if (s->bytes_to_receive || s->bytes_to_send)
 384         s->session_state = ECHO_SESSION_STATE_AWAIT_DATA;
 385       else
 386         s->session_state = ECHO_SESSION_STATE_CLOSING;
 387       clib_atomic_fetch_add (&em->stats.close_count.s, 1);
 388     }
 389   else
 390     {
 391       quic_echo_cleanup_cb (s, 0 /* parent_died */ );   /* We can clean Q/Lsessions right away */
 392       clib_atomic_fetch_add (&em->stats.close_count.q, 1);
 393     }
 394 }
 395
 396 static void
 397 quic_echo_reset_cb (session_reset_msg_t * mp, echo_session_t * s)
 398 {
 399   echo_main_t *em = &echo_main;
 400   if (s->session_type == ECHO_SESSION_TYPE_STREAM)
 401     {
 402       clib_atomic_fetch_add (&em->stats.reset_count.s, 1);
 403       s->session_state = ECHO_SESSION_STATE_CLOSING;
 404     }
 405   else
 406     {
 407       clib_atomic_fetch_add (&em->stats.reset_count.q, 1);
 408       quic_echo_cleanup_cb (s, 0 /* parent_died */ );   /* We can clean Q/Lsessions right away */
 409     }
 410 }
 411
 412 static uword
 413 quic_echo_unformat_setup_vft (unformat_input_t * input, va_list * args)
 414 {
 415   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 416   if (unformat (input, "serverstream"))
 417     eqm->cb_vft = server_stream_cb_vft;
 418   else if (unformat (input, "default"))
 419     ;
 420   else
 421     return 0;
 422   return 1;
 423 }
 424
 425 static int
 426 quic_echo_process_opts_cb (unformat_input_t * a)
 427 {
 428   echo_main_t *em = &echo_main;
 429   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 430   if (unformat (a, "nclients %d/%d", &em->n_clients, &eqm->n_stream_clients))
 431     ;
 432   else if (unformat (a, "quic-setup %U", quic_echo_unformat_setup_vft))
 433     ;
 434   else if (unformat (a, "qclose=%U",
 435                      echo_unformat_close, &eqm->send_quic_disconnects))
 436     ;
 437   else
 438     return 0;
 439   return 1;
 440 }
 441
 442 static void
 443 quic_echo_set_defaults_before_opts_cb ()
 444 {
 445   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 446   eqm->cb_vft = default_cb_vft;
 447   eqm->n_stream_clients = 1;
 448 }
 449
 450 static void
 451 quic_echo_set_defaults_after_opts_cb ()
 452 {
 453   quic_echo_proto_main_t *eqm = &quic_echo_proto_main;
 454   echo_main_t *em = &echo_main;
 455   u8 default_f_active;
 456
 457   em->n_connects = em->n_clients;
 458   em->n_sessions =
 459     clib_max (1, eqm->n_stream_clients) * em->n_clients + em->n_clients + 1;
 460   em->n_clients = eqm->n_stream_clients * em->n_clients;
 461
 462   if (em->i_am_master)
 463     default_f_active =
 464       em->bytes_to_send == 0 ? ECHO_CLOSE_F_ACTIVE : ECHO_CLOSE_F_PASSIVE;
 465   else
 466     default_f_active =
 467       em->bytes_to_receive == 0 ? ECHO_CLOSE_F_PASSIVE : ECHO_CLOSE_F_ACTIVE;
 468   if (eqm->send_quic_disconnects == ECHO_CLOSE_F_INVALID)
 469     eqm->send_quic_disconnects = default_f_active;
 470 }
 471
 472 static void
 473 quic_echo_print_usage_cb ()
 474 {
 475   fprintf (stderr,
 476            "-- QUIC specific options -- \n"
 477            "  quic-setup OPT      OPT=serverstream : Client open N connections. \n"
 478            "                       On each one server opens M streams\n"
 479            "                      OPT=default : Client open N connections.\n"
 480            "                       On each one client opens M streams\n"
 481            "  qclose=[Y|N|W]      When a connection is done pass[N] send[Y] or wait[W] for close\n"
 482            "\n"
 483            "  nclients N[/M]      Open N QUIC connections, each one with M streams (M defaults to 1)\n");
 484 }
 485
 486 echo_proto_cb_vft_t quic_echo_proto_cb_vft = {
 487   .disconnected_cb = quic_echo_disconnected_cb,
 488   .connected_cb = quic_echo_connected_cb,
 489   .accepted_cb = quic_echo_accepted_cb,
 490   .reset_cb = quic_echo_reset_cb,
 491   .disconnected_reply_cb = quic_echo_disconnected_reply_cb,
 492   .cleanup_cb = quic_echo_cleanup_cb,
 493   .process_opts_cb = quic_echo_process_opts_cb,
 494   .print_usage_cb = quic_echo_print_usage_cb,
 495   .set_defaults_before_opts_cb = quic_echo_set_defaults_before_opts_cb,
 496   .set_defaults_after_opts_cb = quic_echo_set_defaults_after_opts_cb,
 497 };
 498
 499 ECHO_REGISTER_PROTO (TRANSPORT_PROTO_QUIC, quic_echo_proto_cb_vft);
 500
 501 /*
 502  * fd.io coding-style-patch-verification: ON
 503  *
 504  * Local Variables:
 505  * eval: (c-set-style "gnu")
 506  * End:
 507  */