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   uint32_t wrk_index;
  43   vcl_test_session_t listener;
  44   int epfd;
  45   struct epoll_event wait_events[VCL_TEST_CFG_MAX_EPOLL_EVENTS];
  46   size_t conn_pool_size;
  47   vcl_test_session_t *conn_pool;
  48   int nfds;
  49   pthread_t thread_handle;
  50 } vcl_test_server_worker_t;
  51
  52 typedef struct
  53 {
  54   vcl_test_server_cfg_t cfg;
  55   vcl_test_server_worker_t *workers;
  56   vcl_test_session_t *ctrl;
  57   int ctrl_listen_fd;
  58   struct sockaddr_storage servaddr;
  59   volatile int worker_fails;
  60   volatile int active_workers;
  61   u8 use_ds;
  62 } vcl_test_server_main_t;
  63
  64 vcl_test_main_t vcl_test_main;
  65
  66 static vcl_test_server_main_t vcl_server_main;
  67
  68 static inline void
  69 conn_pool_expand (vcl_test_server_worker_t * wrk, size_t expand_size)
  70 {
  71   vcl_test_session_t *conn_pool;
  72   size_t new_size = wrk->conn_pool_size + expand_size;
  73   int i;
  74
  75   conn_pool = realloc (wrk->conn_pool, new_size * sizeof (*wrk->conn_pool));
  76   if (conn_pool)
  77     {
  78       for (i = wrk->conn_pool_size; i < new_size; i++)
  79         {
  80           vcl_test_session_t *conn = &conn_pool[i];
  81           memset (conn, 0, sizeof (*conn));
  82           vcl_test_cfg_init (&conn->cfg);
  83           vcl_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */,
  84                               (uint8_t **) &conn->rxbuf, &conn->rxbuf_size);
  85           conn->cfg.txbuf_size = conn->cfg.rxbuf_size;
  86         }
  87
  88       wrk->conn_pool = conn_pool;
  89       wrk->conn_pool_size = new_size;
  90     }
  91   else
  92     {
  93       vterr ("conn_pool_expand()", -errno);
  94     }
  95 }
  96
  97 static inline vcl_test_session_t *
  98 conn_pool_alloc (vcl_test_server_worker_t *wrk)
  99 {
 100   int i, expand = 0;
 101
 102 again:
 103   for (i = 0; i < wrk->conn_pool_size; i++)
 104     {
 105       if (!wrk->conn_pool[i].is_alloc)
 106         {
 107           wrk->conn_pool[i].endpt.ip = wrk->conn_pool[i].ip;
 108           wrk->conn_pool[i].is_alloc = 1;
 109           wrk->conn_pool[i].session_index = i;
 110           return (&wrk->conn_pool[i]);
 111         }
 112     }
 113
 114   if (expand == 0)
 115     {
 116       conn_pool_expand (wrk, 2 * wrk->conn_pool_size);
 117       expand = 1;
 118       goto again;
 119     }
 120   vtwrn ("Failed to allocate connection even after expand");
 121   return 0;
 122 }
 123
 124 static inline void
 125 conn_pool_free (vcl_test_session_t *conn)
 126 {
 127   conn->fd = 0;
 128   conn->is_alloc = 0;
 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->cfg.proto]->close)
 157     vt->protos[vsm->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 *conn)
 165 {
 166   vts_session_close (conn);
 167   conn_pool_free (conn);
 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_close (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   tp = vcl_test_main.protos[vsm->cfg.proto];
 325   if (tp->accept (listen_fd, conn))
 326     return 0;
 327
 328   vtinf ("Got a connection -- fd = %d (0x%08x) on listener fd = %d (0x%08x)",
 329          conn->fd, conn->fd, listen_fd, listen_fd);
 330
 331   ev.events = EPOLLIN;
 332   ev.data.u64 = conn - wrk->conn_pool;
 333   rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, conn->fd, &ev);
 334   if (rv < 0)
 335     {
 336       vterr ("vppcom_epoll_ctl()", rv);
 337       return 0;
 338     }
 339   wrk->nfds++;
 340
 341   return conn;
 342 }
 343
 344 static void
 345 print_usage_and_exit (void)
 346 {
 347   fprintf (stderr,
 348            "vcl_test_server [OPTIONS] <port>\n"
 349            "  OPTIONS\n"
 350            "  -h               Print this message and exit.\n"
 351            "  -6               Use IPv6\n"
 352            "  -w <num>         Number of workers\n"
 353            "  -p <PROTO>       Use <PROTO> transport layer\n"
 354            "  -D               Use UDP transport layer\n"
 355            "  -L               Use TLS transport layer\n");
 356   exit (1);
 357 }
 358
 359 static void
 360 vcl_test_init_endpoint_addr (vcl_test_server_main_t * vsm)
 361 {
 362   struct sockaddr_storage *servaddr = &vsm->servaddr;
 363   memset (servaddr, 0, sizeof (*servaddr));
 364
 365   if (vsm->cfg.address_ip6)
 366     {
 367       struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr;
 368       server_addr->sin6_family = AF_INET6;
 369       server_addr->sin6_addr = in6addr_any;
 370       server_addr->sin6_port = htons (vsm->cfg.port);
 371     }
 372   else
 373     {
 374       struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr;
 375       server_addr->sin_family = AF_INET;
 376       server_addr->sin_addr.s_addr = htonl (INADDR_ANY);
 377       server_addr->sin_port = htons (vsm->cfg.port);
 378     }
 379
 380   if (vsm->cfg.address_ip6)
 381     {
 382       struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr;
 383       vsm->cfg.endpt.is_ip4 = 0;
 384       vsm->cfg.endpt.ip = (uint8_t *) & server_addr->sin6_addr;
 385       vsm->cfg.endpt.port = (uint16_t) server_addr->sin6_port;
 386     }
 387   else
 388     {
 389       struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr;
 390       vsm->cfg.endpt.is_ip4 = 1;
 391       vsm->cfg.endpt.ip = (uint8_t *) & server_addr->sin_addr;
 392       vsm->cfg.endpt.port = (uint16_t) server_addr->sin_port;
 393     }
 394 }
 395
 396 static void
 397 vcl_test_server_process_opts (vcl_test_server_main_t * vsm, int argc,
 398                               char **argv)
 399 {
 400   int v, c;
 401
 402   vsm->cfg.proto = VPPCOM_PROTO_TCP;
 403
 404   opterr = 0;
 405   while ((c = getopt (argc, argv, "6DLsw:hp:")) != -1)
 406     switch (c)
 407       {
 408       case '6':
 409         vsm->cfg.address_ip6 = 1;
 410         break;
 411
 412       case 'p':
 413         if (vppcom_unformat_proto (&vsm->cfg.proto, optarg))
 414           vtwrn ("Invalid vppcom protocol %s, defaulting to TCP", optarg);
 415         break;
 416
 417       case 'D':
 418         vsm->cfg.proto = VPPCOM_PROTO_UDP;
 419         break;
 420
 421       case 'L':
 422         vsm->cfg.proto = VPPCOM_PROTO_TLS;
 423         break;
 424
 425       case 'w':
 426         v = atoi (optarg);
 427         if (v > 1)
 428           vsm->cfg.workers = v;
 429         else
 430           vtwrn ("Invalid number of workers %d", v);
 431         break;
 432       case 's':
 433         vsm->use_ds = 1;
 434         break;
 435       case '?':
 436         switch (optopt)
 437           {
 438           case 'w':
 439           case 'p':
 440             vtwrn ("Option `-%c' requires an argument.", optopt);
 441             break;
 442           default:
 443             if (isprint (optopt))
 444               vtwrn ("Unknown option `-%c'.", optopt);
 445             else
 446               vtwrn ("Unknown option character `\\x%x'.", optopt);
 447           }
 448         /* fall thru */
 449       case 'h':
 450       default:
 451         print_usage_and_exit ();
 452       }
 453
 454   if (argc < (optind + 1))
 455     {
 456       fprintf (stderr, "SERVER: ERROR: Insufficient number of arguments!\n");
 457       print_usage_and_exit ();
 458     }
 459
 460   if (sscanf (argv[optind], "%d", &v) == 1)
 461     vsm->cfg.port = (uint16_t) v;
 462   else
 463     {
 464       fprintf (stderr, "SERVER: ERROR: Invalid port (%s)!\n", argv[optind]);
 465       print_usage_and_exit ();
 466     }
 467
 468   vcl_test_init_endpoint_addr (vsm);
 469 }
 470
 471 int
 472 vts_handle_ctrl_cfg (vcl_test_server_worker_t *wrk, vcl_test_cfg_t *rx_cfg,
 473                      vcl_test_session_t *conn, int rx_bytes)
 474 {
 475   if (rx_cfg->verbose)
 476     {
 477       vtinf ("(fd %d): Received a cfg msg!", conn->fd);
 478       vcl_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 479     }
 480
 481   if (rx_bytes != sizeof (*rx_cfg))
 482     {
 483       vtinf ("(fd %d): Invalid cfg msg size %d expected %lu!", conn->fd,
 484              rx_bytes, sizeof (*rx_cfg));
 485       conn->cfg.rxbuf_size = 0;
 486       conn->cfg.num_writes = 0;
 487       if (conn->cfg.verbose)
 488         {
 489           vtinf ("(fd %d): Replying to cfg msg", conn->fd);
 490           vcl_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 491         }
 492       conn->write (conn, &conn->cfg, sizeof (conn->cfg));
 493       return -1;
 494     }
 495
 496   switch (rx_cfg->test)
 497     {
 498     case VCL_TEST_TYPE_NONE:
 499     case VCL_TEST_TYPE_ECHO:
 500       sync_config_and_reply (conn, rx_cfg);
 501       break;
 502
 503     case VCL_TEST_TYPE_BI:
 504     case VCL_TEST_TYPE_UNI:
 505       vts_test_cmd (wrk, conn, rx_cfg);
 506       break;
 507
 508     case VCL_TEST_TYPE_EXIT:
 509       vtinf ("Ctrl session fd %d closing!", conn->fd);
 510       vts_session_cleanup (conn);
 511       wrk->nfds--;
 512       if (wrk->nfds)
 513         vts_wrk_cleanup_all (wrk);
 514       break;
 515
 516     default:
 517       vtwrn ("Unknown test type %d", rx_cfg->test);
 518       vcl_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 519       break;
 520     }
 521
 522   return 0;
 523 }
 524
 525 static void
 526 vts_worker_init (vcl_test_server_worker_t * wrk)
 527 {
 528   vcl_test_server_main_t *vsm = &vcl_server_main;
 529   vcl_test_main_t *vt = &vcl_test_main;
 530   const vcl_test_proto_vft_t *tp;
 531   struct epoll_event listen_ev;
 532   int rv;
 533
 534   __wrk_index = wrk->wrk_index;
 535
 536   vtinf ("Initializing worker ...");
 537
 538   conn_pool_expand (wrk, VCL_TEST_CFG_MAX_TEST_SESS + 1);
 539   if (wrk->wrk_index)
 540     if (vppcom_worker_register ())
 541       vtfail ("vppcom_worker_register()", 1);
 542
 543   tp = vt->protos[vsm->cfg.proto];
 544   if ((rv = tp->listen (&wrk->listener, &vsm->cfg.endpt)))
 545     vtfail ("proto listen", rv);
 546
 547   /* First worker already has epoll fd */
 548   if (wrk->wrk_index)
 549     {
 550       wrk->epfd = vppcom_epoll_create ();
 551       if (wrk->epfd < 0)
 552         vtfail ("vppcom_epoll_create()", wrk->epfd);
 553     }
 554
 555   listen_ev.events = EPOLLIN;
 556   listen_ev.data.u32 = VCL_TEST_DATA_LISTENER;
 557   rv =
 558     vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, wrk->listener.fd, &listen_ev);
 559   if (rv < 0)
 560     vtfail ("vppcom_epoll_ctl", rv);
 561
 562   vsm->active_workers += 1;
 563   vtinf ("Waiting for client data connections on port %d ...",
 564          ntohs (vsm->cfg.endpt.port));
 565 }
 566
 567 static inline int
 568 vts_conn_read (vcl_test_session_t *conn)
 569 {
 570   vcl_test_server_main_t *vsm = &vcl_server_main;
 571   if (vsm->use_ds)
 572     return vcl_test_read_ds (conn);
 573   else
 574     return conn->read (conn, conn->rxbuf, conn->rxbuf_size);
 575 }
 576
 577 static void *
 578 vts_worker_loop (void *arg)
 579 {
 580   vcl_test_server_main_t *vsm = &vcl_server_main;
 581   vcl_test_server_worker_t *wrk = arg;
 582   vcl_test_session_t *conn;
 583   int i, rx_bytes, num_ev;
 584   vcl_test_cfg_t *rx_cfg;
 585
 586   if (wrk->wrk_index)
 587     vts_worker_init (wrk);
 588
 589   while (1)
 590     {
 591       num_ev = vppcom_epoll_wait (wrk->epfd, wrk->wait_events,
 592                                   VCL_TEST_CFG_MAX_EPOLL_EVENTS, 60000.0);
 593       if (num_ev < 0)
 594         {
 595           vterr ("vppcom_epoll_wait()", num_ev);
 596           goto fail;
 597         }
 598       else if (num_ev == 0)
 599         {
 600           vtinf ("vppcom_epoll_wait() timeout!");
 601           continue;
 602         }
 603       for (i = 0; i < num_ev; i++)
 604         {
 605           conn = &wrk->conn_pool[wrk->wait_events[i].data.u32];
 606           /*
 607            * Check for close events
 608            */
 609           if (wrk->wait_events[i].events & (EPOLLHUP | EPOLLRDHUP))
 610             {
 611               vts_session_close (conn);
 612               wrk->nfds--;
 613               if (!wrk->nfds)
 614                 {
 615                   vtinf ("All client connections closed\n");
 616                   goto done;
 617                 }
 618               continue;
 619             }
 620
 621           /*
 622            * Check if new session needs to be accepted
 623            */
 624
 625           if (wrk->wait_events[i].data.u32 == VCL_TEST_CTRL_LISTENER)
 626             {
 627               if (vsm->ctrl)
 628                 {
 629                   vtwrn ("ctrl already exists");
 630                   continue;
 631                 }
 632               vsm->ctrl = vts_accept_client (wrk, vsm->ctrl_listen_fd);
 633               continue;
 634             }
 635           if (wrk->wait_events[i].data.u32 == VCL_TEST_DATA_LISTENER)
 636             {
 637               conn = vts_accept_client (wrk, wrk->listener.fd);
 638               conn->cfg = vsm->ctrl->cfg;
 639               continue;
 640             }
 641           else if (vppcom_session_is_connectable_listener (conn->fd))
 642             {
 643               vts_accept_client (wrk, conn->fd);
 644               continue;
 645             }
 646
 647           /*
 648            * Message on control session
 649            */
 650
 651           if (!wrk->wrk_index && conn->fd == vsm->ctrl->fd)
 652             {
 653               rx_bytes = conn->read (conn, conn->rxbuf, conn->rxbuf_size);
 654               rx_cfg = (vcl_test_cfg_t *) conn->rxbuf;
 655               if (rx_cfg->magic == VCL_TEST_CFG_CTRL_MAGIC)
 656                 {
 657                   vts_handle_ctrl_cfg (wrk, rx_cfg, conn, rx_bytes);
 658                   if (!wrk->nfds)
 659                     {
 660                       vtinf ("All client connections closed\n");
 661                       goto done;
 662                     }
 663                 }
 664               else if (isascii (conn->rxbuf[0]))
 665                 {
 666                   vts_server_echo (conn, rx_bytes);
 667                 }
 668               else
 669                 {
 670                   vtwrn ("FIFO not drained! extra bytes %d", rx_bytes);
 671                 }
 672               continue;
 673             }
 674
 675           /*
 676            * Read perf test data
 677            */
 678
 679           if (EPOLLIN & wrk->wait_events[i].events)
 680             {
 681             read_again:
 682               rx_bytes = vts_conn_read (conn);
 683
 684               if (rx_bytes <= 0)
 685                 {
 686                   if (errno == ECONNRESET)
 687                     {
 688                       vtinf ("Connection reset by remote peer.\n");
 689                       goto fail;
 690                     }
 691                   else
 692                     continue;
 693                 }
 694               vts_server_process_rx (conn, rx_bytes);
 695               if (vppcom_session_attr (conn->fd, VPPCOM_ATTR_GET_NREAD, 0, 0) >
 696                   0)
 697                 goto read_again;
 698               continue;
 699             }
 700           else
 701             {
 702               vtwrn ("Unhandled event");
 703               goto fail;
 704             }
 705         }
 706     }
 707
 708 fail:
 709   vsm->worker_fails -= 1;
 710
 711 done:
 712   vppcom_session_close (wrk->listener.fd);
 713   if (wrk->conn_pool)
 714     free (wrk->conn_pool);
 715   vsm->active_workers -= 1;
 716   return 0;
 717 }
 718
 719 static void
 720 vts_ctrl_session_init (vcl_test_server_worker_t *wrk)
 721 {
 722   vcl_test_server_main_t *vsm = &vcl_server_main;
 723   struct epoll_event listen_ev;
 724   int rv;
 725
 726   vtinf ("Initializing main ctrl session ...");
 727
 728   vsm->ctrl_listen_fd =
 729     vppcom_session_create (VPPCOM_PROTO_TCP, 0 /* is_nonblocking */);
 730   if (vsm->ctrl_listen_fd < 0)
 731     vtfail ("vppcom_session_create()", vsm->ctrl_listen_fd);
 732
 733   rv = vppcom_session_bind (vsm->ctrl_listen_fd, &vsm->cfg.endpt);
 734   if (rv < 0)
 735     vtfail ("vppcom_session_bind()", rv);
 736
 737   rv = vppcom_session_listen (vsm->ctrl_listen_fd, 10);
 738   if (rv < 0)
 739     vtfail ("vppcom_session_listen()", rv);
 740
 741   wrk->epfd = vppcom_epoll_create ();
 742   if (wrk->epfd < 0)
 743     vtfail ("vppcom_epoll_create()", wrk->epfd);
 744
 745   listen_ev.events = EPOLLIN;
 746   listen_ev.data.u32 = VCL_TEST_CTRL_LISTENER;
 747   rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, vsm->ctrl_listen_fd,
 748                          &listen_ev);
 749   if (rv < 0)
 750     vtfail ("vppcom_epoll_ctl", rv);
 751
 752   vtinf ("Waiting for client ctrl connection on port %d ...", vsm->cfg.port);
 753 }
 754
 755 int
 756 main (int argc, char **argv)
 757 {
 758   vcl_test_server_main_t *vsm = &vcl_server_main;
 759   vcl_test_main_t *vt = &vcl_test_main;
 760   int rv, i;
 761
 762   clib_mem_init_thread_safe (0, 64 << 20);
 763   vsm->cfg.port = VCL_TEST_SERVER_PORT;
 764   vsm->cfg.workers = 1;
 765   vsm->active_workers = 0;
 766   vcl_test_server_process_opts (vsm, argc, argv);
 767
 768   rv = vppcom_app_create ("vcl_test_server");
 769   if (rv)
 770     vtfail ("vppcom_app_create()", rv);
 771
 772   /* Protos like tls/dtls/quic need init */
 773   if (vt->protos[vsm->cfg.proto]->init)
 774     vt->protos[vsm->cfg.proto]->init (0);
 775
 776   vsm->workers = calloc (vsm->cfg.workers, sizeof (*vsm->workers));
 777   vts_ctrl_session_init (&vsm->workers[0]);
 778
 779   /* Update ctrl port to data port */
 780   vsm->cfg.endpt.port += 1;
 781   vts_worker_init (&vsm->workers[0]);
 782   for (i = 1; i < vsm->cfg.workers; i++)
 783     {
 784       vsm->workers[i].wrk_index = i;
 785       rv = pthread_create (&vsm->workers[i].thread_handle, NULL,
 786                            vts_worker_loop, (void *) &vsm->workers[i]);
 787     }
 788
 789   vts_worker_loop (&vsm->workers[0]);
 790
 791   while (vsm->active_workers > 0)
 792     ;
 793
 794   vppcom_app_destroy ();
 795   free (vsm->workers);
 796
 797   return vsm->worker_fails;
 798 }
 799
 800 /*
 801  * fd.io coding-style-patch-verification: ON
 802  *
 803  * Local Variables:
 804  * eval: (c-set-style "gnu")
 805  * End:
 806  */