tools/tools/netrate/kq_sendrecv/kq_sendcli/kq_sendcli.c

   1 #include <sys/param.h>
   2 #include <sys/event.h>
   3 #include <sys/ioctl.h>
   4 #include <sys/queue.h>
   5 #include <sys/socket.h>
   6 #include <sys/sysctl.h>
   7 #include <sys/time.h>
   8
   9 #include <machine/atomic.h>
  10 #ifdef __FreeBSD__
  11 #include <machine/cpu.h>
  12 #endif
  13 #include <machine/cpufunc.h>
  14
  15 #include <arpa/inet.h>
  16 #include <netinet/in.h>
  17
  18 #include <err.h>
  19 #include <errno.h>
  20 #include <pthread.h>
  21 #include <pthread_np.h>
  22 #include <signal.h>
  23 #include <stdbool.h>
  24 #include <stdio.h>
  25 #include <stdint.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <unistd.h>
  29
  30 #include "kq_sendrecv_proto.h"
  31
  32 /*
  33  * Note about the sender start synchronization.
  34  *
  35  * We apply two stage synchronization.  The first stage uses pthread
  36  * condition (it sleeps), which waits for the establishment for all
  37  * connections, which could be slow.  The second stage uses g_nwait
  38  * of send_globctx; all relevant threads spin on g_nwait.  The main
  39  * thread spin-waits for all senders to increase g_nwait.  The sender
  40  * thread increases the g_nwait, then it spin-waits for main thread
  41  * to reset g_nwait.  In this way, we can make sure that all senders
  42  * start roughly at the same time.
  43  */
  44
  45 #ifndef timespecsub
  46 #define timespecsub(vvp, uvp)                                           \
  47         do {                                                            \
  48                 (vvp)->tv_sec -= (uvp)->tv_sec;                         \
  49                 (vvp)->tv_nsec -= (uvp)->tv_nsec;                       \
  50                 if ((vvp)->tv_nsec < 0) {                               \
  51                         (vvp)->tv_sec--;                                \
  52                         (vvp)->tv_nsec += 1000000000;                   \
  53                 }                                                       \
  54         } while (0)
  55 #endif
  56
  57 #ifndef timespeccmp
  58 #define timespeccmp(tvp, uvp, cmp)                                      \
  59         (((tvp)->tv_sec == (uvp)->tv_sec) ?                             \
  60             ((tvp)->tv_nsec cmp (uvp)->tv_nsec) :                       \
  61             ((tvp)->tv_sec cmp (uvp)->tv_sec))
  62 #endif
  63
  64 #if 0
  65 #define SEND_DEBUG
  66 #endif
  67 #if 0
  68 #define SEND_TIME_DEBUG
  69 #endif
  70
  71 #define SEND_DUR                10
  72 #define SEND_EVENT_MAX          64
  73 #define SEND_BUFLEN             (128 * 1024)
  74
  75 /*
  76  * The successful 3-way handshake on the connection does not mean the
  77  * remote application can accept(2) this connection.  Even worse, the
  78  * remote side's network stack may drop the connection silently, i.e.
  79  * w/o RST.  If this happened, the blocking read(2) would not return,
  80  * until the keepalive kicked in, which would take quite some time.
  81  * This is obviously not what we want here, so use synthetic timeout
  82  * for blocking read(2).  Here, we will retry if a blocking read(2)
  83  * times out.
  84  */
  85 #define SEND_READTO_MS          1000            /* unit: ms */
  86
  87 #if defined(__DragonFly__)
  88 #define SEND_CONN_CTX_ALIGN     __VM_CACHELINE_SIZE
  89 #elif defined(__FreeBSD__)
  90 #define SEND_CONN_CTX_ALIGN     CACHE_LINE_SIZE
  91 #else
  92 #define SEND_CONN_CTX_ALIGN     64      /* XXX */
  93 #endif
  94
  95 struct conn_ctx {
  96         int                     c_s;
  97         int                     c_err;
  98         uint64_t                c_stat;
  99         struct timespec         c_terr;
 100
 101         STAILQ_ENTRY(conn_ctx)  c_glob_link;
 102         STAILQ_ENTRY(conn_ctx)  c_link;
 103         struct sockaddr_in      c_in;
 104         int                     c_thr_id;
 105 } __aligned(SEND_CONN_CTX_ALIGN);
 106
 107 STAILQ_HEAD(conn_ctx_list, conn_ctx);
 108
 109 struct send_globctx {
 110         struct conn_ctx_list    g_conn;
 111
 112         int                     g_dur;
 113         int                     g_nconn;
 114         pthread_mutex_t         g_lock;
 115         pthread_cond_t          g_cond;
 116
 117         volatile u_int          g_nwait;
 118         int                     g_readto_ms;    /* unit: ms */
 119         int                     g_buflen;
 120         bool                    g_sendfile;
 121 };
 122
 123 struct send_thrctx {
 124         struct conn_ctx_list    t_conn;
 125         pthread_mutex_t         t_lock;
 126         pthread_cond_t          t_cond;
 127
 128         struct send_globctx     *t_glob;
 129         struct timespec         t_start;
 130         struct timespec         t_end;
 131         double                  t_run_us;       /* unit: us */
 132
 133         pthread_t               t_tid;
 134         int                     t_id;
 135 };
 136
 137 static void     send_build_addrlist(const struct sockaddr_in *, int,
 138                     const struct sockaddr_in **, int *, int);
 139 static void     *send_thread(void *);
 140
 141 static __inline void
 142 send_spinwait(void)
 143 {
 144 #if defined(__DragonFly__)
 145         cpu_pause();
 146 #elif defined(__FreeBSD__)
 147         cpu_spinwait();
 148 #else
 149         /* XXX nothing */
 150 #endif
 151 }
 152
 153 static void
 154 usage(const char *cmd)
 155 {
 156         fprintf(stderr, "%s -4 addr4 [-4 addr4 ...] [-p port] "
 157             "-c conns [-t nthreads] [-l sec] [-r readto_ms] [-S] [-E] "
 158             "[-b buflen] [-B]\n", cmd);
 159         exit(2);
 160 }
 161
 162 int
 163 main(int argc, char *argv[])
 164 {
 165         struct send_globctx glob;
 166         struct send_thrctx *ctx_arr, *ctx;
 167         struct sockaddr_in *in_arr, *in;
 168         const struct sockaddr_in *daddr;
 169         struct timespec run, end, start;
 170         double total_run_us, total, conn_min, conn_max;
 171         double jain, jain_res;
 172         int jain_cnt;
 173         struct conn_ctx *conn;
 174         sigset_t sigset;
 175         int opt, i, ncpus;
 176         int in_arr_cnt, in_arr_sz, ndaddr;
 177         int nthr, nconn, dur, readto_ms, buflen;
 178         int log_err, err_cnt, has_minmax;
 179         u_short port = RECV_PORT;
 180         uint32_t idx;
 181         size_t sz;
 182         bool do_sendfile = false, bindcpu = false;
 183
 184         sigemptyset(&sigset);
 185         sigaddset(&sigset, SIGPIPE);
 186         if (sigprocmask(SIG_BLOCK, &sigset, NULL) < 0)
 187                 err(1, "sigprocmask failed");
 188
 189         sz = sizeof(ncpus);
 190         if (sysctlbyname("hw.ncpu", &ncpus, &sz, NULL, 0) < 0)
 191                 err(1, "sysctl hw.ncpu failed");
 192         nthr = ncpus;
 193
 194         in_arr_sz = 4;
 195         in_arr_cnt = 0;
 196         in_arr = malloc(in_arr_sz * sizeof(struct sockaddr_in));
 197         if (in_arr == NULL)
 198                 err(1, "malloc failed");
 199
 200         log_err = 0;
 201         nconn = 0;
 202         dur = SEND_DUR;
 203         readto_ms = SEND_READTO_MS;
 204         buflen = SEND_BUFLEN;
 205
 206         while ((opt = getopt(argc, argv, "4:BESb:c:l:p:r:t:")) != -1) {
 207                 switch (opt) {
 208                 case '4':
 209                         if (in_arr_cnt == in_arr_sz) {
 210                                 in_arr_sz *= 2;
 211                                 in_arr = reallocf(in_arr,
 212                                     in_arr_sz * sizeof(struct sockaddr_in));
 213                                 if (in_arr == NULL)
 214                                         err(1, "reallocf failed");
 215                         }
 216                         in = &in_arr[in_arr_cnt];
 217                         ++in_arr_cnt;
 218
 219                         memset(in, 0, sizeof(*in));
 220                         in->sin_family = AF_INET;
 221                         if (inet_pton(AF_INET, optarg, &in->sin_addr) <= 0)
 222                                 errx(1, "inet_pton failed %s", optarg);
 223                         break;
 224
 225                 case 'B':
 226                         bindcpu = true;
 227                         break;
 228
 229                 case 'E':
 230                         log_err = 1;
 231                         break;
 232
 233                 case 'S':
 234                         do_sendfile = true;
 235                         break;
 236
 237                 case 'b':
 238                         buflen = strtol(optarg, NULL, 10);
 239                         if (buflen <= 0)
 240                                 errx(1, "invalid -b");
 241                         break;
 242
 243                 case 'c':
 244                         nconn = strtol(optarg, NULL, 10);
 245                         if (nconn <= 0)
 246                                 errx(1, "invalid -c");
 247                         break;
 248
 249                 case 'l':
 250                         dur = strtoul(optarg, NULL, 10);
 251                         if (dur == 0)
 252                                 errx(1, "invalid -l");
 253                         break;
 254
 255                 case 'p':
 256                         port = strtoul(optarg, NULL, 10);
 257                         break;
 258
 259                 case 'r':
 260                         readto_ms = strtol(optarg, NULL, 10);
 261                         if (readto_ms <= 0)
 262                                 errx(1, "invalid -r");
 263                         break;
 264
 265                 case 't':
 266                         nthr = strtol(optarg, NULL, 10);
 267                         if (nthr <= 0)
 268                                 errx(1, "invalid -t");
 269                         break;
 270
 271                 default:
 272                         usage(argv[0]);
 273                 }
 274         }
 275         if (in_arr_cnt == 0 || nconn == 0)
 276                 errx(1, "either -4 or -c are specified");
 277
 278         if (nthr > nconn)
 279                 nthr = nconn;
 280
 281         for (i = 0; i < in_arr_cnt; ++i)
 282                 in_arr[i].sin_port = htons(port);
 283
 284         ctx_arr = calloc(nthr, sizeof(struct send_thrctx));
 285         if (ctx_arr == NULL)
 286                 err(1, "calloc failed");
 287
 288         memset(&glob, 0, sizeof(glob));
 289         STAILQ_INIT(&glob.g_conn);
 290         glob.g_nconn = nconn;
 291         glob.g_nwait = 1; /* count self in */
 292         glob.g_dur = dur;
 293         glob.g_readto_ms = readto_ms;
 294         glob.g_sendfile = do_sendfile;
 295         glob.g_buflen = buflen;
 296         pthread_mutex_init(&glob.g_lock, NULL);
 297         pthread_cond_init(&glob.g_cond, NULL);
 298
 299         pthread_set_name_np(pthread_self(), "main");
 300
 301         /* Build receiver address list */
 302         send_build_addrlist(in_arr, in_arr_cnt, &daddr, &ndaddr, readto_ms);
 303
 304         /*
 305          * Start senders.
 306          */
 307         for (i = 0; i < nthr; ++i) {
 308                 pthread_attr_t attr;
 309                 int error;
 310
 311                 ctx = &ctx_arr[i];
 312                 STAILQ_INIT(&ctx->t_conn);
 313                 ctx->t_id = i;
 314                 ctx->t_glob = &glob;
 315                 pthread_mutex_init(&ctx->t_lock, NULL);
 316                 pthread_cond_init(&ctx->t_cond, NULL);
 317
 318                 pthread_attr_init(&attr);
 319                 if (bindcpu) {
 320                         cpu_set_t mask;
 321
 322                         CPU_ZERO(&mask);
 323                         CPU_SET(i % ncpus, &mask);
 324                         error = pthread_attr_setaffinity_np(&attr,
 325                             sizeof(mask), &mask);
 326                         if (error) {
 327                                 errc(1, error, "pthread_attr_setaffinity_np "
 328                                     "failed");
 329                         }
 330                 }
 331
 332                 error = pthread_create(&ctx->t_tid, &attr, send_thread, ctx);
 333                 if (error)
 334                         errc(1, error, "pthread_create failed");
 335                 pthread_attr_destroy(&attr);
 336         }
 337
 338         /*
 339          * Distribute connections to senders.
 340          *
 341          * NOTE:
 342          * We start from a random position in the address list, so that the
 343          * first several receiving servers will not be abused, if the number
 344          * of connections is small and there are many clients.
 345          */
 346         idx = arc4random_uniform(ndaddr);
 347         for (i = 0; i < nconn; ++i) {
 348                 const struct sockaddr_in *da;
 349
 350                 da = &daddr[idx % ndaddr];
 351                 ++idx;
 352
 353                 conn = aligned_alloc(SEND_CONN_CTX_ALIGN, sizeof(*conn));
 354                 if (conn == NULL)
 355                         err(1, "aligned_alloc failed");
 356                 memset(conn, 0, sizeof(*conn));
 357                 conn->c_in = *da;
 358                 conn->c_s = -1;
 359
 360                 ctx = &ctx_arr[i % nthr];
 361                 conn->c_thr_id = ctx->t_id;
 362
 363                 pthread_mutex_lock(&ctx->t_lock);
 364                 STAILQ_INSERT_TAIL(&ctx->t_conn, conn, c_link);
 365                 pthread_mutex_unlock(&ctx->t_lock);
 366                 pthread_cond_signal(&ctx->t_cond);
 367
 368                 /* Add to the global list for results gathering */
 369                 STAILQ_INSERT_TAIL(&glob.g_conn, conn, c_glob_link);
 370         }
 371
 372         /*
 373          * No more connections; notify the senders.
 374          *
 375          * NOTE:
 376          * The marker for 'the end of connection list' has 0 in its
 377          * c_in.sin_port.
 378          */
 379         for (i = 0; i < nthr; ++i) {
 380                 conn = aligned_alloc(SEND_CONN_CTX_ALIGN, sizeof(*conn));
 381                 if (conn == NULL)
 382                         err(1, "aligned_alloc failed");
 383                 memset(conn, 0, sizeof(*conn));
 384                 conn->c_s = -1;
 385
 386                 ctx = &ctx_arr[i];
 387                 pthread_mutex_lock(&ctx->t_lock);
 388                 STAILQ_INSERT_TAIL(&ctx->t_conn, conn, c_link);
 389                 pthread_mutex_unlock(&ctx->t_lock);
 390                 pthread_cond_signal(&ctx->t_cond);
 391         }
 392
 393         /*
 394          * Sender start sync, stage 1:
 395          * Wait for connections establishment (slow).
 396          */
 397         pthread_mutex_lock(&glob.g_lock);
 398         while (glob.g_nconn != 0)
 399                 pthread_cond_wait(&glob.g_cond, &glob.g_lock);
 400         pthread_mutex_unlock(&glob.g_lock);
 401
 402         /*
 403          * Sender start sync, stage 2:
 404          * Wait for senders to spin-wait; and once all senders spin-wait,
 405          * release them by resetting g_nwait.
 406          */
 407         while (atomic_cmpset_int(&glob.g_nwait, nthr + 1, 0) == 0)
 408                 send_spinwait();
 409
 410         fprintf(stderr, "start %d seconds sending test: %d threads, "
 411             "%d connections\n", dur, nthr, nconn);
 412
 413         /*
 414          * Wait for the senders to finish and gather the results.
 415          */
 416
 417         memset(&end, 0, sizeof(end));           /* XXX stupid gcc warning */
 418         memset(&start, 0, sizeof(start));       /* XXX stupid gcc warning */
 419
 420         for (i = 0; i < nthr; ++i) {
 421                 ctx = &ctx_arr[i];
 422                 pthread_join(ctx->t_tid, NULL);
 423
 424                 run = ctx->t_end;
 425                 timespecsub(&run, &ctx->t_start);
 426                 ctx->t_run_us = ((double)run.tv_sec * 1000000.0) +
 427                     ((double)run.tv_nsec / 1000.0);
 428
 429                 if (i == 0) {
 430                         start = ctx->t_start;
 431                         end = ctx->t_end;
 432                 } else {
 433                         if (timespeccmp(&start, &ctx->t_start, >))
 434                                 start = ctx->t_start;
 435                         if (timespeccmp(&end, &ctx->t_end, <))
 436                                 end = ctx->t_end;
 437                 }
 438
 439 #ifdef SEND_TIME_DEBUG
 440                 fprintf(stderr, "start %ld.%ld, end %ld.%ld\n",
 441                     ctx->t_start.tv_sec, ctx->t_start.tv_nsec,
 442                     ctx->t_end.tv_sec, ctx->t_end.tv_nsec);
 443 #endif
 444         }
 445
 446 #ifdef SEND_TIME_DEBUG
 447         fprintf(stderr, "start %ld.%ld, end %ld.%ld (final)\n",
 448             start.tv_sec, start.tv_nsec, end.tv_sec, end.tv_nsec);
 449 #endif
 450
 451         run = end;
 452         timespecsub(&run, &start);
 453         total_run_us = ((double)run.tv_sec * 1000000.0) +
 454             ((double)run.tv_nsec / 1000.0);
 455         total = 0.0;
 456
 457         err_cnt = 0;
 458         has_minmax = 0;
 459         conn_min = 0.0;
 460         conn_max = 0.0;
 461
 462         jain = 0.0;
 463         jain_res = 0.0;
 464         jain_cnt = 0;
 465
 466         STAILQ_FOREACH(conn, &glob.g_conn, c_glob_link) {
 467                 total += conn->c_stat;
 468                 if (conn->c_err == 0) {
 469                         double perf;    /* unit: Mbps */
 470
 471                         perf = (conn->c_stat * 8.0) /
 472                             ctx_arr[conn->c_thr_id].t_run_us;
 473                         if (!has_minmax) {
 474                                 conn_min = perf;
 475                                 conn_max = perf;
 476                                 has_minmax = 1;
 477                         } else {
 478                                 if (perf > conn_max)
 479                                         conn_max = perf;
 480                                 if (perf < conn_min)
 481                                         conn_min = perf;
 482                         }
 483                         jain += (perf * perf);
 484                         jain_res += perf;
 485                         ++jain_cnt;
 486                 } else {
 487                         ++err_cnt;
 488                 }
 489         }
 490
 491         jain *= jain_cnt;
 492         jain = (jain_res * jain_res) / jain;
 493
 494         printf("Total: %.2lf Mbps, min/max %.2lf Mbps/%.2lf Mbps, jain %.2lf, "
 495             "error %d\n", (total * 8.0) / total_run_us, conn_min, conn_max,
 496             jain, err_cnt);
 497
 498         if (log_err && err_cnt) {
 499                 STAILQ_FOREACH(conn, &glob.g_conn, c_glob_link) {
 500                         char name[INET_ADDRSTRLEN];
 501                         double tmp_run;
 502
 503                         if (conn->c_err == 0)
 504                                 continue;
 505
 506                         run = conn->c_terr;
 507                         timespecsub(&run, &ctx_arr[conn->c_thr_id].t_start);
 508                         tmp_run = ((double)run.tv_sec * 1000000.0) +
 509                             ((double)run.tv_nsec / 1000.0);
 510                         fprintf(stderr, "snd%d ->%s:%d, %ld sec, %.2lf Mbps, "
 511                             "errno %d\n",
 512                             conn->c_thr_id,
 513                             inet_ntop(AF_INET, &conn->c_in.sin_addr,
 514                                 name, sizeof(name)),
 515                             ntohs(conn->c_in.sin_port),
 516                             run.tv_sec, (conn->c_stat * 8.0) / tmp_run,
 517                             conn->c_err);
 518                         --err_cnt;
 519                         if (err_cnt == 0)
 520                                 break;
 521                 }
 522         }
 523
 524         exit(0);
 525 }
 526
 527 static void
 528 send_build_addrlist(const struct sockaddr_in *in_arr, int in_arr_cnt,
 529     const struct sockaddr_in **daddr0, int *ndaddr0, int readto_ms)
 530 {
 531         struct sockaddr_in *daddr;
 532         struct timeval readto;
 533         int i, ndaddr;
 534
 535         daddr = NULL;
 536         ndaddr = 0;
 537
 538         memset(&readto, 0, sizeof(readto));
 539         readto.tv_sec = readto_ms / 1000;
 540         readto.tv_usec = (readto_ms % 1000) * 1000;
 541
 542         for (i = 0; i < in_arr_cnt; ++i) {
 543                 const struct sockaddr_in *in = &in_arr[i];
 544                 struct recv_info info_hdr;
 545                 uint16_t *ports;
 546                 int s, n, ports_sz, d;
 547
 548 again:
 549                 s = socket(AF_INET, SOCK_STREAM, 0);
 550                 if (s < 0)
 551                         err(1, "socket failed");
 552
 553                 if (connect(s, (const struct sockaddr *)in, sizeof(*in)) < 0)
 554                         err(1, "connect failed");
 555
 556                 if (setsockopt(s, SOL_SOCKET, SO_RCVTIMEO,
 557                     &readto, sizeof(readto)) < 0)
 558                         err(1, "setsockopt(RCVTIMEO) failed");
 559
 560                 n = read(s, &info_hdr, sizeof(info_hdr));
 561                 if (n != sizeof(info_hdr)) {
 562                         if (n < 0) {
 563                                 if (errno == EAGAIN) {
 564                                         close(s);
 565                                         goto again;
 566                                 }
 567                                 err(1, "read info hdr failed");
 568                         } else {
 569                                 errx(1, "read truncated info hdr");
 570                         }
 571                 }
 572                 if (info_hdr.ndport == 0) {
 573                         close(s);
 574                         continue;
 575                 }
 576
 577                 ports_sz = info_hdr.ndport * sizeof(uint16_t);
 578                 ports = malloc(ports_sz);
 579                 if (ports == NULL)
 580                         err(1, "malloc failed");
 581
 582                 n = read(s, ports, ports_sz);
 583                 if (n != ports_sz) {
 584                         if (n < 0) {
 585                                 if (errno == EAGAIN) {
 586                                         free(ports);
 587                                         close(s);
 588                                         goto again;
 589                                 }
 590                                 err(1, "read ports failed");
 591                         } else {
 592                                 errx(1, "read truncated ports");
 593                         }
 594                 }
 595
 596                 daddr = reallocf(daddr,
 597                     (ndaddr + info_hdr.ndport) * sizeof(struct sockaddr_in));
 598                 if (daddr == NULL)
 599                         err(1, "reallocf failed");
 600
 601                 for (d = ndaddr; d < ndaddr + info_hdr.ndport; ++d) {
 602                         struct sockaddr_in *da = &daddr[d];
 603
 604                         *da = *in;
 605                         da->sin_port = ports[d - ndaddr];
 606                 }
 607                 ndaddr += info_hdr.ndport;
 608
 609                 free(ports);
 610                 close(s);
 611         }
 612
 613 #ifdef SEND_DEBUG
 614         for (i = 0; i < ndaddr; ++i) {
 615                 const struct sockaddr_in *da = &daddr[i];
 616                 char name[INET_ADDRSTRLEN];
 617
 618                 fprintf(stderr, "%s:%d\n",
 619                     inet_ntop(AF_INET, &da->sin_addr, name, sizeof(name)),
 620                     ntohs(da->sin_port));
 621         }
 622 #endif
 623
 624         *daddr0 = daddr;
 625         *ndaddr0 = ndaddr;
 626 }
 627
 628 static void *
 629 send_thread(void *xctx)
 630 {
 631         struct send_thrctx *ctx = xctx;
 632         struct conn_ctx *timeo;
 633         struct kevent chg_evt;
 634         uint8_t *buf;
 635         int nconn = 0, kq, n, fd = -1, buflen;
 636         char name[32];
 637
 638         snprintf(name, sizeof(name), "snd%d", ctx->t_id);
 639         pthread_set_name_np(pthread_self(), name);
 640
 641         buflen = ctx->t_glob->g_buflen;
 642         buf = malloc(buflen);
 643         if (buf == NULL)
 644                 err(1, "malloc(%d) failed", buflen);
 645
 646         if (ctx->t_glob->g_sendfile) {
 647                 char filename[] = "sendtmpXXX";
 648
 649                 fd = mkstemp(filename);
 650                 if (fd < 0)
 651                         err(1, "mkstemp failed");
 652                 if (write(fd, buf, buflen) != buflen)
 653                         err(1, "write to file failed");
 654                 unlink(filename);
 655                 free(buf);
 656                 buf = NULL;
 657         }
 658
 659         kq = kqueue();
 660         if (kq < 0)
 661                 err(1, "kqueue failed");
 662
 663         /*
 664          * Establish the connections assigned to us and add the
 665          * established connections to kqueue.
 666          */
 667         for (;;) {
 668 #ifdef SEND_DEBUG
 669                 char addr_name[INET_ADDRSTRLEN];
 670 #endif
 671                 struct timeval readto;
 672                 struct conn_ctx *conn;
 673                 struct conn_ack ack;
 674                 int on;
 675
 676                 pthread_mutex_lock(&ctx->t_lock);
 677                 while (STAILQ_EMPTY(&ctx->t_conn))
 678                         pthread_cond_wait(&ctx->t_cond, &ctx->t_lock);
 679                 conn = STAILQ_FIRST(&ctx->t_conn);
 680                 STAILQ_REMOVE_HEAD(&ctx->t_conn, c_link);
 681                 pthread_mutex_unlock(&ctx->t_lock);
 682
 683                 if (conn->c_in.sin_port == 0) {
 684                         /*
 685                          * The marker for 'the end of connection list'.
 686                          * See the related comment in main thread.
 687                          *
 688                          * NOTE:
 689                          * We reuse the marker as the udata for the
 690                          * kqueue timer.
 691                          */
 692                         timeo = conn;
 693                         break;
 694                 }
 695
 696                 ++nconn;
 697 #ifdef SEND_DEBUG
 698                 fprintf(stderr, "%s %s:%d\n", name,
 699                     inet_ntop(AF_INET, &conn->c_in.sin_addr,
 700                         addr_name, sizeof(addr_name)),
 701                     ntohs(conn->c_in.sin_port));
 702 #endif
 703
 704 again:
 705                 conn->c_s = socket(AF_INET, SOCK_STREAM, 0);
 706                 if (conn->c_s < 0)
 707                         err(1, "socket failed");
 708
 709                 if (connect(conn->c_s, (const struct sockaddr *)&conn->c_in,
 710                     sizeof(conn->c_in)) < 0)
 711                         err(1, "connect failed");
 712
 713                 memset(&readto, 0, sizeof(readto));
 714                 readto.tv_sec = ctx->t_glob->g_readto_ms / 1000;
 715                 readto.tv_usec = (ctx->t_glob->g_readto_ms % 1000) * 1000;
 716                 if (setsockopt(conn->c_s, SOL_SOCKET, SO_RCVTIMEO, &readto,
 717                     sizeof(readto)) < 0)
 718                         err(1, "setsockopt(RCVTIMEO) failed");
 719
 720                 n = read(conn->c_s, &ack, sizeof(ack));
 721                 if (n != sizeof(ack)) {
 722                         if (n < 0) {
 723                                 if (errno == EAGAIN) {
 724                                         close(conn->c_s);
 725                                         goto again;
 726                                 }
 727                                 err(1, "read ack failed");
 728                         } else {
 729                                 errx(1, "read truncated ack");
 730                         }
 731                 }
 732
 733                 on = 1;
 734                 if (ioctl(conn->c_s, FIONBIO, &on, sizeof(on)) < 0)
 735                         err(1, "ioctl(FIONBIO) failed");
 736
 737                 EV_SET(&chg_evt, conn->c_s, EVFILT_WRITE, EV_ADD, 0, 0, conn);
 738                 n = kevent(kq, &chg_evt, 1, NULL, 0, NULL);
 739                 if (n < 0)
 740                         err(1, "kevent add failed");
 741         }
 742 #ifdef SEND_DEBUG
 743         fprintf(stderr, "%s conn %d\n", name, nconn);
 744 #endif
 745
 746         /*
 747          * Sender start sync, stage 1:
 748          * Wait for connections establishment (slow).
 749          */
 750         pthread_mutex_lock(&ctx->t_glob->g_lock);
 751         ctx->t_glob->g_nconn -= nconn;
 752         pthread_cond_broadcast(&ctx->t_glob->g_cond);
 753         while (ctx->t_glob->g_nconn != 0)
 754                 pthread_cond_wait(&ctx->t_glob->g_cond, &ctx->t_glob->g_lock);
 755         pthread_mutex_unlock(&ctx->t_glob->g_lock);
 756
 757         /*
 758          * Sender start sync, stage2.
 759          */
 760         /* Increase the g_nwait. */
 761         atomic_add_int(&ctx->t_glob->g_nwait, 1);
 762         /* Spin-wait for main thread to release us (reset g_nwait). */
 763         while (ctx->t_glob->g_nwait)
 764                 send_spinwait();
 765
 766 #ifdef SEND_DEBUG
 767         fprintf(stderr, "%s start\n", name);
 768 #endif
 769
 770         /*
 771          * Wire a kqueue timer, so that the sending can be terminated
 772          * as requested.
 773          *
 774          * NOTE:
 775          * Set -2 to c_s for timer udata, so we could distinguish it
 776          * from real connections.
 777          */
 778         timeo->c_s = -2;
 779         EV_SET(&chg_evt, 0, EVFILT_TIMER, EV_ADD | EV_ONESHOT, 0,
 780             ctx->t_glob->g_dur * 1000L, timeo);
 781         n = kevent(kq, &chg_evt, 1, NULL, 0, NULL);
 782         if (n < 0)
 783                 err(1, "kevent add failed");
 784
 785         clock_gettime(CLOCK_MONOTONIC_PRECISE, &ctx->t_start);
 786         for (;;) {
 787                 struct kevent evt[SEND_EVENT_MAX];
 788                 int nevt, i;
 789
 790                 nevt = kevent(kq, NULL, 0, evt, SEND_EVENT_MAX, NULL);
 791                 if (nevt < 0)
 792                         err(1, "kevent failed");
 793
 794                 for (i = 0; i < nevt; ++i) {
 795                         struct conn_ctx *conn = evt[i].udata;
 796
 797                         if (conn->c_s < 0) {
 798                                 if (conn->c_s == -2) {
 799                                         /* Timer expired */
 800                                         goto done;
 801                                 }
 802                                 continue;
 803                         }
 804
 805                         if (fd >= 0) {
 806                                 off_t m, off;
 807                                 size_t len;
 808
 809                                 off = conn->c_stat % buflen;
 810                                 len = buflen - off;
 811
 812                                 n = sendfile(fd, conn->c_s, off, len, NULL,
 813                                     &m, 0);
 814                                 if (n == 0 || (n < 0 && errno == EAGAIN))
 815                                         n = m;
 816                         } else {
 817                                 n = write(conn->c_s, buf, buflen);
 818                         }
 819
 820                         if (n < 0) {
 821                                 if (errno != EAGAIN) {
 822                                         conn->c_err = errno;
 823                                         clock_gettime(CLOCK_MONOTONIC_PRECISE,
 824                                             &conn->c_terr);
 825                                         close(conn->c_s);
 826                                         conn->c_s = -1;
 827                                 }
 828                         } else {
 829                                 conn->c_stat += n;
 830                         }
 831                 }
 832         }
 833 done:
 834         clock_gettime(CLOCK_MONOTONIC_PRECISE, &ctx->t_end);
 835
 836         if (fd >= 0)
 837                 close(fd);
 838         if (buf != NULL)
 839                 free(buf);
 840         return NULL;
 841 }