src/vcl/vcl_test_server.c

   1 /*
   2  * Copyright (c) 2017-2018 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 <vcl/vcl_test.h>
  27 #include <sys/epoll.h>
  28 #include <vppinfra/mem.h>
  29 #include <pthread.h>
  30
  31 typedef struct
  32 {
  33   uint8_t is_alloc;
  34   int fd;
  35   uint8_t *buf;
  36   uint32_t buf_size;
  37   sock_test_cfg_t cfg;
  38   sock_test_stats_t stats;
  39   vppcom_endpt_t endpt;
  40   uint8_t ip[16];
  41 } vcl_test_server_conn_t;
  42
  43 typedef struct
  44 {
  45   uint16_t port;
  46   uint32_t address_ip6;
  47   u8 proto;
  48   u8 workers;
  49   vppcom_endpt_t endpt;
  50 } vcl_test_server_cfg_t;
  51
  52 #define SOCK_SERVER_MAX_TEST_CONN  16
  53 #define SOCK_SERVER_MAX_EPOLL_EVENTS 16
  54
  55 typedef struct
  56 {
  57   uint32_t wrk_index;
  58   int listen_fd;
  59   int epfd;
  60   struct epoll_event wait_events[SOCK_SERVER_MAX_EPOLL_EVENTS];
  61   size_t conn_pool_size;
  62   vcl_test_server_conn_t *conn_pool;
  63   int nfds;
  64   pthread_t thread_handle;
  65 } vcl_test_server_worker_t;
  66
  67 typedef struct
  68 {
  69   vcl_test_server_cfg_t cfg;
  70   vcl_test_server_worker_t *workers;
  71
  72   struct sockaddr_storage servaddr;
  73   volatile int worker_fails;
  74   volatile int active_workers;
  75 } vcl_test_server_main_t;
  76
  77 static __thread int __wrk_index = 0;
  78
  79 static vcl_test_server_main_t sock_server_main;
  80
  81 static inline void
  82 conn_pool_expand (vcl_test_server_worker_t * wrk, size_t expand_size)
  83 {
  84   vcl_test_server_conn_t *conn_pool;
  85   size_t new_size = wrk->conn_pool_size + expand_size;
  86   int i;
  87
  88   conn_pool = realloc (wrk->conn_pool, new_size * sizeof (*wrk->conn_pool));
  89   if (conn_pool)
  90     {
  91       for (i = wrk->conn_pool_size; i < new_size; i++)
  92         {
  93           vcl_test_server_conn_t *conn = &conn_pool[i];
  94           memset (conn, 0, sizeof (*conn));
  95           sock_test_cfg_init (&conn->cfg);
  96           sock_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */ ,
  97                                &conn->buf, &conn->buf_size);
  98           conn->cfg.txbuf_size = conn->cfg.rxbuf_size;
  99         }
 100
 101       wrk->conn_pool = conn_pool;
 102       wrk->conn_pool_size = new_size;
 103     }
 104   else
 105     {
 106       vterr ("conn_pool_expand()", -errno);
 107     }
 108 }
 109
 110 static inline vcl_test_server_conn_t *
 111 conn_pool_alloc (vcl_test_server_worker_t * wrk)
 112 {
 113   int i;
 114
 115   for (i = 0; i < wrk->conn_pool_size; i++)
 116     {
 117       if (!wrk->conn_pool[i].is_alloc)
 118         {
 119           wrk->conn_pool[i].endpt.ip = wrk->conn_pool[i].ip;
 120           wrk->conn_pool[i].is_alloc = 1;
 121           return (&wrk->conn_pool[i]);
 122         }
 123     }
 124
 125   return 0;
 126 }
 127
 128 static inline void
 129 conn_pool_free (vcl_test_server_conn_t * conn)
 130 {
 131   conn->fd = 0;
 132   conn->is_alloc = 0;
 133 }
 134
 135 static inline void
 136 sync_config_and_reply (vcl_test_server_conn_t * conn,
 137                        sock_test_cfg_t * rx_cfg)
 138 {
 139   conn->cfg = *rx_cfg;
 140   sock_test_buf_alloc (&conn->cfg, 1 /* is_rxbuf */ ,
 141                        &conn->buf, &conn->buf_size);
 142   conn->cfg.txbuf_size = conn->cfg.rxbuf_size;
 143
 144   if (conn->cfg.verbose)
 145     {
 146       vtinf ("(fd %d): Replying to cfg message!\n", conn->fd);
 147       sock_test_cfg_dump (&conn->cfg, 0 /* is_client */ );
 148     }
 149   (void) vcl_test_write (conn->fd, (uint8_t *) & conn->cfg,
 150                          sizeof (conn->cfg), NULL, conn->cfg.verbose);
 151 }
 152
 153 static void
 154 vts_server_start_stop (vcl_test_server_worker_t * wrk,
 155                        vcl_test_server_conn_t * conn,
 156                        sock_test_cfg_t * rx_cfg)
 157 {
 158   u8 is_bi = rx_cfg->test == SOCK_TEST_TYPE_BI;
 159   int client_fd = conn->fd, i;
 160   vcl_test_server_conn_t *tc;
 161   char buf[64];
 162
 163   if (rx_cfg->ctrl_handle == conn->fd)
 164     {
 165       clock_gettime (CLOCK_REALTIME, &conn->stats.stop);
 166
 167       for (i = 0; i < wrk->conn_pool_size; i++)
 168         {
 169           tc = &wrk->conn_pool[i];
 170           if (tc->cfg.ctrl_handle != conn->fd)
 171             continue;
 172
 173           sock_test_stats_accumulate (&conn->stats, &tc->stats);
 174
 175           if (conn->cfg.verbose)
 176             {
 177               sprintf (buf, "SERVER (fd %d) RESULTS", tc->fd);
 178               sock_test_stats_dump (buf, &tc->stats, 1 /* show_rx */ ,
 179                                     is_bi /* show tx */ , conn->cfg.verbose);
 180             }
 181         }
 182
 183       sock_test_stats_dump ("SERVER RESULTS", &conn->stats, 1 /* show_rx */ ,
 184                             is_bi /* show_tx */ , conn->cfg.verbose);
 185       sock_test_cfg_dump (&conn->cfg, 0 /* is_client */ );
 186       if (conn->cfg.verbose)
 187         {
 188           vtinf ("  sock server main\n"
 189                  SOCK_TEST_SEPARATOR_STRING
 190                  "       buf:  %p\n"
 191                  "  buf size:  %u (0x%08x)\n"
 192                  SOCK_TEST_SEPARATOR_STRING,
 193                  conn->buf, conn->buf_size, conn->buf_size);
 194         }
 195
 196       sync_config_and_reply (conn, rx_cfg);
 197       vtinf ("(fd %d): %s-directional Stream Test Complete!\n"
 198              SOCK_TEST_BANNER_STRING "\n", conn->fd, is_bi ? "Bi" : "Uni");
 199     }
 200   else
 201     {
 202       vtinf (SOCK_TEST_BANNER_STRING "(fd %d): %s-directional Stream Test!\n"
 203              "  Sending client the test cfg to start streaming data...\n",
 204              client_fd, is_bi ? "Bi" : "Uni");
 205
 206       if (rx_cfg->ctrl_handle == ~0)
 207         rx_cfg->ctrl_handle = conn->fd;
 208
 209       sync_config_and_reply (conn, rx_cfg);
 210
 211       /* read the 1st chunk, record start time */
 212       memset (&conn->stats, 0, sizeof (conn->stats));
 213       clock_gettime (CLOCK_REALTIME, &conn->stats.start);
 214     }
 215 }
 216
 217 static inline void
 218 vts_server_rx (vcl_test_server_conn_t * conn, int rx_bytes)
 219 {
 220   int client_fd = conn->fd;
 221   sock_test_t test = conn->cfg.test;
 222
 223   if (test == SOCK_TEST_TYPE_BI)
 224     (void) vcl_test_write (client_fd, conn->buf, rx_bytes, &conn->stats,
 225                            conn->cfg.verbose);
 226
 227   if (conn->stats.rx_bytes >= conn->cfg.total_bytes)
 228     clock_gettime (CLOCK_REALTIME, &conn->stats.stop);
 229 }
 230
 231 static void
 232 vts_server_echo (vcl_test_server_conn_t * conn, int rx_bytes)
 233 {
 234   int tx_bytes, nbytes, pos;
 235
 236   /* If it looks vaguely like a string,
 237    * make sure it's terminated
 238    */
 239   pos = rx_bytes < conn->buf_size ? rx_bytes : conn->buf_size - 1;
 240   ((char *) conn->buf)[pos] = 0;
 241   vtinf ("(fd %d): RX (%d bytes) - '%s'", conn->fd, rx_bytes, conn->buf);
 242
 243   if (conn->cfg.verbose)
 244     vtinf ("(fd %d): Echoing back", conn->fd);
 245
 246   nbytes = strlen ((const char *) conn->buf) + 1;
 247   tx_bytes = vcl_test_write (conn->fd, conn->buf, nbytes, &conn->stats,
 248                              conn->cfg.verbose);
 249   if (tx_bytes >= 0)
 250     vtinf ("(fd %d): TX (%d bytes) - '%s'", conn->fd, tx_bytes, conn->buf);
 251 }
 252
 253 static inline void
 254 vts_new_client (vcl_test_server_worker_t * wrk)
 255 {
 256   int client_fd;
 257   vcl_test_server_conn_t *conn;
 258
 259   conn = conn_pool_alloc (wrk);
 260   if (!conn)
 261     {
 262       vtwrn ("No free connections!");
 263       return;
 264     }
 265
 266   client_fd = vppcom_session_accept (wrk->listen_fd, &conn->endpt, 0);
 267   if (client_fd < 0)
 268     {
 269       vterr ("vppcom_session_accept()", client_fd);
 270       return;
 271     }
 272
 273   vtinf ("Got a connection -- fd = %d (0x%08x)!", client_fd, client_fd);
 274
 275   conn->fd = client_fd;
 276
 277   {
 278     struct epoll_event ev;
 279     int rv;
 280
 281     ev.events = EPOLLIN;
 282     ev.data.u64 = conn - wrk->conn_pool;
 283     rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, client_fd, &ev);
 284     if (rv < 0)
 285       {
 286         vterr ("vppcom_epoll_ctl()", rv);
 287         return;
 288       }
 289     wrk->nfds++;
 290   }
 291 }
 292
 293 void
 294 print_usage_and_exit (void)
 295 {
 296   fprintf (stderr,
 297            "sock_test_server [OPTIONS] <port>\n"
 298            "  OPTIONS\n"
 299            "  -h               Print this message and exit.\n"
 300            "  -6               Use IPv6\n"
 301            "  -w <num>         Number of workers\n"
 302            "  -u               Use UDP transport layer\n");
 303   exit (1);
 304 }
 305
 306 static void
 307 vcl_test_init_endpoint_addr (vcl_test_server_main_t * ssm)
 308 {
 309   struct sockaddr_storage *servaddr = &ssm->servaddr;
 310   memset (servaddr, 0, sizeof (*servaddr));
 311
 312   if (ssm->cfg.address_ip6)
 313     {
 314       struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr;
 315       server_addr->sin6_family = AF_INET6;
 316       server_addr->sin6_addr = in6addr_any;
 317       server_addr->sin6_port = htons (ssm->cfg.port);
 318     }
 319   else
 320     {
 321       struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr;
 322       server_addr->sin_family = AF_INET;
 323       server_addr->sin_addr.s_addr = htonl (INADDR_ANY);
 324       server_addr->sin_port = htons (ssm->cfg.port);
 325     }
 326
 327   if (ssm->cfg.address_ip6)
 328     {
 329       struct sockaddr_in6 *server_addr = (struct sockaddr_in6 *) servaddr;
 330       ssm->cfg.endpt.is_ip4 = 0;
 331       ssm->cfg.endpt.ip = (uint8_t *) & server_addr->sin6_addr;
 332       ssm->cfg.endpt.port = (uint16_t) server_addr->sin6_port;
 333     }
 334   else
 335     {
 336       struct sockaddr_in *server_addr = (struct sockaddr_in *) servaddr;
 337       ssm->cfg.endpt.is_ip4 = 1;
 338       ssm->cfg.endpt.ip = (uint8_t *) & server_addr->sin_addr;
 339       ssm->cfg.endpt.port = (uint16_t) server_addr->sin_port;
 340     }
 341 }
 342
 343 static void
 344 vcl_test_server_process_opts (vcl_test_server_main_t * ssm, int argc,
 345                               char **argv)
 346 {
 347   int v, c;
 348
 349   ssm->cfg.proto = VPPCOM_PROTO_TCP;
 350
 351   opterr = 0;
 352   while ((c = getopt (argc, argv, "6Dw:")) != -1)
 353     switch (c)
 354       {
 355       case '6':
 356         ssm->cfg.address_ip6 = 1;
 357         break;
 358
 359       case 'D':
 360         ssm->cfg.proto = VPPCOM_PROTO_UDP;
 361         break;
 362
 363       case 'w':
 364         v = atoi (optarg);
 365         if (v > 1)
 366           ssm->cfg.workers = v;
 367         else
 368           vtwrn ("Invalid number of workers %d", v);
 369         break;
 370
 371       case '?':
 372         switch (optopt)
 373           {
 374           default:
 375             if (isprint (optopt))
 376               vtwrn ("Unknown option `-%c'.", optopt);
 377             else
 378               vtwrn ("Unknown option character `\\x%x'.", optopt);
 379           }
 380         /* fall thru */
 381       case 'h':
 382       default:
 383         print_usage_and_exit ();
 384       }
 385
 386   if (argc < (optind + 1))
 387     {
 388       fprintf (stderr, "SERVER: ERROR: Insufficient number of arguments!\n");
 389       print_usage_and_exit ();
 390     }
 391
 392   if (sscanf (argv[optind], "%d", &v) == 1)
 393     ssm->cfg.port = (uint16_t) v;
 394   else
 395     {
 396       fprintf (stderr, "SERVER: ERROR: Invalid port (%s)!\n", argv[optind]);
 397       print_usage_and_exit ();
 398     }
 399
 400   vcl_test_init_endpoint_addr (ssm);
 401 }
 402
 403 int
 404 vts_handle_cfg (vcl_test_server_worker_t * wrk, sock_test_cfg_t * rx_cfg,
 405                 vcl_test_server_conn_t * conn, int rx_bytes)
 406 {
 407   if (rx_cfg->verbose)
 408     {
 409       vtinf ("(fd %d): Received a cfg msg!", conn->fd);
 410       sock_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 411     }
 412
 413   if (rx_bytes != sizeof (*rx_cfg))
 414     {
 415       vtinf ("(fd %d): Invalid cfg msg size %d expected %lu!", conn->fd,
 416              rx_bytes, sizeof (*rx_cfg));
 417       conn->cfg.rxbuf_size = 0;
 418       conn->cfg.num_writes = 0;
 419       if (conn->cfg.verbose)
 420         {
 421           vtinf ("(fd %d): Replying to cfg msg", conn->fd);
 422           sock_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 423         }
 424       vcl_test_write (conn->fd, (uint8_t *) & conn->cfg,
 425                       sizeof (conn->cfg), NULL, conn->cfg.verbose);
 426       return -1;
 427     }
 428
 429   switch (rx_cfg->test)
 430     {
 431     case SOCK_TEST_TYPE_NONE:
 432     case SOCK_TEST_TYPE_ECHO:
 433       sync_config_and_reply (conn, rx_cfg);
 434       break;
 435
 436     case SOCK_TEST_TYPE_BI:
 437     case SOCK_TEST_TYPE_UNI:
 438       vts_server_start_stop (wrk, conn, rx_cfg);
 439       break;
 440
 441     case SOCK_TEST_TYPE_EXIT:
 442       vtinf ("Have a great day conn %d (closing)!", conn->fd);
 443       vppcom_session_close (conn->fd);
 444       conn_pool_free (conn);
 445       wrk->nfds--;
 446       break;
 447
 448     default:
 449       vtwrn ("Unknown test type %d", rx_cfg->test);
 450       sock_test_cfg_dump (rx_cfg, 0 /* is_client */ );
 451       break;
 452     }
 453
 454   return 0;
 455 }
 456
 457 static void
 458 vts_worker_init (vcl_test_server_worker_t * wrk)
 459 {
 460   vcl_test_server_main_t *ssm = &sock_server_main;
 461   struct epoll_event listen_ev;
 462   int rv;
 463
 464   __wrk_index = wrk->wrk_index;
 465
 466   vtinf ("Initializing worker ...");
 467
 468   conn_pool_expand (wrk, SOCK_SERVER_MAX_TEST_CONN + 1);
 469   if (wrk->wrk_index)
 470     if (vppcom_worker_register ())
 471       vtfail ("vppcom_worker_register()", 1);
 472
 473   wrk->listen_fd = vppcom_session_create (ssm->cfg.proto,
 474                                           0 /* is_nonblocking */ );
 475   if (wrk->listen_fd < 0)
 476     vtfail ("vppcom_session_create()", wrk->listen_fd);
 477
 478   rv = vppcom_session_bind (wrk->listen_fd, &ssm->cfg.endpt);
 479   if (rv < 0)
 480     vtfail ("vppcom_session_bind()", rv);
 481
 482   if (!(ssm->cfg.proto == VPPCOM_PROTO_UDP))
 483     {
 484       rv = vppcom_session_listen (wrk->listen_fd, 10);
 485       if (rv < 0)
 486         vtfail ("vppcom_session_listen()", rv);
 487     }
 488
 489   wrk->epfd = vppcom_epoll_create ();
 490   if (wrk->epfd < 0)
 491     vtfail ("vppcom_epoll_create()", wrk->epfd);
 492
 493   listen_ev.events = EPOLLIN;
 494   listen_ev.data.u32 = ~0;
 495   rv = vppcom_epoll_ctl (wrk->epfd, EPOLL_CTL_ADD, wrk->listen_fd,
 496                          &listen_ev);
 497   if (rv < 0)
 498     vtfail ("vppcom_epoll_ctl", rv);
 499
 500   ssm->active_workers += 1;
 501   vtinf ("Waiting for a client to connect on port %d ...", ssm->cfg.port);
 502 }
 503
 504 static void *
 505 vts_worker_loop (void *arg)
 506 {
 507   vcl_test_server_main_t *ssm = &sock_server_main;
 508   vcl_test_server_worker_t *wrk = arg;
 509   vcl_test_server_conn_t *conn;
 510   int i, rx_bytes, num_ev;
 511   sock_test_cfg_t *rx_cfg;
 512
 513   if (wrk->wrk_index)
 514     vts_worker_init (wrk);
 515
 516   while (1)
 517     {
 518       num_ev = vppcom_epoll_wait (wrk->epfd, wrk->wait_events,
 519                                   SOCK_SERVER_MAX_EPOLL_EVENTS, 60000.0);
 520       if (num_ev < 0)
 521         {
 522           vterr ("vppcom_epoll_wait()", num_ev);
 523           goto fail;
 524         }
 525       else if (num_ev == 0)
 526         {
 527           vtinf ("vppcom_epoll_wait() timeout!");
 528           continue;
 529         }
 530       for (i = 0; i < num_ev; i++)
 531         {
 532           conn = &wrk->conn_pool[wrk->wait_events[i].data.u32];
 533           if (wrk->wait_events[i].events & (EPOLLHUP | EPOLLRDHUP))
 534             {
 535               vppcom_session_close (conn->fd);
 536               wrk->nfds -= 1;
 537               if (!wrk->nfds)
 538                 {
 539                   vtinf ("All client connections closed\n");
 540                   goto done;
 541                 }
 542               continue;
 543             }
 544           if (wrk->wait_events[i].data.u32 == ~0)
 545             {
 546               vts_new_client (wrk);
 547               continue;
 548             }
 549
 550           if (EPOLLIN & wrk->wait_events[i].events)
 551             {
 552             read_again:
 553               rx_bytes = vcl_test_read (conn->fd, conn->buf,
 554                                         conn->buf_size, &conn->stats);
 555
 556               if (rx_bytes <= 0)
 557                 {
 558                   if (errno == ECONNRESET)
 559                     {
 560                       vtinf ("Connection reset by remote peer.\n");
 561                       goto fail;
 562                     }
 563                   else
 564                     continue;
 565                 }
 566
 567               rx_cfg = (sock_test_cfg_t *) conn->buf;
 568               if (rx_cfg->magic == SOCK_TEST_CFG_CTRL_MAGIC)
 569                 {
 570                   vts_handle_cfg (wrk, rx_cfg, conn, rx_bytes);
 571                   if (!wrk->nfds)
 572                     {
 573                       vtinf ("All client connections closed\n");
 574                       goto done;
 575                     }
 576                   continue;
 577                 }
 578               if ((conn->cfg.test == SOCK_TEST_TYPE_UNI)
 579                   || (conn->cfg.test == SOCK_TEST_TYPE_BI))
 580                 {
 581                   vts_server_rx (conn, rx_bytes);
 582                   if (vppcom_session_attr (conn->fd, VPPCOM_ATTR_GET_NREAD, 0,
 583                                            0) > 0)
 584                     goto read_again;
 585                   continue;
 586                 }
 587               if (isascii (conn->buf[0]))
 588                 {
 589                   vts_server_echo (conn, rx_bytes);
 590                 }
 591               else
 592                 {
 593                   vtwrn ("FIFO not drained! extra bytes %d", rx_bytes);
 594                 }
 595             }
 596           else
 597             {
 598               vtwrn ("Unhandled event");
 599               goto fail;
 600             }
 601         }
 602     }
 603
 604 fail:
 605   ssm->worker_fails -= 1;
 606
 607 done:
 608   vppcom_session_close (wrk->listen_fd);
 609   if (wrk->conn_pool)
 610     free (wrk->conn_pool);
 611   ssm->active_workers -= 1;
 612   return 0;
 613 }
 614
 615 int
 616 main (int argc, char **argv)
 617 {
 618   vcl_test_server_main_t *vsm = &sock_server_main;
 619   int rv, i;
 620
 621   clib_mem_init_thread_safe (0, 64 << 20);
 622   vsm->cfg.port = SOCK_TEST_SERVER_PORT;
 623   vsm->cfg.workers = 1;
 624   vsm->active_workers = 1;
 625   vcl_test_server_process_opts (vsm, argc, argv);
 626
 627   rv = vppcom_app_create ("vcl_test_server");
 628   if (rv)
 629     vtfail ("vppcom_app_create()", rv);
 630
 631   vsm->workers = calloc (vsm->cfg.workers, sizeof (*vsm->workers));
 632   vts_worker_init (&vsm->workers[0]);
 633   for (i = 1; i < vsm->cfg.workers; i++)
 634     {
 635       vsm->workers[i].wrk_index = i;
 636       rv = pthread_create (&vsm->workers[i].thread_handle, NULL,
 637                            vts_worker_loop, (void *) &vsm->workers[i]);
 638     }
 639   vts_worker_loop (&vsm->workers[0]);
 640
 641   while (vsm->active_workers > 0)
 642     ;
 643
 644   vppcom_app_destroy ();
 645   free (vsm->workers);
 646
 647   return vsm->worker_fails;
 648 }
 649
 650 /*
 651  * fd.io coding-style-patch-verification: ON
 652  *
 653  * Local Variables:
 654  * eval: (c-set-style "gnu")
 655  * End:
 656  */