netspeed.c

   1 /* -*- Mode: C ; c-basic-offset: 2 -*- */
   2 /*
   3  * netspeed: a program to generate network traffic
   4  * Copyright (C) 2013 Free Software Foundation, Inc
   5  *
   6  * Author: Nedko Arnaudov
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License as published by
  10  * the Free Software Foundation; either version 3 of the License, or
  11  * (at your option) any later version.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  20  */
  21
  22 /* C99 */
  23 #include <stdio.h>
  24 #include <string.h>
  25 #include <stdbool.h>
  26 #include <errno.h>
  27 #include <assert.h>
  28 #include <ctype.h>
  29 #include <stdlib.h>
  30 #include <stdarg.h>
  31 #include <stdint.h>
  32 #include <limits.h>
  33
  34 /* some platforms (i.e. OpenBSD 6.4) don't define SIZE_MAX, but have SIZE_T_MAX */
  35 #if !defined(SIZE_MAX) && defined(SIZE_T_MAX)
  36 #define SIZE_MAX SIZE_T_MAX
  37 #endif
  38
  39 /* POSIX */
  40 #include <poll.h>
  41 #include <sys/mman.h>
  42 #include <sys/types.h>
  43 #include <sys/socket.h>
  44 #include <netinet/in.h>
  45 #include <netinet/tcp.h>
  46 #include <unistd.h>
  47 #include <fcntl.h>
  48 #include <netdb.h>
  49 #include <signal.h>
  50 #if !defined(NO_SCHED_FIFO)
  51 #include <sched.h>
  52 #endif
  53
  54 #define DEFAULT_WORKERS 4
  55
  56 #define DEFAULT_DL_RESOURCE "/speedtest/random4000x4000.jpg"
  57 //#define DL_RESOURCE "/speedtest/random500x500.jpg"
  58 #define DEFAULT_UL_RESOURCE "/speedtest/upload.php"
  59 #define DEFAULT_UL_SIZE ((size_t)(3 * 1024 * 1024))
  60
  61 #define ABOUT "Generate network traffic. Written by Nedko Arnaudov."
  62
  63 /* OpenBSD */
  64 #if !defined(MSG_NOSIGNAL)
  65 #define MSG_NOSIGNAL 0
  66 #endif
  67
  68 typedef int (* work_fn)(void * ctx, short revents, int * fd_ptr, short * events_ptr);
  69 typedef void (* cleanup_fn)(void * ctx);
  70
  71 #define STATE_NOT_CONNECTED           0
  72 #define STATE_CONNECTING              1
  73 #define STATE_SENDING_REQUEST_HEADER  2
  74 #define STATE_SENDING_REQUEST_BODY    3
  75 #define STATE_READING_REPLY_HEADER    4
  76 #define STATE_READING_REPLY_BODY      5
  77 #define STATE_ERROR                  -1
  78
  79 struct connection
  80 {
  81   int no;
  82   bool upload;
  83   const char * host;
  84   int state;
  85   int socket;
  86   uint32_t ip;                  /* network byte order */
  87   size_t offset;
  88   size_t size;
  89   char buffer[1024 * 1024];
  90 };
  91
  92 #define LOGLVL_NO      -1
  93 #define LOGLVL_FORCE    0
  94 #define LOGLVL_ERROR    1
  95 #define LOGLVL_WARNING  2
  96 #define LOGLVL_INFO     3
  97 #define LOGLVL_DEBUG1   4
  98 #define LOGLVL_DEBUG2   5
  99
 100 #define LOGLVL_DEFAULT_MAX LOGLVL_WARNING
 101
 102 static int g_log_max = LOGLVL_DEFAULT_MAX;
 103 static int g_progress = 0;
 104 static size_t g_workers = DEFAULT_WORKERS;
 105 static const char * g_dl_resource = DEFAULT_DL_RESOURCE;
 106 static const char * g_ul_resource = DEFAULT_UL_RESOURCE;
 107 static size_t g_ul_size = DEFAULT_UL_SIZE;
 108
 109 void log_msg(int level, const char * format, ...) __attribute__((format(printf, 2, 3)));
 110 void log_msg(int level, const char * format, ...)
 111 {
 112   va_list ap;
 113
 114   if (level > g_log_max) return;
 115
 116   va_start(ap, format);
 117   vfprintf(level > 0 ? stdout : stderr, format, ap);
 118   va_end(ap);
 119 }
 120
 121 #define connection_ptr ((struct connection *)ctx)
 122
 123 #define LOG_MSG_(level, format, ...)             \
 124   log_msg(level, format "\n", ##__VA_ARGS__)
 125 #define LOG_WORKER(level, no, format, ...)                \
 126   log_msg(level, "[%d] " format "\n", (int)(no), ##__VA_ARGS__)
 127 #define LOG_WORKER_(level, format, ...)                         \
 128   LOG_WORKER(level, connection_ptr->no, format, ##__VA_ARGS__)
 129
 130 #define LFRC( format, ...) LOG_MSG(LOGLVL_FORCE,   format, ##__VA_ARGS__)
 131 #define LERR( format, ...) LOG_MSG(LOGLVL_ERROR,   format, ##__VA_ARGS__)
 132 #define LWRN( format, ...) LOG_MSG(LOGLVL_WARNING, format, ##__VA_ARGS__)
 133 #define LINF( format, ...) LOG_MSG(LOGLVL_INFO,    format, ##__VA_ARGS__)
 134 #define LDBG1(format, ...) LOG_MSG(LOGLVL_DEBUG1,  format, ##__VA_ARGS__)
 135 #define LDBG2(format, ...) LOG_MSG(LOGLVL_DEBUG2,  format, ##__VA_ARGS__)
 136
 137 #define LOG_MSG LOG_WORKER_
 138
 139 int worker(void * ctx, short revents, int * fd_ptr, short * events_ptr)
 140 {
 141   int ret, val;
 142   struct sockaddr_in sin;
 143   socklen_t len;
 144   ssize_t sret;
 145   size_t i;
 146   const char * ptr;
 147   char size_str[100];
 148   size_t size;
 149
 150   LDBG2("state=%d", connection_ptr->state);
 151
 152   switch (connection_ptr->state)
 153   {
 154   case STATE_NOT_CONNECTED:
 155     goto connect;
 156   case STATE_CONNECTING:
 157     assert((revents & POLLOUT) == POLLOUT);
 158     goto async_connect_done;
 159   case STATE_SENDING_REQUEST_HEADER:
 160   case STATE_SENDING_REQUEST_BODY:
 161     assert((revents & POLLOUT) == POLLOUT);
 162     if ((revents & (POLLERR | POLLHUP)) != 0)
 163     {
 164       LERR("async send fd error. revents=%#hx", revents);
 165
 166       len = sizeof(val);
 167       ret = getsockopt(connection_ptr->socket, SOL_SOCKET, SO_ERROR, &val, &len);
 168       if (ret == -1)
 169       {
 170         LERR("getsockopt() failed to get socket send error. %d (%s)", errno, strerror(errno));
 171       }
 172       else
 173       {
 174         LERR("async send() error %d (%s)", val, strerror(val));
 175       }
 176       goto error;
 177     }
 178     goto send_request_continue;
 179   case STATE_READING_REPLY_HEADER:
 180     assert((revents & POLLIN) == POLLIN);
 181     if ((revents & (POLLERR | POLLHUP)) != 0)
 182     {
 183       LERR("async reply header recv fd error. revents=%#hx", revents);
 184
 185       len = sizeof(val);
 186       ret = getsockopt(connection_ptr->socket, SOL_SOCKET, SO_ERROR, &val, &len);
 187       if (ret == -1)
 188       {
 189         LERR("getsockopt() failed to get socket recv error. %d (%s)", errno, strerror(errno));
 190       }
 191       else
 192       {
 193         LERR("async recv() error %d (%s)", val, strerror(val));
 194       }
 195       goto error;
 196     }
 197     goto read_reply_header;
 198   case STATE_READING_REPLY_BODY:
 199     assert((revents & POLLIN) == POLLIN);
 200     if ((revents & (POLLERR | POLLHUP)) != 0)
 201     {
 202       LERR("async reply body recv fd error. revents=%#hx", revents);
 203
 204       len = sizeof(val);
 205       ret = getsockopt(connection_ptr->socket, SOL_SOCKET, SO_ERROR, &val, &len);
 206       if (ret == -1)
 207       {
 208         LERR("getsockopt() failed to get socket recv error. %d (%s)", errno, strerror(errno));
 209       }
 210       else
 211       {
 212         LERR("async recv() error %d (%s)", val, strerror(val));
 213       }
 214       goto error;
 215     }
 216     goto read_reply_body;
 217   default:
 218     assert(false);
 219     goto error;
 220   }
 221
 222   assert(false);
 223
 224 connect:
 225   connection_ptr->socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
 226   if (connection_ptr->socket == -1)
 227   {
 228     LERR("socket() failed. %d (%s)", errno, strerror(errno));
 229     goto error;
 230   }
 231
 232   ret = fcntl(connection_ptr->socket, F_SETFL, O_NONBLOCK);
 233   if (ret == -1)
 234   {
 235     LERR("fcntl() failed to set socket non-blocking mode. %d (%s)", errno, strerror(errno));
 236     goto error;
 237   }
 238
 239
 240   sin.sin_family = AF_INET;
 241   sin.sin_port = htons(80);
 242   sin.sin_addr.s_addr = connection_ptr->ip;
 243
 244   ret = connect(connection_ptr->socket, (struct sockaddr *)&sin, sizeof(struct sockaddr_in));
 245   if (ret == -1)
 246   {
 247     if (errno == EINPROGRESS)
 248     {
 249       connection_ptr->state = STATE_CONNECTING;
 250       *fd_ptr = connection_ptr->socket;
 251       *events_ptr = POLLOUT;
 252       return 1;
 253     }
 254
 255     LERR("connect() failed. %d (%s)", errno, strerror(errno));
 256     goto error;
 257   }
 258
 259   LINF("connect complete.");
 260   goto send_request;
 261
 262 async_connect_done:
 263   if ((revents & (POLLERR | POLLHUP)) != 0)
 264   {
 265     LERR("async connect failed. revents=%#hx", revents);
 266   }
 267
 268   len = sizeof(val);
 269   ret = getsockopt(connection_ptr->socket, SOL_SOCKET, SO_ERROR, &val, &len);
 270   if (ret == -1)
 271   {
 272     LERR("getsockopt() failed to get socket connect error. %d (%s)", errno, strerror(errno));
 273     goto error;
 274   }
 275   if (val != 0)
 276   {
 277     LERR("async connect() failed. %d (%s)", val, strerror(val));
 278     goto error;
 279   }
 280
 281   LINF("async connect complete.");
 282
 283 send_request:
 284   LINF("sending request header...");
 285
 286   if (connection_ptr->upload)
 287   {
 288     snprintf(size_str, sizeof(size_str), "%zu", g_ul_size);
 289   }
 290
 291   ret = snprintf(
 292     connection_ptr->buffer,
 293     sizeof(connection_ptr->buffer),
 294     "%s %s HTTP/1.1\r\n"
 295     "User-Agent: netspeed/0.0\r\n"
 296     "Accept: */*\r\n"
 297     "Host: %s\r\n"
 298     "%s%s%s"
 299     "\r\n",
 300     connection_ptr->upload ? "POST" : "GET",
 301     connection_ptr->upload ? g_ul_resource : g_dl_resource,
 302     connection_ptr->host,
 303     connection_ptr->upload ? "Content-Length: " : "",
 304     connection_ptr->upload ? size_str : "",
 305     connection_ptr->upload ? "\r\n" : "");
 306   if (ret < -1 || ret >= (int)sizeof(connection_ptr->buffer))
 307   {
 308     LERR("snprintf() failed compose request. %d", ret);
 309     goto error;
 310   }
 311
 312   LDBG1("request-header:\n%s", connection_ptr->buffer);
 313
 314   connection_ptr->state = STATE_SENDING_REQUEST_HEADER;
 315   connection_ptr->offset = 0;
 316   connection_ptr->size = (size_t)ret;
 317
 318 send_request_continue:
 319   while (connection_ptr->size > 0)
 320   {
 321     if (connection_ptr->state == STATE_SENDING_REQUEST_BODY &&
 322         connection_ptr->size >= sizeof(connection_ptr->buffer))
 323     {
 324       size = sizeof(connection_ptr->buffer);
 325     }
 326     else
 327     {
 328       size = connection_ptr->size;
 329     }
 330
 331     sret = send(
 332       connection_ptr->socket,
 333       connection_ptr->buffer + connection_ptr->offset,
 334       size,
 335       MSG_NOSIGNAL);
 336     if (sret == -1)
 337     {
 338       if (errno == EINTR)
 339       {
 340         continue;
 341       }
 342
 343       if (errno == EAGAIN || errno == EWOULDBLOCK)
 344       {
 345         *fd_ptr = connection_ptr->socket;
 346         *events_ptr = POLLOUT;
 347         return 1;
 348       }
 349
 350       LERR("send() failed. %d (%s)", errno, strerror(errno));
 351       goto error;
 352     }
 353
 354     if (connection_ptr->state == STATE_SENDING_REQUEST_HEADER)
 355     {
 356       connection_ptr->offset += sret;
 357     }
 358
 359     connection_ptr->size -= sret;
 360   }
 361
 362   if (connection_ptr->state == STATE_SENDING_REQUEST_HEADER)
 363   {
 364     LINF("request header sent");
 365
 366     if (connection_ptr->upload)
 367     {
 368       connection_ptr->state = STATE_SENDING_REQUEST_BODY;
 369       connection_ptr->offset = 0;
 370       connection_ptr->size = g_ul_size;
 371       LINF("sending request body...");
 372       goto send_request_continue;
 373     }
 374   }
 375   else
 376   {
 377     LINF("request body sent");
 378   }
 379
 380   connection_ptr->state = STATE_READING_REPLY_HEADER;
 381   connection_ptr->offset = 0;   /* parsed size */
 382   connection_ptr->size = 0;     /* read size */
 383
 384 read_reply_header:
 385   if (connection_ptr->size >= sizeof(connection_ptr->buffer))
 386   {
 387     LERR("HTTP reply header too big");
 388     goto error;
 389   }
 390
 391   sret = recv(
 392     connection_ptr->socket,
 393     connection_ptr->buffer + connection_ptr->size,
 394     sizeof(connection_ptr->buffer) - connection_ptr->size,
 395     MSG_NOSIGNAL);
 396   if (sret == -1)
 397   {
 398     if (errno == EINTR)
 399     {
 400       goto read_reply_header;
 401     }
 402
 403     if (errno == EAGAIN || errno == EWOULDBLOCK)
 404     {
 405       *fd_ptr = connection_ptr->socket;
 406       *events_ptr = POLLIN;
 407       return 1;
 408     }
 409
 410     LERR("recv() failed. %d (%s)", errno, strerror(errno));
 411     goto error;
 412   }
 413
 414   connection_ptr->size += sret;
 415
 416   for (i = connection_ptr->offset; i + 3 < connection_ptr->size; i++)
 417   {
 418     if (connection_ptr->buffer[i    ] == '\r' &&
 419         connection_ptr->buffer[i + 1] == '\n' &&
 420         connection_ptr->buffer[i + 2] == '\r' &&
 421         connection_ptr->buffer[i + 3] == '\n')
 422     {
 423       connection_ptr->offset = i + 4;
 424       LINF("header size is %zu bytes", connection_ptr->offset);
 425       for (i = 0; i < connection_ptr->offset; i++)
 426       {
 427         if ((signed char)connection_ptr->buffer[i] < 0)
 428         {
 429           LERR("invalid char in HTTP reply header");
 430           goto error;
 431         }
 432
 433         connection_ptr->buffer[i] = tolower(connection_ptr->buffer[i]);
 434       }
 435
 436       connection_ptr->buffer[connection_ptr->offset] = 0;
 437       LDBG1("reply-header:\n%s", connection_ptr->buffer);
 438
 439       /* calculate the size of body bytes we already read */
 440       i = connection_ptr->size - connection_ptr->offset;
 441
 442       ptr = strstr(connection_ptr->buffer, "content-length");
 443       if (ptr == NULL)
 444       {
 445         goto unknown_size;
 446       }
 447
 448       ptr += sizeof("content-length") - 1;
 449
 450       while (*ptr == ' ') ptr++;
 451
 452       if (*ptr != ':')
 453       {
 454         goto unknown_size;
 455       }
 456       ptr++;
 457
 458       while (*ptr == ' ') ptr++;
 459
 460       val = atoi(ptr);
 461
 462       if (val > 0)
 463       {
 464         LINF("total body size is %d bytes", val);
 465
 466         if ((size_t)val < i)
 467         {
 468           LERR("body bigger than announced");
 469           goto error;
 470         }
 471
 472         /* substract the already received body bytes */
 473         connection_ptr->size = (size_t)val - i;
 474       }
 475       else
 476       {
 477       unknown_size:
 478        /* server didnt provide body size,
 479            assume body end will be marked by connection close */
 480         LWRN("unknown body size");
 481         goto error;
 482         connection_ptr->size = SIZE_MAX;
 483       }
 484
 485       connection_ptr->state = STATE_READING_REPLY_BODY;
 486       connection_ptr->offset = i;
 487       goto read_reply_body;
 488     }
 489   }
 490
 491   if (i >= 4)
 492   {
 493     /* next time don't parse the bytes already parsed */
 494     connection_ptr->offset = i - 4;
 495   }
 496
 497   goto read_reply_header;
 498
 499 read_reply_body:
 500   while (connection_ptr->size > 0)
 501   {
 502     sret = recv(
 503       connection_ptr->socket,
 504       connection_ptr->buffer,
 505       sizeof(connection_ptr->buffer),
 506       MSG_NOSIGNAL);
 507     if (sret == -1)
 508     {
 509       if (errno == EINTR)
 510       {
 511         goto read_reply_header;
 512       }
 513
 514       if (errno == EAGAIN || errno == EWOULDBLOCK)
 515       {
 516         *fd_ptr = connection_ptr->socket;
 517         *events_ptr = POLLIN;
 518         return 1;
 519       }
 520
 521       LERR("recv() failed. %d (%s)", errno, strerror(errno));
 522       goto error;
 523     }
 524
 525     connection_ptr->size -= sret;
 526     connection_ptr->offset += sret;
 527     if (g_progress > 0)
 528     {
 529       if (g_progress == 1)
 530       {
 531         printf(".");
 532       }
 533       else
 534       {
 535         printf("(%zd)", sret);
 536       }
 537
 538       fflush(stdout);
 539     }
 540   }
 541
 542   LINF("%zu body bytes read", connection_ptr->offset);
 543   goto send_request;
 544   //return 0;                     /* done */
 545
 546 error:
 547   connection_ptr->state = STATE_ERROR;
 548   return -1;
 549 }
 550
 551 void connection_cleanup(void * ctx)
 552 {
 553   if (connection_ptr->socket != -1)
 554   {
 555     LINF("closing socket...");
 556     close(connection_ptr->socket);
 557   }
 558 }
 559
 560 #undef connection_ptr
 561 #undef LOG_MSG
 562 #define LOG_MSG LOG_MSG_
 563
 564 uint32_t resolve_host(const char * hostname)
 565 {
 566   struct hostent * he_ptr;
 567
 568   he_ptr = gethostbyname(hostname);
 569   if (he_ptr == NULL)
 570   {
 571     LERR("Cannot resolve \"%s\". h_errno is %d", hostname, h_errno);
 572     return 0;
 573   }
 574
 575   return *(uint32_t *)(he_ptr->h_addr);
 576 }
 577
 578 bool
 579 create_worker(
 580   int worker_no,
 581   const char * type,
 582   uint32_t ip,
 583   const char * hostname,
 584   void ** ctx,
 585   work_fn * work,
 586   cleanup_fn * cleanup)
 587 {
 588   struct connection * connection_ptr;
 589   bool upload;
 590
 591   if (strcmp(type, "d") == 0)
 592   {
 593     upload = false;
 594   }
 595   else if (strcmp(type, "u") == 0)
 596   {
 597     upload = true;
 598   }
 599   else
 600   {
 601     LOG_WORKER(LOGLVL_ERROR, worker_no, "unknown type \"%s\".", type);
 602     return false;
 603   }
 604
 605   LOG_WORKER(LOGLVL_INFO, worker_no, "connecting to %s for %s", hostname, upload ? "uploading" : "downloading");
 606
 607   connection_ptr = malloc(sizeof(struct connection));
 608   if (connection_ptr == NULL)
 609   {
 610     LOG_WORKER(LOGLVL_ERROR, worker_no, "memory allocation failed.");
 611     return false;
 612   }
 613
 614   connection_ptr->no = worker_no;
 615   connection_ptr->upload = upload;
 616   connection_ptr->host = hostname;
 617   connection_ptr->state = STATE_NOT_CONNECTED;
 618   connection_ptr->socket = -1;
 619   connection_ptr->ip = ip;
 620
 621   *ctx = connection_ptr;
 622   *work = worker;
 623   *cleanup = connection_cleanup;
 624
 625   return true;
 626 }
 627
 628 #if !defined(NO_SCHED_FIFO)
 629 static bool disable_preemption(int priority)
 630 {
 631   struct sched_param sched_param;
 632
 633   sched_param.sched_priority = priority;
 634   if (sched_setscheduler(0, SCHED_FIFO, &sched_param) != 0)
 635   {
 636     LERR("Cannot set scheduling policy %d (%s)", errno, strerror(errno));
 637     return false;
 638   }
 639
 640   return true;
 641 }
 642 #endif
 643
 644 int main(int argc, char ** argv)
 645 {
 646   int ret;
 647   uint32_t ip;
 648   struct worker
 649   {
 650     void * ctx;
 651     work_fn work;
 652     cleanup_fn cleanup;
 653     struct pollfd pollfd;
 654   };
 655   struct worker * workers = NULL;
 656   struct pollfd * pollfds = NULL;
 657   int nfds, poll_index;
 658   size_t i;
 659   bool worker_count_supplied = false;
 660   size_t workers_per_host;
 661   int host_index;
 662
 663   argc--; argv++;
 664
 665   /* process options */
 666   while (argc > 0 && **argv == '-')
 667   {
 668     if (strcmp(*argv, "-v") == 0)
 669     {
 670       g_log_max++;
 671     }
 672     else if (strcmp(*argv, "-q") == 0)
 673     {
 674       g_log_max--;
 675     }
 676     else if (strcmp(*argv, "-s") == 0)
 677     {
 678       g_log_max = -1;
 679     }
 680     else if (strcmp(*argv, "-p") == 0)
 681     {
 682       g_progress++;
 683     }
 684     else if (strcmp(*argv, "--help") == 0)
 685     {
 686       goto help;
 687     }
 688     else if (strcmp(*argv, "-w") == 0 && argc >= 2)
 689     {
 690       i = atoi(argv[1]);
 691       if (i < 1)
 692       {
 693         LERR("Bad value for workers option");
 694         goto optfail;
 695       }
 696
 697       g_workers = i;
 698       worker_count_supplied = true;
 699       argc--; argv++;
 700     }
 701     else if (strcmp(*argv, "-u") == 0 && argc >= 2)
 702     {
 703       i = atoi(argv[1]);
 704       if (i < 1)
 705       {
 706         LERR("Bad value for upload size option");
 707         goto optfail;
 708       }
 709
 710       g_ul_size = i;
 711       argc--; argv++;
 712     }
 713     else
 714     {
 715       LWRN("Ignoring unknown option %s", *argv);
 716     }
 717
 718     //printf("log level max is %d\n", g_log_max);
 719     argc--; argv++;
 720   }
 721
 722   if (argc == 0 || (argc % 2) != 0)
 723   {
 724   help:
 725     LFRC(ABOUT);
 726     LFRC("Usage: netspeed [options] <type> <host> [<type> <host>] ...");
 727     LFRC("");
 728     LFRC("Options:");
 729     LFRC("  -v Increase verbosity. May be used more than once.");
 730     LFRC("  -q Decrease verbosity. May be used more than once.");
 731     LFRC("  -s Be completely silent.");
 732     LFRC("  -p Print progress (dots). Use twice for printing chunk sizes.");
 733     LFRC("  -w <num> Worker count (per type/host pair).");
 734     LFRC("  -u <num> Upload size. Ignored when downloading.");
 735     LFRC("  --help Shows this help text");
 736     LFRC("");
 737     LFRC("  <type> is either 'u' (upload) or 'd' (download)");
 738     LFRC("  <host> is a ookla speedtest host");
 739     ret = 0;
 740     goto exit;
 741   }
 742
 743   LFRC(ABOUT);
 744   if (g_log_max != LOGLVL_DEFAULT_MAX)
 745   {
 746     LFRC("log level max is %d", g_log_max);
 747   }
 748
 749   if (argc > 2)
 750   {
 751     workers_per_host = worker_count_supplied ? g_workers : 1;
 752
 753     LFRC("%zu worker(s) per host", workers_per_host);
 754
 755     g_workers = workers_per_host * (argc / 2);
 756
 757     LFRC("%zu total worker(s)", g_workers);
 758   }
 759   else
 760   {
 761     workers_per_host = g_workers;
 762     LFRC("%zu worker(s)", g_workers);
 763   }
 764
 765   LFRC("POST body size: %zu bytes", g_ul_size);
 766   //LFRC("download resource: %s", g_dl_resource);
 767   //LFRC("upload resource: %s", g_ul_resource);
 768
 769   workers = calloc(g_workers, sizeof(struct worker));
 770   if (workers == NULL)
 771   {
 772     LERR("memory allocation failed. (workers)");
 773     ret = 1;
 774     goto free;
 775   }
 776
 777   pollfds = calloc(g_workers, sizeof(struct pollfd));
 778   if (pollfds == NULL)
 779   {
 780     LERR("memory allocation failed. (pollfds)");
 781     ret = 1;
 782     goto free;
 783   }
 784
 785   if (signal(SIGPIPE, SIG_IGN) == SIG_ERR)
 786   {
 787     LERR("Cannot ignore SIGPIPE. %d (%s)", errno, strerror(errno));
 788     goto fail;
 789   }
 790
 791   host_index = 0;
 792   while (argc > 0)
 793   {
 794     assert(argc >= 2);
 795     assert(argc % 2 == 0);
 796
 797     ip = resolve_host(argv[1]);
 798     if (ip == 0)
 799     {
 800       goto fail;
 801     }
 802
 803     for (i = host_index * workers_per_host; i < (host_index + 1) * workers_per_host; i++)
 804     {
 805       if (!create_worker(
 806             i,
 807             argv[0],
 808             ip,
 809             argv[1],
 810             &workers[i].ctx,
 811             &workers[i].work,
 812             &workers[i].cleanup))
 813       {
 814         g_workers = 0;
 815         goto fail;
 816       }
 817
 818       workers[i].pollfd.fd = -1;
 819       workers[i].pollfd.revents = 0;
 820     }
 821
 822     host_index++;
 823     argc -= 2;
 824     argv += 2;
 825   }
 826
 827   ret = mlockall(MCL_CURRENT | MCL_FUTURE);
 828   if (ret == -1)
 829   {
 830     LERR("mlockall() failed. %d (%s)", errno, strerror(errno));
 831     //goto fail;
 832   }
 833
 834 #if !defined(NO_SCHED_FIFO)
 835   if (!disable_preemption(10))
 836   {
 837     //goto fail;
 838   }
 839 #endif
 840
 841   poll_index = 0;
 842 loop:
 843   assert(poll_index == 0);
 844   for (i = 0; i < g_workers; i++)
 845   {
 846     if (workers[i].work != NULL)
 847     {
 848       if (workers[i].pollfd.fd == -1 || /* first time */
 849           workers[i].pollfd.revents != 0) /* or when there are pending events */
 850       {
 851         ret = workers[i].work(
 852           workers[i].ctx,
 853           workers[i].pollfd.revents,
 854           &workers[i].pollfd.fd,
 855           &workers[i].pollfd.events);
 856         if (ret < 0)
 857         {
 858           ret = -ret;
 859           goto cleanup;
 860         }
 861
 862         if (ret == 0)
 863         {
 864           /* worker done */
 865           workers[i].work = NULL;
 866           LOG_WORKER(LOGLVL_INFO, i, "worker done");
 867           continue;
 868         }
 869
 870         workers[i].pollfd.revents = 0;
 871
 872         assert(workers[i].pollfd.fd != -1);
 873         assert(workers[i].pollfd.events != 0);
 874         LOG_WORKER(LOGLVL_DEBUG2, i, "worker waits on %d\n", workers[i].pollfd.fd);
 875       }
 876       else
 877       {
 878         LOG_WORKER(LOGLVL_DEBUG2, i, "worker still waits on %d\n", workers[i].pollfd.fd);
 879       }
 880
 881       pollfds[poll_index].fd = workers[i].pollfd.fd;
 882       pollfds[poll_index].events = workers[i].pollfd.events;
 883       pollfds[poll_index].revents = 0;
 884       poll_index++;
 885     }
 886   }
 887
 888   if (poll_index == 0)
 889   {
 890     ret = 0;
 891     LINF("no more workers");
 892     goto cleanup;
 893   }
 894
 895   nfds = poll_index;
 896   LDBG2("polling %d fds", nfds);
 897   ret = poll(pollfds, nfds, -1);
 898   LDBG2("poll() returns %d", ret);
 899   if (ret == -1)
 900   {
 901     LERR("poll() failed. %d (%s)", errno, strerror(errno));
 902     goto fail;
 903   }
 904
 905   assert(ret > 0);
 906   poll_index = 0;
 907   while (ret > 0)
 908   {
 909     assert(poll_index < nfds);
 910     if (pollfds[poll_index].revents != 0)
 911     {
 912       for (i = 0; i < g_workers; i++)
 913       {
 914         if (workers[i].work != NULL &&
 915             workers[i].pollfd.fd == pollfds[poll_index].fd)
 916         {
 917           workers[i].pollfd.revents = pollfds[poll_index].revents;
 918           assert(workers[i].pollfd.revents != 0);
 919           break;
 920         }
 921       }
 922       assert(i < g_workers);        /* fd/worker not found */
 923       ret--;
 924     }
 925     poll_index++;
 926   }
 927   poll_index = 0;
 928   goto loop;
 929
 930 fail:
 931   ret = 1;
 932 cleanup:
 933   for (i = 0; i < g_workers; i++)
 934   {
 935     if (workers[i].cleanup != NULL)
 936     {
 937       workers[i].cleanup(workers[i].ctx);
 938     }
 939   }
 940 free:
 941   free(workers);
 942   free(pollfds);
 943 exit:
 944   return ret;
 945 optfail:
 946   ret = 1;
 947   goto exit;
 948 }