src/plugins/hs_apps/vcl/vcl_test_server.c

   1 /*
   2  * Copyright (c) 2017-2021 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 <unistd.h>
  17 #include <errno.h>
  18 #include <sys/types.h>
  19 #include <sys/socket.h>
  20 #include <stdio.h>
  21 #include <string.h>
  22 #include <time.h>
  23 #include <ctype.h>
  24 #include <sys/stat.h>
  25 #include <fcntl.h>
  26 #include <hs_apps/vcl/vcl_test.h>
  27 #include <sys/epoll.h>
  28 #include <vppinfra/mem.h>
  29 #include <pthread.h>
  30
  31 typedef struct
  32 {
  33   uint16_t port;
  34   uint32_t address_ip6;
  35   u8 proto;
  36   u8 workers;
  37   vppcom_endpt_t endpt;
  38 } vcl_test_server_cfg_t;
  39
  40 typedef struct
  41 {
  42   vcl_test_session_t *conn_pool;
  43   uint32_t wrk_index;
  44   int epfd;
  45   int conn_pool_size;
  46   int nfds;
  47   vcl_test_session_t listener;
  48   pthread_t thread_handle;
  49 } vcl_test_server_worker_t;
  50
  51 typedef struct
  52 {
  53   vcl_test_server_worker_t *workers;
  54   vcl_test_session_t *ctrl;
  55   vcl_test_server_cfg_t server_cfg;
  56   int ctrl_listen_fd;
  57   struct sockaddr_storage servaddr;
  58   volatile int worker_fails;
  59   volatile int active_workers;
  60   u8 use_ds;
  61 } vcl_test_server_main_t;
  62
  63 vcl_test_main_t vcl_test_main;
  64
  65 static vcl_test_server_main_t vcl_server_main;
  66
  67 static inline void
  68 conn_pool_expand (vcl_test_server_worker_t * wrk, size_t expand_size)
  69 {
  70   vcl_test_session_t *conn_pool;
  71   size_t new_size = wrk->conn_pool_size + expand_size;
  72   int i;
  73
  74   conn_pool = realloc (wrk->conn_pool, new_size * sizeof (*wrk->conn_pool));
  75   if (conn_pool)
  76     {
  77       for (i = wrk->conn_pool_size; i < new_size; i++)
  78         {
  79           vcl_test_session_t *conn = &conn_pool[i];
  80           memset (conn, 0, sizeof (*conn));
  81         }
  82
  83       wrk->conn_pool = conn_pool;
  84       wrk->conn_pool_size = new_size;
  85     }
  86   else
  87     {
  88       vterr ("conn_pool_expand()", -errno);
  89     }
  90 }
  91
  92 static inline vcl_test_session_t *
  93 conn_pool_alloc (vcl_test_server_worker_t *wrk)
  94 {
  95   vcl_test_session_t *conn;
  96   int i, expand = 0;
  97
  98 again:
  99   for (i = 0; i < wrk->conn_pool_size; i++)
 100     {
 101       if (!wrk->conn_pool[i].is_alloc)
 102         {
 103           conn = &wrk->conn_pool[i];
 104           conn->endpt.ip = wrk->conn_pool[i].ip;
 105           conn->is_alloc = 1;
 106           conn->session_index = i;
 107           memset (&conn->stats, 0, sizeof (vcl_test_stats_t));
 108           vcl_test_cfg_init (&conn->cfg);
 109           return (&wrk->conn_pool[i]);
 110         }
 111     }
 112
 113   if (expand == 0)
 114     {
 115       conn_pool_expand (wrk, 2 * wrk->conn_pool_size);
 116       expand = 1;
 117       goto again;
 118     }
 119   vtwrn ("Failed to allocate connection even after expand");
 120   return 0;
 121 }
 122
 123 static inline void
 124 conn_pool_free (vcl_test_session_t *ts)
 125 {
 126   ts->fd = 0;
 127   ts->is_alloc = 0;
 128   vcl_test_session_buf_free (ts);
 129 }
 130
 131 static inline void
 132 sync_config_and_reply (vcl_test_session_t *conn, vcl_test_cfg_t *rx_cfg)
 133 {
 134   conn->cfg = *rx_cfg;
 135   vcl_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */, (uint8_t **) &conn->rxbuf,
 136                       &conn->rxbuf_size);
 137   conn->cfg.txbuf_size = conn->cfg.rxbuf_size;
 138
 139   if (conn->cfg.verbose)
 140     {
 141       vtinf ("(fd %d): Replying to cfg message!\n", conn->fd);
 142       vcl_test_cfg_dump (&conn->cfg, 0 /* is_client */ );
 143     }
 144   (void) vcl_test_write (conn, &conn->cfg, sizeof (conn->cfg));
 145 }
 146
 147 static void
 148 vts_session_close (vcl_test_session_t *conn)
 149 {
 150   vcl_test_server_main_t *vsm = &vcl_server_main;
 151   vcl_test_main_t *vt = &vcl_test_main;
 152
 153   if (!conn->is_open)
 154     return;
 155
 156   if (vt->protos[vsm->server_cfg.proto]->close)
 157     vt->protos[vsm->server_cfg.proto]->close (conn);
 158
 159   vppcom_session_close (conn->fd);
 160   conn->is_open = 0;
 161 }
 162
 163 static void
 164 vts_session_cleanup (vcl_test_session_t *ts)
 165 {
 166   vts_session_close (ts);
 167   conn_pool_free (ts);
 168 }
 169
 170 static void
 171 vts_wrk_cleanup_all (vcl_test_server_worker_t *wrk)
 172 {
 173   vcl_test_session_t *conn;
 174   int i;
 175
 176   for (i = 0; i < wrk->conn_pool_size; i++)
 177     {
 178       conn = &wrk->conn_pool[i];
 179       vts_session_cleanup (conn);
 180     }
 181
 182   wrk->nfds = 0;
 183 }
 184
 185 static void
 186 vts_test_cmd (vcl_test_server_worker_t *wrk, vcl_test_session_t *conn,
 187               vcl_test_cfg_t *rx_cfg)
 188 {
 189   u8 is_bi = rx_cfg->test == VCL_TEST_TYPE_BI;
 190   vcl_test_session_t *tc;
 191   char buf[64];
 192   int i;
 193
 194   if (rx_cfg->cmd == VCL_TEST_CMD_STOP)
 195     {
 196       struct timespec stop;
 197       clock_gettime (CLOCK_REALTIME, &stop);
 198
 199       /* Test session are not closed, e.g., connection-less or errors */
 200       if (wrk->nfds > 1)
 201         {
 202           vtinf ("%u sessions are still open", wrk->nfds - 1);
 203           stop.tv_sec -= VCL_TEST_DELAY_DISCONNECT;
 204           conn->stats.stop = stop;
 205         }
 206
 207       /* Accumulate stats over all of the worker's sessions */
 208       for (i = 0; i < wrk->conn_pool_size; i++)
 209         {
 210           tc = &wrk->conn_pool[i];
 211           if (tc == conn)
 212             continue;
 213
 214           vcl_test_stats_accumulate (&conn->stats, &tc->stats);
 215           if (tc->is_open)
 216             {
 217               vts_session_cleanup (tc);
 218               continue;
 219             }
 220           /* Only relevant if all connections previously closed */
 221           if (vcl_comp_tspec (&conn->stats.stop, &tc->stats.stop) < 0)
 222             conn->stats.stop = tc->stats.stop;
 223         }
 224
 225       if (conn->cfg.verbose)
 226         {
 227           snprintf (buf, sizeof (buf), "SERVER (fd %d) RESULTS", conn->fd);
 228           vcl_test_stats_dump (buf, &conn->stats, 1 /* show_rx */,
 229                                is_bi /* show tx */, conn->cfg.verbose);
 230         }
 231
 232       vcl_test_stats_dump ("SERVER RESULTS", &conn->stats, 1 /* show_rx */ ,
 233                            is_bi /* show_tx */ , conn->cfg.verbose);
 234       vcl_test_cfg_dump (&conn->cfg, 0 /* is_client */ );
 235       if (conn->cfg.verbose)
 236         {
 237           vtinf ("  vcl server main\n" VCL_TEST_SEPARATOR_STRING
 238                  "       buf:  %p\n"
 239                  "  buf size:  %u (0x%08x)\n" VCL_TEST_SEPARATOR_STRING,
 240                  conn->rxbuf, conn->rxbuf_size, conn->rxbuf_size);
 241         }
 242
 243       sync_config_and_reply (conn, rx_cfg);
 244       memset (&conn->stats, 0, sizeof (conn->stats));
 245     }
 246   else if (rx_cfg->cmd == VCL_TEST_CMD_SYNC)
 247     {
 248       rx_cfg->ctrl_handle = conn->fd;
 249       vtinf ("Set control fd %d for test!", conn->fd);
 250       sync_config_and_reply (conn, rx_cfg);
 251     }
 252   else if (rx_cfg->cmd == VCL_TEST_CMD_START)
 253     {
 254       vtinf ("Starting %s-directional Stream Test (fd %d)!",
 255              is_bi ? "Bi" : "Uni", conn->fd);
 256       rx_cfg->ctrl_handle = conn->fd;
 257       sync_config_and_reply (conn, rx_cfg);
 258
 259       /* read the 1st chunk, record start time */
 260       memset (&conn->stats, 0, sizeof (conn->stats));
 261       clock_gettime (CLOCK_REALTIME, &conn->stats.start);
 262     }
 263 }
 264
 265 static inline void
 266 vts_server_process_rx (vcl_test_session_t *conn, int rx_bytes)
 267 {
 268   vcl_test_server_main_t *vsm = &vcl_server_main;
 269
 270   if (conn->cfg.test == VCL_TEST_TYPE_BI)
 271     {
 272       if (vsm->use_ds)
 273         {
 274           (void) vcl_test_write (conn, conn->ds[0].data, conn->ds[0].len);
 275           if (conn->ds[1].len)
 276             (void) vcl_test_write (conn, conn->ds[1].data, conn->ds[1].len);
 277         }
 278       else
 279         (void) vcl_test_write (conn, conn->rxbuf, rx_bytes);
 280     }
 281
 282   if (vsm->use_ds)
 283     vppcom_session_free_segments (conn->fd, rx_bytes);
 284
 285   if (conn->stats.rx_bytes >= conn->cfg.total_bytes)
 286     clock_gettime (CLOCK_REALTIME, &conn->stats.stop);
 287 }
 288
 289 static void
 290 vts_server_echo (vcl_test_session_t *conn, int rx_bytes)
 291 {
 292   int tx_bytes, nbytes, pos;
 293
 294   /* If it looks vaguely like a string, make sure it's terminated */
 295   pos = rx_bytes < conn->rxbuf_size ? rx_bytes : conn->rxbuf_size - 1;
 296   ((char *) conn->rxbuf)[pos] = 0;
 297   vtinf ("(fd %d): RX (%d bytes) - '%s'", conn->fd, rx_bytes, conn->rxbuf);
 298
 299   if (conn->cfg.verbose)
 300     vtinf ("(fd %d): Echoing back", conn->fd);
 301
 302   nbytes = strlen ((const char *) conn->rxbuf) + 1;
 303   tx_bytes = conn->write (conn, conn->rxbuf, nbytes);
 304   if (tx_bytes >= 0)
 305     vtinf ("(fd %d): TX (%d bytes) - '%s'", conn->fd, tx_bytes, conn->rxbuf);
 306 }
 307
 308 static vcl_test_session_t *
 309 vts_accept_client (vcl_test_server_worker_t *wrk, int listen_fd)
 310 {
 311   vcl_test_server_main_t *vsm = &vcl_server_main;
 312   const vcl_test_proto_vft_t *tp;
 313   vcl_test_session_t *conn;
 314   struct epoll_event ev;
 315   int rv;
 316
 317   conn = conn_pool_alloc (wrk);
 318   if (!conn)
 319     {
 320       vtwrn ("No free connections!");
 321       return 0;
 322     }
 323
 324   if (vsm->ctrl)
 325     conn->cfg = vsm->ctrl->cfg;
 326   vcl_test_session_buf_alloc (conn);
 327
 328   tp = vcl_test_main.protos[vsm->server_cfg.proto];
 329   if (tp->accept (listen_fd, conn))
 330     return 0;
 331
 332   vtinf ("Got a connection -- fd = %d (0x%08x) on listener fd = %d (0x%08x)",
 333          conn->fd, conn->fd, listen_fd, listen_fd);
 334
 335   ev.events = EPOLLIN;
 336   ev.data.u64 = conn - wrk->conn_pool;
 337   rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, conn->fd, &ev);
 338   if (rv < 0)
 339     {
 340       vterr ("vppcom_epoll_ctl()", rv);
 341       return 0;
 342     }
 343   wrk->nfds++;
 344
 345   return conn;
 346 }
 347
 348 static void
 349 print_usage_and_exit (void)
 350 {
 351   fprintf (stderr,
 352            "vcl_test_server [OPTIONS] <port>\n"
 353            "  OPTIONS\n"
 354            "  -h               Print this message and exit.\n"
 355            "  -6               Use IPv6\n"
 356            "  -w <num>         Number of workers\n"
 357            "  -p <PROTO>       Use <PROTO> transport layer\n"
 358            "  -D               Use UDP transport layer\n"
 359            "  -L               Use TLS transport layer\n");
 360   exit (1);
 361 }
 362
 363 static void
 364 vcl_test_init_endpoint_addr (vcl_test_server_main_t * vsm)
 365 {
 366   struct sockaddr_storage *servaddr = &vsm->servaddr;
 367   memset (servaddr, 0, sizeof (*servaddr));
 368
 369   if (vsm->server_cfg.address_ip6)
 370     {
 371       struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr;
 372       server_addr->sin6_family = AF_INET6;
 373       server_addr->sin6_addr = in6addr_any;
 374       server_addr->sin6_port = htons (vsm->server_cfg.port);
 375     }
 376   else
 377     {
 378       struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr;
 379       server_addr->sin_family = AF_INET;
 380       server_addr->sin_addr.s_addr = htonl (INADDR_ANY);
 381       server_addr->sin_port = htons (vsm->server_cfg.port);
 382     }
 383
 384   if (vsm->server_cfg.address_ip6)
 385     {
 386       struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr;
 387       vsm->server_cfg.endpt.is_ip4 = 0;
 388       vsm->server_cfg.endpt.ip = (uint8_t *) &server_addr->sin6_addr;
 389       vsm->server_cfg.endpt.port = (uint16_t) server_addr->sin6_port;
 390     }
 391   else
 392     {
 393       struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr;
 394       vsm->server_cfg.endpt.is_ip4 = 1;
 395       vsm->server_cfg.endpt.ip = (uint8_t *) &server_addr->sin_addr;
 396       vsm->server_cfg.endpt.port = (uint16_t) server_addr->sin_port;
 397     }
 398 }
 399
 400 static void
 401 vcl_test_server_process_opts (vcl_test_server_main_t * vsm, int argc,
 402                               char **argv)
 403 {
 404   int v, c;
 405
 406   vsm->server_cfg.proto = VPPCOM_PROTO_TCP;
 407
 408   opterr = 0;
 409   while ((c = getopt (argc, argv, "6DLsw:hp:")) != -1)
 410     switch (c)
 411       {
 412       case '6':
 413         vsm->server_cfg.address_ip6 = 1;
 414         break;
 415
 416       case 'p':
 417         if (vppcom_unformat_proto (&vsm->server_cfg.proto, optarg))
 418           vtwrn ("Invalid vppcom protocol %s, defaulting to TCP", optarg);
 419         break;
 420
 421       case 'D':
 422         vsm->server_cfg.proto = VPPCOM_PROTO_UDP;
 423         break;
 424
 425       case 'L':
 426         vsm->server_cfg.proto = VPPCOM_PROTO_TLS;
 427         break;
 428
 429       case 'w':
 430         v = atoi (optarg);
 431         if (v > 1)
 432           vsm->server_cfg.workers = v;
 433         else
 434           vtwrn ("Invalid number of workers %d", v);
 435         break;
 436       case 's':
 437         vsm->use_ds = 1;
 438         break;
 439       case '?':
 440         switch (optopt)
 441           {
 442           case 'w':
 443           case 'p':
 444             vtwrn ("Option `-%c' requires an argument.", optopt);
 445             break;
 446           default:
 447             if (isprint (optopt))
 448               vtwrn ("Unknown option `-%c'.", optopt);
 449             else
 450               vtwrn ("Unknown option character `\\x%x'.", optopt);
 451           }
 452         /* fall thru */
 453       case 'h':
 454       default:
 455         print_usage_and_exit ();
 456       }
 457
 458   if (argc < (optind + 1))
 459     {
 460       fprintf (stderr, "SERVER: ERROR: Insufficient number of arguments!\n");
 461       print_usage_and_exit ();
 462     }
 463
 464   if (sscanf (argv[optind], "%d", &v) == 1)
 465     vsm->server_cfg.port = (uint16_t) v;
 466   else
 467     {
 468       fprintf (stderr, "SERVER: ERROR: Invalid port (%s)!\n", argv[optind]);
 469       print_usage_and_exit ();
 470     }
 471
 472   vcl_test_init_endpoint_addr (vsm);
 473 }
 474
 475 int
 476 vts_handle_ctrl_cfg (vcl_test_server_worker_t *wrk, vcl_test_cfg_t *rx_cfg,
 477                      vcl_test_session_t *conn, int rx_bytes)
 478 {
 479   if (rx_cfg->verbose)
 480     {
 481       vtinf ("(fd %d): Received a cfg msg!", conn->fd);
 482       vcl_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 483     }
 484
 485   if (rx_bytes != sizeof (*rx_cfg))
 486     {
 487       vtinf ("(fd %d): Invalid cfg msg size %d expected %lu!", conn->fd,
 488              rx_bytes, sizeof (*rx_cfg));
 489       conn->cfg.rxbuf_size = 0;
 490       conn->cfg.num_writes = 0;
 491       if (conn->cfg.verbose)
 492         {
 493           vtinf ("(fd %d): Replying to cfg msg", conn->fd);
 494           vcl_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 495         }
 496       conn->write (conn, &conn->cfg, sizeof (conn->cfg));
 497       return -1;
 498     }
 499
 500   switch (rx_cfg->test)
 501     {
 502     case VCL_TEST_TYPE_NONE:
 503     case VCL_TEST_TYPE_ECHO:
 504       sync_config_and_reply (conn, rx_cfg);
 505       break;
 506
 507     case VCL_TEST_TYPE_BI:
 508     case VCL_TEST_TYPE_UNI:
 509       vts_test_cmd (wrk, conn, rx_cfg);
 510       break;
 511
 512     case VCL_TEST_TYPE_EXIT:
 513       vtinf ("Ctrl session fd %d closing!", conn->fd);
 514       vts_session_cleanup (conn);
 515       wrk->nfds--;
 516       if (wrk->nfds)
 517         vts_wrk_cleanup_all (wrk);
 518       break;
 519
 520     default:
 521       vtwrn ("Unknown test type %d", rx_cfg->test);
 522       vcl_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 523       break;
 524     }
 525
 526   return 0;
 527 }
 528
 529 static void
 530 vts_worker_init (vcl_test_server_worker_t * wrk)
 531 {
 532   vcl_test_server_main_t *vsm = &vcl_server_main;
 533   vcl_test_main_t *vt = &vcl_test_main;
 534   const vcl_test_proto_vft_t *tp;
 535   struct epoll_event listen_ev;
 536   int rv;
 537
 538   __wrk_index = wrk->wrk_index;
 539
 540   vtinf ("Initializing worker ...");
 541
 542   conn_pool_expand (wrk, VCL_TEST_CFG_MAX_TEST_SESS + 1);
 543   if (wrk->wrk_index)
 544     if (vppcom_worker_register ())
 545       vtfail ("vppcom_worker_register()", 1);
 546
 547   tp = vt->protos[vsm->server_cfg.proto];
 548   if ((rv = tp->listen (&wrk->listener, &vsm->server_cfg.endpt)))
 549     vtfail ("proto listen", rv);
 550
 551   /* First worker already has epoll fd */
 552   if (wrk->wrk_index)
 553     {
 554       wrk->epfd = vppcom_epoll_create ();
 555       if (wrk->epfd < 0)
 556         vtfail ("vppcom_epoll_create()", wrk->epfd);
 557     }
 558
 559   listen_ev.events = EPOLLIN;
 560   listen_ev.data.u32 = VCL_TEST_DATA_LISTENER;
 561   rv =
 562     vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, wrk->listener.fd, &listen_ev);
 563   if (rv < 0)
 564     vtfail ("vppcom_epoll_ctl", rv);
 565
 566   vsm->active_workers += 1;
 567   vtinf ("Waiting for client data connections on port %d ...",
 568          ntohs (vsm->server_cfg.endpt.port));
 569 }
 570
 571 static inline int
 572 vts_conn_read (vcl_test_session_t *conn)
 573 {
 574   vcl_test_server_main_t *vsm = &vcl_server_main;
 575   if (vsm->use_ds)
 576     return vcl_test_read_ds (conn);
 577   else
 578     return conn->read (conn, conn->rxbuf, conn->rxbuf_size);
 579 }
 580
 581 static void *
 582 vts_worker_loop (void *arg)
 583 {
 584   struct epoll_event ep_evts[VCL_TEST_CFG_MAX_EPOLL_EVENTS];
 585   vcl_test_server_main_t *vsm = &vcl_server_main;
 586   vcl_test_server_worker_t *wrk = arg;
 587   vcl_test_session_t *conn;
 588   int i, rx_bytes, num_ev;
 589   vcl_test_cfg_t *rx_cfg;
 590
 591   if (wrk->wrk_index)
 592     vts_worker_init (wrk);
 593
 594   while (1)
 595     {
 596       num_ev =
 597         vppcom_epoll_wait (wrk->epfd, ep_evts, VCL_TEST_CFG_MAX_EPOLL_EVENTS,
 598                            0 /* poll session events */);
 599       if (num_ev < 0)
 600         {
 601           vterr ("vppcom_epoll_wait()", num_ev);
 602           goto fail;
 603         }
 604       else if (num_ev == 0)
 605         {
 606           continue;
 607         }
 608       for (i = 0; i < num_ev; i++)
 609         {
 610           conn = &wrk->conn_pool[ep_evts[i].data.u32];
 611           /*
 612            * Check for close events
 613            */
 614           if (ep_evts[i].events & (EPOLLHUP | EPOLLRDHUP))
 615             {
 616               vts_session_cleanup (conn);
 617               wrk->nfds--;
 618               if (!wrk->nfds)
 619                 {
 620                   vtinf ("All client connections closed\n");
 621                   goto done;
 622                 }
 623               continue;
 624             }
 625
 626           /*
 627            * Check if new session needs to be accepted
 628            */
 629
 630           if (!wrk->wrk_index && ep_evts[i].data.u32 == VCL_TEST_CTRL_LISTENER)
 631             {
 632               if (vsm->ctrl)
 633                 {
 634                   vtwrn ("ctrl already exists");
 635                   continue;
 636                 }
 637               vsm->ctrl = vts_accept_client (wrk, vsm->ctrl_listen_fd);
 638               continue;
 639             }
 640           if (ep_evts[i].data.u32 == VCL_TEST_DATA_LISTENER)
 641             {
 642               conn = vts_accept_client (wrk, wrk->listener.fd);
 643               conn->cfg = vsm->ctrl->cfg;
 644               continue;
 645             }
 646           else if (vppcom_session_is_connectable_listener (conn->fd))
 647             {
 648               vts_accept_client (wrk, conn->fd);
 649               continue;
 650             }
 651
 652           /*
 653            * Message on control session
 654            */
 655
 656           if (!wrk->wrk_index && conn->fd == vsm->ctrl->fd)
 657             {
 658               rx_bytes = conn->read (conn, conn->rxbuf, conn->rxbuf_size);
 659               rx_cfg = (vcl_test_cfg_t *) conn->rxbuf;
 660               if (rx_cfg->magic == VCL_TEST_CFG_CTRL_MAGIC)
 661                 {
 662                   vts_handle_ctrl_cfg (wrk, rx_cfg, conn, rx_bytes);
 663                   if (!wrk->nfds)
 664                     {
 665                       vtinf ("All client connections closed\n");
 666                       goto done;
 667                     }
 668                 }
 669               else if (isascii (conn->rxbuf[0]))
 670                 {
 671                   vts_server_echo (conn, rx_bytes);
 672                 }
 673               else
 674                 {
 675                   vtwrn ("FIFO not drained! extra bytes %d", rx_bytes);
 676                 }
 677               continue;
 678             }
 679
 680           /*
 681            * Read perf test data
 682            */
 683
 684           if (EPOLLIN & ep_evts[i].events)
 685             {
 686             read_again:
 687               rx_bytes = vts_conn_read (conn);
 688
 689               if (rx_bytes <= 0)
 690                 {
 691                   if (errno == ECONNRESET)
 692                     {
 693                       vtinf ("Connection reset by remote peer.\n");
 694                       goto fail;
 695                     }
 696                   else
 697                     continue;
 698                 }
 699               vts_server_process_rx (conn, rx_bytes);
 700               if (vppcom_session_attr (conn->fd, VPPCOM_ATTR_GET_NREAD, 0, 0) >
 701                   0)
 702                 goto read_again;
 703               continue;
 704             }
 705           else
 706             {
 707               vtwrn ("Unhandled event");
 708               goto fail;
 709             }
 710         }
 711     }
 712
 713 fail:
 714   vsm->worker_fails -= 1;
 715
 716 done:
 717   vppcom_session_close (wrk->listener.fd);
 718   if (wrk->conn_pool)
 719     free (wrk->conn_pool);
 720   vsm->active_workers -= 1;
 721   return 0;
 722 }
 723
 724 static void
 725 vts_ctrl_session_init (vcl_test_server_worker_t *wrk)
 726 {
 727   vcl_test_server_main_t *vsm = &vcl_server_main;
 728   struct epoll_event listen_ev;
 729   int rv;
 730
 731   vtinf ("Initializing main ctrl session ...");
 732
 733   vsm->ctrl_listen_fd =
 734     vppcom_session_create (VPPCOM_PROTO_TCP, 0 /* is_nonblocking */);
 735   if (vsm->ctrl_listen_fd < 0)
 736     vtfail ("vppcom_session_create()", vsm->ctrl_listen_fd);
 737
 738   rv = vppcom_session_bind (vsm->ctrl_listen_fd, &vsm->server_cfg.endpt);
 739   if (rv < 0)
 740     vtfail ("vppcom_session_bind()", rv);
 741
 742   rv = vppcom_session_listen (vsm->ctrl_listen_fd, 10);
 743   if (rv < 0)
 744     vtfail ("vppcom_session_listen()", rv);
 745
 746   wrk->epfd = vppcom_epoll_create ();
 747   if (wrk->epfd < 0)
 748     vtfail ("vppcom_epoll_create()", wrk->epfd);
 749
 750   listen_ev.events = EPOLLIN;
 751   listen_ev.data.u32 = VCL_TEST_CTRL_LISTENER;
 752   rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, vsm->ctrl_listen_fd,
 753                          &listen_ev);
 754   if (rv < 0)
 755     vtfail ("vppcom_epoll_ctl", rv);
 756
 757   vtinf ("Waiting for client ctrl connection on port %d ...",
 758          vsm->server_cfg.port);
 759 }
 760
 761 int
 762 main (int argc, char **argv)
 763 {
 764   vcl_test_server_main_t *vsm = &vcl_server_main;
 765   vcl_test_main_t *vt = &vcl_test_main;
 766   int rv, i;
 767
 768   clib_mem_init_thread_safe (0, 64 << 20);
 769   vsm->server_cfg.port = VCL_TEST_SERVER_PORT;
 770   vsm->server_cfg.workers = 1;
 771   vsm->active_workers = 0;
 772   vcl_test_server_process_opts (vsm, argc, argv);
 773
 774   rv = vppcom_app_create ("vcl_test_server");
 775   if (rv)
 776     vtfail ("vppcom_app_create()", rv);
 777
 778   /* Protos like tls/dtls/quic need init */
 779   if (vt->protos[vsm->server_cfg.proto]->init)
 780     vt->protos[vsm->server_cfg.proto]->init (0);
 781
 782   vsm->workers = calloc (vsm->server_cfg.workers, sizeof (*vsm->workers));
 783   vts_ctrl_session_init (&vsm->workers[0]);
 784
 785   /* Update ctrl port to data port */
 786   vsm->server_cfg.endpt.port += 1;
 787   vts_worker_init (&vsm->workers[0]);
 788   for (i = 1; i < vsm->server_cfg.workers; i++)
 789     {
 790       vsm->workers[i].wrk_index = i;
 791       rv = pthread_create (&vsm->workers[i].thread_handle, NULL,
 792                            vts_worker_loop, (void *) &vsm->workers[i]);
 793     }
 794
 795   vts_worker_loop (&vsm->workers[0]);
 796
 797   while (vsm->active_workers > 0)
 798     ;
 799
 800   vppcom_app_destroy ();
 801   free (vsm->workers);
 802
 803   return vsm->worker_fails;
 804 }
 805
 806 /*
 807  * fd.io coding-style-patch-verification: ON
 808  *
 809  * Local Variables:
 810  * eval: (c-set-style "gnu")
 811  * End:
 812  */