client.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <unistd.h>
   4 #include <limits.h>
   5 #include <errno.h>
   6 #include <fcntl.h>
   7 #include <sys/poll.h>
   8 #include <sys/types.h>
   9 #include <sys/stat.h>
  10 #include <sys/wait.h>
  11 #include <sys/socket.h>
  12 #include <sys/un.h>
  13 #include <netinet/in.h>
  14 #include <arpa/inet.h>
  15 #include <netdb.h>
  16 #include <signal.h>
  17
  18 #include "fio.h"
  19 #include "client.h"
  20 #include "server.h"
  21 #include "flist.h"
  22 #include "hash.h"
  23
  24 struct client_eta {
  25         struct jobs_eta eta;
  26         unsigned int pending;
  27 };
  28
  29 struct fio_client {
  30         struct flist_head list;
  31         struct flist_head hash_list;
  32         struct flist_head arg_list;
  33         union {
  34                 struct sockaddr_in addr;
  35                 struct sockaddr_in6 addr6;
  36                 struct sockaddr_un addr_un;
  37         };
  38         char *hostname;
  39         int port;
  40         int fd;
  41
  42         char *name;
  43
  44         int state;
  45
  46         int skip_newline;
  47         int is_sock;
  48         int disk_stats_shown;
  49         unsigned int jobs;
  50         int error;
  51         int ipv6;
  52         int sent_job;
  53
  54         struct flist_head eta_list;
  55         struct client_eta *eta_in_flight;
  56
  57         struct flist_head cmd_list;
  58
  59         uint16_t argc;
  60         char **argv;
  61 };
  62
  63 static void fio_client_text_op(struct fio_client *client,
  64                 FILE *f, __u16 pdu_len, const char *buf)
  65 {
  66         const char *name;
  67         int fio_unused ret;
  68
  69         name = client->name ? client->name : client->hostname;
  70
  71         if (!client->skip_newline)
  72                 fprintf(f, "<%s> ", name);
  73         ret = fwrite(buf, pdu_len, 1, f);
  74         fflush(f);
  75         client->skip_newline = strchr(buf, '\n') == NULL;
  76 }
  77
  78 static void handle_du(struct fio_client *client, struct fio_net_cmd *cmd);
  79 static void handle_ts(struct fio_net_cmd *cmd);
  80 static void handle_gs(struct fio_net_cmd *cmd);
  81 static void handle_eta(struct fio_client *client, struct fio_net_cmd *cmd);
  82 static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd);
  83
  84 struct client_ops fio_client_ops = {
  85         fio_client_text_op,
  86         handle_du,
  87         handle_ts,
  88         handle_gs,
  89         handle_eta,
  90         handle_probe,
  91 };
  92
  93 static struct timeval eta_tv;
  94
  95 enum {
  96         Client_created          = 0,
  97         Client_connected        = 1,
  98         Client_started          = 2,
  99         Client_running          = 3,
 100         Client_stopped          = 4,
 101         Client_exited           = 5,
 102 };
 103
 104 static FLIST_HEAD(client_list);
 105 static FLIST_HEAD(eta_list);
 106
 107 static FLIST_HEAD(arg_list);
 108
 109 static struct thread_stat client_ts;
 110 static struct group_run_stats client_gs;
 111 static int sum_stat_clients;
 112 static int sum_stat_nr;
 113
 114 #define FIO_CLIENT_HASH_BITS    7
 115 #define FIO_CLIENT_HASH_SZ      (1 << FIO_CLIENT_HASH_BITS)
 116 #define FIO_CLIENT_HASH_MASK    (FIO_CLIENT_HASH_SZ - 1)
 117 static struct flist_head client_hash[FIO_CLIENT_HASH_SZ];
 118
 119 static int handle_client(struct fio_client *client, struct client_ops *ops);
 120 static void dec_jobs_eta(struct client_eta *eta);
 121
 122 static void fio_client_add_hash(struct fio_client *client)
 123 {
 124         int bucket = hash_long(client->fd, FIO_CLIENT_HASH_BITS);
 125
 126         bucket &= FIO_CLIENT_HASH_MASK;
 127         flist_add(&client->hash_list, &client_hash[bucket]);
 128 }
 129
 130 static void fio_client_remove_hash(struct fio_client *client)
 131 {
 132         if (!flist_empty(&client->hash_list))
 133                 flist_del_init(&client->hash_list);
 134 }
 135
 136 static void fio_init fio_client_hash_init(void)
 137 {
 138         int i;
 139
 140         for (i = 0; i < FIO_CLIENT_HASH_SZ; i++)
 141                 INIT_FLIST_HEAD(&client_hash[i]);
 142 }
 143
 144 static struct fio_client *find_client_by_fd(int fd)
 145 {
 146         int bucket = hash_long(fd, FIO_CLIENT_HASH_BITS) & FIO_CLIENT_HASH_MASK;
 147         struct fio_client *client;
 148         struct flist_head *entry;
 149
 150         flist_for_each(entry, &client_hash[bucket]) {
 151                 client = flist_entry(entry, struct fio_client, hash_list);
 152
 153                 if (client->fd == fd)
 154                         return client;
 155         }
 156
 157         return NULL;
 158 }
 159
 160 static void remove_client(struct fio_client *client)
 161 {
 162         dprint(FD_NET, "client: removed <%s>\n", client->hostname);
 163         flist_del(&client->list);
 164
 165         fio_client_remove_hash(client);
 166
 167         if (!flist_empty(&client->eta_list)) {
 168                 flist_del_init(&client->eta_list);
 169                 dec_jobs_eta(client->eta_in_flight);
 170         }
 171
 172         free(client->hostname);
 173         if (client->argv)
 174                 free(client->argv);
 175         if (client->name)
 176                 free(client->name);
 177
 178         free(client);
 179         nr_clients--;
 180         sum_stat_clients--;
 181 }
 182
 183 static void __fio_client_add_cmd_option(struct fio_client *client,
 184                                         const char *opt)
 185 {
 186         int index;
 187
 188         index = client->argc++;
 189         client->argv = realloc(client->argv, sizeof(char *) * client->argc);
 190         client->argv[index] = strdup(opt);
 191         dprint(FD_NET, "client: add cmd %d: %s\n", index, opt);
 192 }
 193
 194 void fio_client_add_cmd_option(void *cookie, const char *opt)
 195 {
 196         struct fio_client *client = cookie;
 197         struct flist_head *entry;
 198
 199         if (!client || !opt)
 200                 return;
 201
 202         __fio_client_add_cmd_option(client, opt);
 203
 204         /*
 205          * Duplicate arguments to shared client group
 206          */
 207         flist_for_each(entry, &arg_list) {
 208                 client = flist_entry(entry, struct fio_client, arg_list);
 209
 210                 __fio_client_add_cmd_option(client, opt);
 211         }
 212 }
 213
 214 int fio_client_add(const char *hostname, void **cookie)
 215 {
 216         struct fio_client *existing = *cookie;
 217         struct fio_client *client;
 218
 219         if (existing) {
 220                 /*
 221                  * We always add our "exec" name as the option, hence 1
 222                  * means empty.
 223                  */
 224                 if (existing->argc == 1)
 225                         flist_add_tail(&existing->arg_list, &arg_list);
 226                 else {
 227                         while (!flist_empty(&arg_list))
 228                                 flist_del_init(arg_list.next);
 229                 }
 230         }
 231
 232         client = malloc(sizeof(*client));
 233         memset(client, 0, sizeof(*client));
 234
 235         INIT_FLIST_HEAD(&client->list);
 236         INIT_FLIST_HEAD(&client->hash_list);
 237         INIT_FLIST_HEAD(&client->arg_list);
 238         INIT_FLIST_HEAD(&client->eta_list);
 239         INIT_FLIST_HEAD(&client->cmd_list);
 240
 241         if (fio_server_parse_string(hostname, &client->hostname,
 242                                         &client->is_sock, &client->port,
 243                                         &client->addr.sin_addr,
 244                                         &client->addr6.sin6_addr,
 245                                         &client->ipv6))
 246                 return -1;
 247
 248         client->fd = -1;
 249
 250         __fio_client_add_cmd_option(client, "fio");
 251
 252         flist_add(&client->list, &client_list);
 253         nr_clients++;
 254         dprint(FD_NET, "client: added <%s>\n", client->hostname);
 255         *cookie = client;
 256         return 0;
 257 }
 258
 259 static int fio_client_connect_ip(struct fio_client *client)
 260 {
 261         struct sockaddr *addr;
 262         fio_socklen_t socklen;
 263         int fd, domain;
 264
 265         if (client->ipv6) {
 266                 client->addr6.sin6_family = AF_INET6;
 267                 client->addr6.sin6_port = htons(client->port);
 268                 domain = AF_INET6;
 269                 addr = (struct sockaddr *) &client->addr6;
 270                 socklen = sizeof(client->addr6);
 271         } else {
 272                 client->addr.sin_family = AF_INET;
 273                 client->addr.sin_port = htons(client->port);
 274                 domain = AF_INET;
 275                 addr = (struct sockaddr *) &client->addr;
 276                 socklen = sizeof(client->addr);
 277         }
 278
 279         fd = socket(domain, SOCK_STREAM, 0);
 280         if (fd < 0) {
 281                 log_err("fio: socket: %s\n", strerror(errno));
 282                 return -1;
 283         }
 284
 285         if (connect(fd, addr, socklen) < 0) {
 286                 log_err("fio: connect: %s\n", strerror(errno));
 287                 log_err("fio: failed to connect to %s:%u\n", client->hostname,
 288                                                                 client->port);
 289                 close(fd);
 290                 return -1;
 291         }
 292
 293         return fd;
 294 }
 295
 296 static int fio_client_connect_sock(struct fio_client *client)
 297 {
 298         struct sockaddr_un *addr = &client->addr_un;
 299         fio_socklen_t len;
 300         int fd;
 301
 302         memset(addr, 0, sizeof(*addr));
 303         addr->sun_family = AF_UNIX;
 304         strcpy(addr->sun_path, client->hostname);
 305
 306         fd = socket(AF_UNIX, SOCK_STREAM, 0);
 307         if (fd < 0) {
 308                 log_err("fio: socket: %s\n", strerror(errno));
 309                 return -1;
 310         }
 311
 312         len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
 313         if (connect(fd, (struct sockaddr *) addr, len) < 0) {
 314                 log_err("fio: connect; %s\n", strerror(errno));
 315                 close(fd);
 316                 return -1;
 317         }
 318
 319         return fd;
 320 }
 321
 322 static int fio_client_connect(struct fio_client *client)
 323 {
 324         int fd;
 325
 326         dprint(FD_NET, "client: connect to host %s\n", client->hostname);
 327
 328         if (client->is_sock)
 329                 fd = fio_client_connect_sock(client);
 330         else
 331                 fd = fio_client_connect_ip(client);
 332
 333         dprint(FD_NET, "client: %s connected %d\n", client->hostname, fd);
 334
 335         if (fd < 0)
 336                 return 1;
 337
 338         client->fd = fd;
 339         fio_client_add_hash(client);
 340         client->state = Client_connected;
 341         return 0;
 342 }
 343
 344 void fio_clients_terminate(void)
 345 {
 346         struct flist_head *entry;
 347         struct fio_client *client;
 348
 349         dprint(FD_NET, "client: terminate clients\n");
 350
 351         flist_for_each(entry, &client_list) {
 352                 client = flist_entry(entry, struct fio_client, list);
 353
 354                 fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0, NULL);
 355         }
 356 }
 357
 358 static void sig_int(int sig)
 359 {
 360         dprint(FD_NET, "client: got signal %d\n", sig);
 361         fio_clients_terminate();
 362 }
 363
 364 static void client_signal_handler(void)
 365 {
 366         struct sigaction act;
 367
 368         memset(&act, 0, sizeof(act));
 369         act.sa_handler = sig_int;
 370         act.sa_flags = SA_RESTART;
 371         sigaction(SIGINT, &act, NULL);
 372
 373         memset(&act, 0, sizeof(act));
 374         act.sa_handler = sig_int;
 375         act.sa_flags = SA_RESTART;
 376         sigaction(SIGTERM, &act, NULL);
 377 }
 378
 379 static void probe_client(struct fio_client *client)
 380 {
 381         dprint(FD_NET, "client: send probe\n");
 382
 383         fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0, &client->cmd_list);
 384 }
 385
 386 static int send_client_cmd_line(struct fio_client *client)
 387 {
 388         struct cmd_single_line_pdu *cslp;
 389         struct cmd_line_pdu *clp;
 390         unsigned long offset;
 391         unsigned int *lens;
 392         void *pdu;
 393         size_t mem;
 394         int i, ret;
 395
 396         dprint(FD_NET, "client: send cmdline %d\n", client->argc);
 397
 398         lens = malloc(client->argc * sizeof(unsigned int));
 399
 400         /*
 401          * Find out how much mem we need
 402          */
 403         for (i = 0, mem = 0; i < client->argc; i++) {
 404                 lens[i] = strlen(client->argv[i]) + 1;
 405                 mem += lens[i];
 406         }
 407
 408         /*
 409          * We need one cmd_line_pdu, and argc number of cmd_single_line_pdu
 410          */
 411         mem += sizeof(*clp) + (client->argc * sizeof(*cslp));
 412
 413         pdu = malloc(mem);
 414         clp = pdu;
 415         offset = sizeof(*clp);
 416
 417         for (i = 0; i < client->argc; i++) {
 418                 uint16_t arg_len = lens[i];
 419
 420                 cslp = pdu + offset;
 421                 strcpy((char *) cslp->text, client->argv[i]);
 422                 cslp->len = cpu_to_le16(arg_len);
 423                 offset += sizeof(*cslp) + arg_len;
 424         }
 425
 426         free(lens);
 427         clp->lines = cpu_to_le16(client->argc);
 428         ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
 429         free(pdu);
 430         return ret;
 431 }
 432
 433 int fio_clients_connect(void)
 434 {
 435         struct fio_client *client;
 436         struct flist_head *entry, *tmp;
 437         int ret;
 438
 439 #ifdef WIN32
 440         WSADATA wsd;
 441         WSAStartup(MAKEWORD(2,2), &wsd);
 442 #endif
 443
 444         dprint(FD_NET, "client: connect all\n");
 445
 446         client_signal_handler();
 447
 448         flist_for_each_safe(entry, tmp, &client_list) {
 449                 client = flist_entry(entry, struct fio_client, list);
 450
 451                 ret = fio_client_connect(client);
 452                 if (ret) {
 453                         remove_client(client);
 454                         continue;
 455                 }
 456
 457                 probe_client(client);
 458
 459                 if (client->argc > 1)
 460                         send_client_cmd_line(client);
 461         }
 462
 463         return !nr_clients;
 464 }
 465
 466 /*
 467  * Send file contents to server backend. We could use sendfile(), but to remain
 468  * more portable lets just read/write the darn thing.
 469  */
 470 static int fio_client_send_ini(struct fio_client *client, const char *filename)
 471 {
 472         struct stat sb;
 473         char *p, *buf;
 474         off_t len;
 475         int fd, ret;
 476
 477         dprint(FD_NET, "send ini %s to %s\n", filename, client->hostname);
 478
 479         fd = open(filename, O_RDONLY);
 480         if (fd < 0) {
 481                 log_err("fio: job file <%s> open: %s\n", filename, strerror(errno));
 482                 return 1;
 483         }
 484
 485         if (fstat(fd, &sb) < 0) {
 486                 log_err("fio: job file stat: %s\n", strerror(errno));
 487                 close(fd);
 488                 return 1;
 489         }
 490
 491         buf = malloc(sb.st_size);
 492
 493         len = sb.st_size;
 494         p = buf;
 495         do {
 496                 ret = read(fd, p, len);
 497                 if (ret > 0) {
 498                         len -= ret;
 499                         if (!len)
 500                                 break;
 501                         p += ret;
 502                         continue;
 503                 } else if (!ret)
 504                         break;
 505                 else if (errno == EAGAIN || errno == EINTR)
 506                         continue;
 507         } while (1);
 508
 509         if (len) {
 510                 log_err("fio: failed reading job file %s\n", filename);
 511                 close(fd);
 512                 free(buf);
 513                 return 1;
 514         }
 515
 516         client->sent_job = 1;
 517         ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size, 0);
 518         free(buf);
 519         close(fd);
 520         return ret;
 521 }
 522
 523 int fio_clients_send_ini(const char *filename)
 524 {
 525         struct fio_client *client;
 526         struct flist_head *entry, *tmp;
 527
 528         flist_for_each_safe(entry, tmp, &client_list) {
 529                 client = flist_entry(entry, struct fio_client, list);
 530
 531                 if (fio_client_send_ini(client, filename))
 532                         remove_client(client);
 533
 534                 client->sent_job = 1;
 535         }
 536
 537         return !nr_clients;
 538 }
 539
 540 static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
 541 {
 542         dst->max_val    = le64_to_cpu(src->max_val);
 543         dst->min_val    = le64_to_cpu(src->min_val);
 544         dst->samples    = le64_to_cpu(src->samples);
 545
 546         /*
 547          * Floats arrive as IEEE 754 encoded uint64_t, convert back to double
 548          */
 549         dst->mean.u.f   = fio_uint64_to_double(le64_to_cpu(dst->mean.u.i));
 550         dst->S.u.f      = fio_uint64_to_double(le64_to_cpu(dst->S.u.i));
 551 }
 552
 553 static void convert_ts(struct thread_stat *dst, struct thread_stat *src)
 554 {
 555         int i, j;
 556
 557         dst->error      = le32_to_cpu(src->error);
 558         dst->groupid    = le32_to_cpu(src->groupid);
 559         dst->pid        = le32_to_cpu(src->pid);
 560         dst->members    = le32_to_cpu(src->members);
 561
 562         for (i = 0; i < 2; i++) {
 563                 convert_io_stat(&dst->clat_stat[i], &src->clat_stat[i]);
 564                 convert_io_stat(&dst->slat_stat[i], &src->slat_stat[i]);
 565                 convert_io_stat(&dst->lat_stat[i], &src->lat_stat[i]);
 566                 convert_io_stat(&dst->bw_stat[i], &src->bw_stat[i]);
 567         }
 568
 569         dst->usr_time           = le64_to_cpu(src->usr_time);
 570         dst->sys_time           = le64_to_cpu(src->sys_time);
 571         dst->ctx                = le64_to_cpu(src->ctx);
 572         dst->minf               = le64_to_cpu(src->minf);
 573         dst->majf               = le64_to_cpu(src->majf);
 574         dst->clat_percentiles   = le64_to_cpu(src->clat_percentiles);
 575
 576         for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) {
 577                 fio_fp64_t *fps = &src->percentile_list[i];
 578                 fio_fp64_t *fpd = &dst->percentile_list[i];
 579
 580                 fpd->u.f = fio_uint64_to_double(le64_to_cpu(fps->u.i));
 581         }
 582
 583         for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
 584                 dst->io_u_map[i]        = le32_to_cpu(src->io_u_map[i]);
 585                 dst->io_u_submit[i]     = le32_to_cpu(src->io_u_submit[i]);
 586                 dst->io_u_complete[i]   = le32_to_cpu(src->io_u_complete[i]);
 587         }
 588
 589         for (i = 0; i < FIO_IO_U_LAT_U_NR; i++) {
 590                 dst->io_u_lat_u[i]      = le32_to_cpu(src->io_u_lat_u[i]);
 591                 dst->io_u_lat_m[i]      = le32_to_cpu(src->io_u_lat_m[i]);
 592         }
 593
 594         for (i = 0; i < 2; i++)
 595                 for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
 596                         dst->io_u_plat[i][j] = le32_to_cpu(src->io_u_plat[i][j]);
 597
 598         for (i = 0; i < 3; i++) {
 599                 dst->total_io_u[i]      = le64_to_cpu(src->total_io_u[i]);
 600                 dst->short_io_u[i]      = le64_to_cpu(src->short_io_u[i]);
 601         }
 602
 603         dst->total_submit       = le64_to_cpu(src->total_submit);
 604         dst->total_complete     = le64_to_cpu(src->total_complete);
 605
 606         for (i = 0; i < 2; i++) {
 607                 dst->io_bytes[i]        = le64_to_cpu(src->io_bytes[i]);
 608                 dst->runtime[i]         = le64_to_cpu(src->runtime[i]);
 609         }
 610
 611         dst->total_run_time     = le64_to_cpu(src->total_run_time);
 612         dst->continue_on_error  = le16_to_cpu(src->continue_on_error);
 613         dst->total_err_count    = le64_to_cpu(src->total_err_count);
 614         dst->first_error        = le32_to_cpu(src->first_error);
 615         dst->kb_base            = le32_to_cpu(src->kb_base);
 616 }
 617
 618 static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
 619 {
 620         int i;
 621
 622         for (i = 0; i < 2; i++) {
 623                 dst->max_run[i]         = le64_to_cpu(src->max_run[i]);
 624                 dst->min_run[i]         = le64_to_cpu(src->min_run[i]);
 625                 dst->max_bw[i]          = le64_to_cpu(src->max_bw[i]);
 626                 dst->min_bw[i]          = le64_to_cpu(src->min_bw[i]);
 627                 dst->io_kb[i]           = le64_to_cpu(src->io_kb[i]);
 628                 dst->agg[i]             = le64_to_cpu(src->agg[i]);
 629         }
 630
 631         dst->kb_base    = le32_to_cpu(src->kb_base);
 632         dst->groupid    = le32_to_cpu(src->groupid);
 633 }
 634
 635 static void handle_ts(struct fio_net_cmd *cmd)
 636 {
 637         struct cmd_ts_pdu *p = (struct cmd_ts_pdu *) cmd->payload;
 638
 639         convert_ts(&p->ts, &p->ts);
 640         convert_gs(&p->rs, &p->rs);
 641
 642         show_thread_status(&p->ts, &p->rs);
 643
 644         if (sum_stat_clients == 1)
 645                 return;
 646
 647         sum_thread_stats(&client_ts, &p->ts, sum_stat_nr);
 648         sum_group_stats(&client_gs, &p->rs);
 649
 650         client_ts.members++;
 651         client_ts.groupid = p->ts.groupid;
 652
 653         if (++sum_stat_nr == sum_stat_clients) {
 654                 strcpy(client_ts.name, "All clients");
 655                 show_thread_status(&client_ts, &client_gs);
 656         }
 657 }
 658
 659 static void handle_gs(struct fio_net_cmd *cmd)
 660 {
 661         struct group_run_stats *gs = (struct group_run_stats *) cmd->payload;
 662
 663         convert_gs(gs, gs);
 664         show_group_stats(gs);
 665 }
 666
 667 static void convert_agg(struct disk_util_agg *agg)
 668 {
 669         int i;
 670
 671         for (i = 0; i < 2; i++) {
 672                 agg->ios[i]     = le32_to_cpu(agg->ios[i]);
 673                 agg->merges[i]  = le32_to_cpu(agg->merges[i]);
 674                 agg->sectors[i] = le64_to_cpu(agg->sectors[i]);
 675                 agg->ticks[i]   = le32_to_cpu(agg->ticks[i]);
 676         }
 677
 678         agg->io_ticks           = le32_to_cpu(agg->io_ticks);
 679         agg->time_in_queue      = le32_to_cpu(agg->time_in_queue);
 680         agg->slavecount         = le32_to_cpu(agg->slavecount);
 681         agg->max_util.u.f       = fio_uint64_to_double(__le64_to_cpu(agg->max_util.u.i));
 682 }
 683
 684 static void convert_dus(struct disk_util_stat *dus)
 685 {
 686         int i;
 687
 688         for (i = 0; i < 2; i++) {
 689                 dus->ios[i]     = le32_to_cpu(dus->ios[i]);
 690                 dus->merges[i]  = le32_to_cpu(dus->merges[i]);
 691                 dus->sectors[i] = le64_to_cpu(dus->sectors[i]);
 692                 dus->ticks[i]   = le32_to_cpu(dus->ticks[i]);
 693         }
 694
 695         dus->io_ticks           = le32_to_cpu(dus->io_ticks);
 696         dus->time_in_queue      = le32_to_cpu(dus->time_in_queue);
 697         dus->msec               = le64_to_cpu(dus->msec);
 698 }
 699
 700 static void handle_du(struct fio_client *client, struct fio_net_cmd *cmd)
 701 {
 702         struct cmd_du_pdu *du = (struct cmd_du_pdu *) cmd->payload;
 703
 704         convert_dus(&du->dus);
 705         convert_agg(&du->agg);
 706
 707         if (!client->disk_stats_shown) {
 708                 client->disk_stats_shown = 1;
 709                 log_info("\nDisk stats (read/write):\n");
 710         }
 711
 712         print_disk_util(&du->dus, &du->agg, terse_output);
 713 }
 714
 715 static void convert_jobs_eta(struct jobs_eta *je)
 716 {
 717         int i;
 718
 719         je->nr_running          = le32_to_cpu(je->nr_running);
 720         je->nr_ramp             = le32_to_cpu(je->nr_ramp);
 721         je->nr_pending          = le32_to_cpu(je->nr_pending);
 722         je->files_open          = le32_to_cpu(je->files_open);
 723         je->m_rate              = le32_to_cpu(je->m_rate);
 724         je->t_rate              = le32_to_cpu(je->t_rate);
 725         je->m_iops              = le32_to_cpu(je->m_iops);
 726         je->t_iops              = le32_to_cpu(je->t_iops);
 727
 728         for (i = 0; i < 2; i++) {
 729                 je->rate[i]     = le32_to_cpu(je->rate[i]);
 730                 je->iops[i]     = le32_to_cpu(je->iops[i]);
 731         }
 732
 733         je->elapsed_sec         = le64_to_cpu(je->elapsed_sec);
 734         je->eta_sec             = le64_to_cpu(je->eta_sec);
 735 }
 736
 737 static void sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je)
 738 {
 739         int i;
 740
 741         dst->nr_running         += je->nr_running;
 742         dst->nr_ramp            += je->nr_ramp;
 743         dst->nr_pending         += je->nr_pending;
 744         dst->files_open         += je->files_open;
 745         dst->m_rate             += je->m_rate;
 746         dst->t_rate             += je->t_rate;
 747         dst->m_iops             += je->m_iops;
 748         dst->t_iops             += je->t_iops;
 749
 750         for (i = 0; i < 2; i++) {
 751                 dst->rate[i]    += je->rate[i];
 752                 dst->iops[i]    += je->iops[i];
 753         }
 754
 755         dst->elapsed_sec        += je->elapsed_sec;
 756
 757         if (je->eta_sec > dst->eta_sec)
 758                 dst->eta_sec = je->eta_sec;
 759 }
 760
 761 static void dec_jobs_eta(struct client_eta *eta)
 762 {
 763         if (!--eta->pending) {
 764                 display_thread_status(&eta->eta);
 765                 free(eta);
 766         }
 767 }
 768
 769 static void remove_reply_cmd(struct fio_client *client, struct fio_net_cmd *cmd)
 770 {
 771         struct fio_net_int_cmd *icmd = NULL;
 772         struct flist_head *entry;
 773
 774         flist_for_each(entry, &client->cmd_list) {
 775                 icmd = flist_entry(entry, struct fio_net_int_cmd, list);
 776
 777                 if (cmd->tag == (uintptr_t) icmd)
 778                         break;
 779
 780                 icmd = NULL;
 781         }
 782
 783         if (!icmd) {
 784                 log_err("fio: client: unable to find matching tag\n");
 785                 return;
 786         }
 787
 788         flist_del(&icmd->list);
 789         cmd->tag = icmd->saved_tag;
 790         free(icmd);
 791 }
 792
 793 static void handle_eta(struct fio_client *client, struct fio_net_cmd *cmd)
 794 {
 795         struct jobs_eta *je = (struct jobs_eta *) cmd->payload;
 796         struct client_eta *eta = (struct client_eta *) (uintptr_t) cmd->tag;
 797
 798         dprint(FD_NET, "client: got eta tag %p, %d\n", eta, eta->pending);
 799
 800         assert(client->eta_in_flight == eta);
 801
 802         client->eta_in_flight = NULL;
 803         flist_del_init(&client->eta_list);
 804
 805         convert_jobs_eta(je);
 806         sum_jobs_eta(&eta->eta, je);
 807         dec_jobs_eta(eta);
 808 }
 809
 810 static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
 811 {
 812         struct cmd_probe_pdu *probe = (struct cmd_probe_pdu *) cmd->payload;
 813         const char *os, *arch;
 814         char bit[16];
 815
 816         os = fio_get_os_string(probe->os);
 817         if (!os)
 818                 os = "unknown";
 819
 820         arch = fio_get_arch_string(probe->arch);
 821         if (!arch)
 822                 os = "unknown";
 823
 824         sprintf(bit, "%d-bit", probe->bpp * 8);
 825
 826         log_info("hostname=%s, be=%u, %s, os=%s, arch=%s, fio=%u.%u.%u\n",
 827                 probe->hostname, probe->bigendian, bit, os, arch,
 828                 probe->fio_major, probe->fio_minor, probe->fio_patch);
 829
 830         if (!client->name)
 831                 client->name = strdup((char *) probe->hostname);
 832 }
 833
 834 static void handle_start(struct fio_client *client, struct fio_net_cmd *cmd)
 835 {
 836         struct cmd_start_pdu *pdu = (struct cmd_start_pdu *) cmd->payload;
 837
 838         client->state = Client_started;
 839         client->jobs = le32_to_cpu(pdu->jobs);
 840 }
 841
 842 static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd)
 843 {
 844         struct cmd_end_pdu *pdu = (struct cmd_end_pdu *) cmd->payload;
 845
 846         client->state = Client_stopped;
 847         client->error = le32_to_cpu(pdu->error);
 848
 849         if (client->error)
 850                 log_info("client <%s>: exited with error %d\n", client->hostname, client->error);
 851 }
 852
 853 static int handle_client(struct fio_client *client, struct client_ops *ops)
 854 {
 855         struct fio_net_cmd *cmd;
 856
 857         dprint(FD_NET, "client: handle %s\n", client->hostname);
 858
 859         cmd = fio_net_recv_cmd(client->fd);
 860         if (!cmd)
 861                 return 0;
 862
 863         dprint(FD_NET, "client: got cmd op %s from %s\n",
 864                                 fio_server_op(cmd->opcode), client->hostname);
 865
 866         switch (cmd->opcode) {
 867         case FIO_NET_CMD_QUIT:
 868                 remove_client(client);
 869                 free(cmd);
 870                 break;
 871         case FIO_NET_CMD_TEXT: {
 872                 const char *buf = (const char *) cmd->payload;
 873                 ops->text_op(client, f_out, cmd->pdu_len, buf);
 874                 free(cmd);
 875                 break;
 876                 }
 877         case FIO_NET_CMD_DU:
 878                 ops->disk_util(client, cmd);
 879                 free(cmd);
 880                 break;
 881         case FIO_NET_CMD_TS:
 882                 ops->thread_status(cmd);
 883                 free(cmd);
 884                 break;
 885         case FIO_NET_CMD_GS:
 886                 ops->group_stats(cmd);
 887                 free(cmd);
 888                 break;
 889         case FIO_NET_CMD_ETA:
 890                 remove_reply_cmd(client, cmd);
 891                 ops->eta(client, cmd);
 892                 free(cmd);
 893                 break;
 894         case FIO_NET_CMD_PROBE:
 895                 remove_reply_cmd(client, cmd);
 896                 ops->probe(client, cmd);
 897                 free(cmd);
 898                 break;
 899         case FIO_NET_CMD_RUN:
 900                 client->state = Client_running;
 901                 free(cmd);
 902                 break;
 903         case FIO_NET_CMD_START:
 904                 handle_start(client, cmd);
 905                 free(cmd);
 906                 break;
 907         case FIO_NET_CMD_STOP:
 908                 handle_stop(client, cmd);
 909                 free(cmd);
 910                 break;
 911         default:
 912                 log_err("fio: unknown client op: %s\n", fio_server_op(cmd->opcode));
 913                 free(cmd);
 914                 break;
 915         }
 916
 917         return 1;
 918 }
 919
 920 static void request_client_etas(void)
 921 {
 922         struct fio_client *client;
 923         struct flist_head *entry;
 924         struct client_eta *eta;
 925         int skipped = 0;
 926
 927         dprint(FD_NET, "client: request eta (%d)\n", nr_clients);
 928
 929         eta = malloc(sizeof(*eta));
 930         memset(&eta->eta, 0, sizeof(eta->eta));
 931         eta->pending = nr_clients;
 932
 933         flist_for_each(entry, &client_list) {
 934                 client = flist_entry(entry, struct fio_client, list);
 935
 936                 if (!flist_empty(&client->eta_list)) {
 937                         skipped++;
 938                         continue;
 939                 }
 940                 if (client->state != Client_running)
 941                         continue;
 942
 943                 assert(!client->eta_in_flight);
 944                 flist_add_tail(&client->eta_list, &eta_list);
 945                 client->eta_in_flight = eta;
 946                 fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_SEND_ETA,
 947                                         (uintptr_t) eta, &client->cmd_list);
 948         }
 949
 950         while (skipped--)
 951                 dec_jobs_eta(eta);
 952
 953         dprint(FD_NET, "client: requested eta tag %p\n", eta);
 954 }
 955
 956 static int client_check_cmd_timeout(struct fio_client *client,
 957                                     struct timeval *now)
 958 {
 959         struct fio_net_int_cmd *cmd;
 960         struct flist_head *entry, *tmp;
 961         int ret = 0;
 962
 963         flist_for_each_safe(entry, tmp, &client->cmd_list) {
 964                 cmd = flist_entry(entry, struct fio_net_int_cmd, list);
 965
 966                 if (mtime_since(&cmd->tv, now) < FIO_NET_CLIENT_TIMEOUT)
 967                         continue;
 968
 969                 log_err("fio: client %s, timeout on cmd %s\n", client->hostname,
 970                                                 fio_server_op(cmd->cmd.opcode));
 971                 flist_del(&cmd->list);
 972                 free(cmd);
 973                 ret = 1;
 974         }
 975
 976         return flist_empty(&client->cmd_list) && ret;
 977 }
 978
 979 static int fio_client_timed_out(void)
 980 {
 981         struct fio_client *client;
 982         struct flist_head *entry, *tmp;
 983         struct timeval tv;
 984         int ret = 0;
 985
 986         gettimeofday(&tv, NULL);
 987
 988         flist_for_each_safe(entry, tmp, &client_list) {
 989                 client = flist_entry(entry, struct fio_client, list);
 990
 991                 if (flist_empty(&client->cmd_list))
 992                         continue;
 993
 994                 if (!client_check_cmd_timeout(client, &tv))
 995                         continue;
 996
 997                 log_err("fio: client %s timed out\n", client->hostname);
 998                 remove_client(client);
 999                 ret = 1;
1000         }
1001
1002         return ret;
1003 }
1004
1005 int fio_handle_clients(struct client_ops *ops)
1006 {
1007         struct pollfd *pfds;
1008         int i, ret = 0, retval = 0;
1009
1010         gettimeofday(&eta_tv, NULL);
1011
1012         pfds = malloc(nr_clients * sizeof(struct pollfd));
1013
1014         sum_stat_clients = nr_clients;
1015         init_thread_stat(&client_ts);
1016         init_group_run_stat(&client_gs);
1017
1018         while (!exit_backend && nr_clients) {
1019                 struct flist_head *entry, *tmp;
1020                 struct fio_client *client;
1021
1022                 i = 0;
1023                 flist_for_each_safe(entry, tmp, &client_list) {
1024                         client = flist_entry(entry, struct fio_client, list);
1025
1026                         if (!client->sent_job &&
1027                             flist_empty(&client->cmd_list)) {
1028                                 remove_client(client);
1029                                 continue;
1030                         }
1031
1032                         pfds[i].fd = client->fd;
1033                         pfds[i].events = POLLIN;
1034                         i++;
1035                 }
1036
1037                 if (!nr_clients)
1038                         break;
1039
1040                 assert(i == nr_clients);
1041
1042                 do {
1043                         struct timeval tv;
1044
1045                         gettimeofday(&tv, NULL);
1046                         if (mtime_since(&eta_tv, &tv) >= 900) {
1047                                 request_client_etas();
1048                                 memcpy(&eta_tv, &tv, sizeof(tv));
1049
1050                                 if (fio_client_timed_out())
1051                                         break;
1052                         }
1053
1054                         ret = poll(pfds, nr_clients, 100);
1055                         if (ret < 0) {
1056                                 if (errno == EINTR)
1057                                         continue;
1058                                 log_err("fio: poll clients: %s\n", strerror(errno));
1059                                 break;
1060                         } else if (!ret)
1061                                 continue;
1062                 } while (ret <= 0);
1063
1064                 for (i = 0; i < nr_clients; i++) {
1065                         if (!(pfds[i].revents & POLLIN))
1066                                 continue;
1067
1068                         client = find_client_by_fd(pfds[i].fd);
1069                         if (!client) {
1070                                 log_err("fio: unknown client fd %d\n", pfds[i].fd);
1071                                 continue;
1072                         }
1073                         if (!handle_client(client, ops)) {
1074                                 log_info("client: host=%s disconnected\n",
1075                                                 client->hostname);
1076                                 remove_client(client);
1077                                 retval = 1;
1078                         } else if (client->error)
1079                                 retval = 1;
1080                 }
1081         }
1082
1083         free(pfds);
1084         return retval;
1085 }