client: sum running ETA of jobs
[fio.git] / 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 "server.h"
20 #include "flist.h"
21 #include "hash.h"
22
23 struct fio_client {
24         struct flist_head list;
25         struct flist_head hash_list;
26         struct sockaddr_in addr;
27         struct sockaddr_un addr_un;
28         char *hostname;
29         int port;
30         int fd;
31
32         char *name;
33
34         int state;
35         int skip_newline;
36         int is_sock;
37
38         uint16_t argc;
39         char **argv;
40 };
41
42 static struct jobs_eta client_etas;
43 static int received_etas;
44
45 enum {
46         Client_created          = 0,
47         Client_connected        = 1,
48         Client_started          = 2,
49         Client_stopped          = 3,
50         Client_exited           = 4,
51 };
52
53 static FLIST_HEAD(client_list);
54
55 #define FIO_CLIENT_HASH_BITS    7
56 #define FIO_CLIENT_HASH_SZ      (1 << FIO_CLIENT_HASH_BITS)
57 #define FIO_CLIENT_HASH_MASK    (FIO_CLIENT_HASH_SZ - 1)
58 static struct flist_head client_hash[FIO_CLIENT_HASH_SZ];
59
60 static int handle_client(struct fio_client *client);
61
62 static void fio_client_add_hash(struct fio_client *client)
63 {
64         int bucket = hash_long(client->fd, FIO_CLIENT_HASH_BITS);
65
66         bucket &= FIO_CLIENT_HASH_MASK;
67         flist_add(&client->hash_list, &client_hash[bucket]);
68 }
69
70 static void fio_client_remove_hash(struct fio_client *client)
71 {
72         if (!flist_empty(&client->hash_list))
73                 flist_del_init(&client->hash_list);
74 }
75
76 static void fio_init fio_client_hash_init(void)
77 {
78         int i;
79
80         for (i = 0; i < FIO_CLIENT_HASH_SZ; i++)
81                 INIT_FLIST_HEAD(&client_hash[i]);
82 }
83
84 static struct fio_client *find_client_by_fd(int fd)
85 {
86         int bucket = hash_long(fd, FIO_CLIENT_HASH_BITS) & FIO_CLIENT_HASH_MASK;
87         struct fio_client *client;
88         struct flist_head *entry;
89
90         flist_for_each(entry, &client_hash[bucket]) {
91                 client = flist_entry(entry, struct fio_client, hash_list);
92
93                 if (client->fd == fd)
94                         return client;
95         }
96
97         return NULL;
98 }
99
100 static void remove_client(struct fio_client *client)
101 {
102         dprint(FD_NET, "client: removed <%s>\n", client->hostname);
103         flist_del(&client->list);
104
105         fio_client_remove_hash(client);
106
107         free(client->hostname);
108         if (client->argv)
109                 free(client->argv);
110         if (client->name)
111                 free(client->name);
112
113         free(client);
114         nr_clients--;
115 }
116
117 static void __fio_client_add_cmd_option(struct fio_client *client,
118                                         const char *opt)
119 {
120         int index;
121
122         index = client->argc++;
123         client->argv = realloc(client->argv, sizeof(char *) * client->argc);
124         client->argv[index] = strdup(opt);
125         dprint(FD_NET, "client: add cmd %d: %s\n", index, opt);
126 }
127
128 void fio_client_add_cmd_option(void *cookie, const char *opt)
129 {
130         struct fio_client *client = cookie;
131
132         if (!client || !opt)
133                 return;
134
135         __fio_client_add_cmd_option(client, opt);
136 }
137
138 int fio_client_add(const char *hostname, void **cookie)
139 {
140         struct fio_client *client;
141
142         client = malloc(sizeof(*client));
143         memset(client, 0, sizeof(*client));
144
145         INIT_FLIST_HEAD(&client->list);
146         INIT_FLIST_HEAD(&client->hash_list);
147
148         if (fio_server_parse_string(hostname, &client->hostname,
149                                         &client->is_sock, &client->port,
150                                         &client->addr.sin_addr))
151                 return -1;
152
153         client->fd = -1;
154
155         __fio_client_add_cmd_option(client, "fio");
156
157         flist_add(&client->list, &client_list);
158         nr_clients++;
159         dprint(FD_NET, "client: added <%s>\n", client->hostname);
160         *cookie = client;
161         return 0;
162 }
163
164 static int fio_client_connect_ip(struct fio_client *client)
165 {
166         int fd;
167
168         client->addr.sin_family = AF_INET;
169         client->addr.sin_port = htons(client->port);
170
171         fd = socket(AF_INET, SOCK_STREAM, 0);
172         if (fd < 0) {
173                 log_err("fio: socket: %s\n", strerror(errno));
174                 return -1;
175         }
176
177         if (connect(fd, (struct sockaddr *) &client->addr, sizeof(client->addr)) < 0) {
178                 log_err("fio: connect: %s\n", strerror(errno));
179                 log_err("fio: failed to connect to %s:%u\n", client->hostname,
180                                                                 client->port);
181                 close(fd);
182                 return -1;
183         }
184
185         return fd;
186 }
187
188 static int fio_client_connect_sock(struct fio_client *client)
189 {
190         struct sockaddr_un *addr = &client->addr_un;
191         fio_socklen_t len;
192         int fd;
193
194         memset(addr, 0, sizeof(*addr));
195         addr->sun_family = AF_UNIX;
196         strcpy(addr->sun_path, client->hostname);
197
198         fd = socket(AF_UNIX, SOCK_STREAM, 0);
199         if (fd < 0) {
200                 log_err("fio: socket: %s\n", strerror(errno));
201                 return -1;
202         }
203
204         len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
205         if (connect(fd, (struct sockaddr *) addr, len) < 0) {
206                 log_err("fio: connect; %s\n", strerror(errno));
207                 close(fd);
208                 return -1;
209         }
210
211         return fd;
212 }
213
214 static int fio_client_connect(struct fio_client *client)
215 {
216         int fd;
217
218         dprint(FD_NET, "client: connect to host %s\n", client->hostname);
219
220         if (client->is_sock)
221                 fd = fio_client_connect_sock(client);
222         else
223                 fd = fio_client_connect_ip(client);
224
225         if (fd < 0)
226                 return 1;
227
228         client->fd = fd;
229         fio_client_add_hash(client);
230         client->state = Client_connected;
231         return 0;
232 }
233
234 void fio_clients_terminate(void)
235 {
236         struct flist_head *entry;
237         struct fio_client *client;
238
239         dprint(FD_NET, "client: terminate clients\n");
240
241         flist_for_each(entry, &client_list) {
242                 client = flist_entry(entry, struct fio_client, list);
243
244                 fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0);
245         }
246 }
247
248 static void sig_int(int sig)
249 {
250         dprint(FD_NET, "client: got signal %d\n", sig);
251         fio_clients_terminate();
252 }
253
254 static void client_signal_handler(void)
255 {
256         struct sigaction act;
257
258         memset(&act, 0, sizeof(act));
259         act.sa_handler = sig_int;
260         act.sa_flags = SA_RESTART;
261         sigaction(SIGINT, &act, NULL);
262
263         memset(&act, 0, sizeof(act));
264         act.sa_handler = sig_int;
265         act.sa_flags = SA_RESTART;
266         sigaction(SIGTERM, &act, NULL);
267 }
268
269 static void probe_client(struct fio_client *client)
270 {
271         dprint(FD_NET, "client: send probe\n");
272
273         fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0);
274         handle_client(client);
275 }
276
277 static int send_client_cmd_line(struct fio_client *client)
278 {
279         struct cmd_single_line_pdu *cslp;
280         struct cmd_line_pdu *clp;
281         unsigned long offset;
282         void *pdu;
283         size_t mem;
284         int i, ret;
285
286         dprint(FD_NET, "client: send cmdline %d\n", client->argc);
287
288         /*
289          * Find out how much mem we need
290          */
291         for (i = 0, mem = 0; i < client->argc; i++)
292                 mem += strlen(client->argv[i]) + 1;
293
294         /*
295          * We need one cmd_line_pdu, and argc number of cmd_single_line_pdu
296          */
297         mem += sizeof(*clp) + (client->argc * sizeof(*cslp));
298
299         pdu = malloc(mem);
300         clp = pdu;
301         offset = sizeof(*clp);
302
303         for (i = 0; i < client->argc; i++) {
304                 uint16_t arg_len = strlen(client->argv[i]) + 1;
305
306                 cslp = pdu + offset;
307                 strcpy((char *) cslp->text, client->argv[i]);
308                 cslp->len = cpu_to_le16(arg_len);
309                 offset += sizeof(*cslp) + arg_len;
310         }
311
312         clp->lines = cpu_to_le16(client->argc);
313         ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem);
314         free(pdu);
315         return ret;
316 }
317
318 int fio_clients_connect(void)
319 {
320         struct fio_client *client;
321         struct flist_head *entry, *tmp;
322         int ret;
323
324         dprint(FD_NET, "client: connect all\n");
325
326         client_signal_handler();
327
328         flist_for_each_safe(entry, tmp, &client_list) {
329                 client = flist_entry(entry, struct fio_client, list);
330
331                 ret = fio_client_connect(client);
332                 if (ret) {
333                         remove_client(client);
334                         continue;
335                 }
336
337                 probe_client(client);
338
339                 if (client->argc > 1)
340                         send_client_cmd_line(client);
341         }
342
343         return !nr_clients;
344 }
345
346 /*
347  * Send file contents to server backend. We could use sendfile(), but to remain
348  * more portable lets just read/write the darn thing.
349  */
350 static int fio_client_send_ini(struct fio_client *client, const char *filename)
351 {
352         struct stat sb;
353         char *p, *buf;
354         off_t len;
355         int fd, ret;
356
357         dprint(FD_NET, "send ini %s to %s\n", filename, client->hostname);
358
359         fd = open(filename, O_RDONLY);
360         if (fd < 0) {
361                 log_err("fio: job file <%s> open: %s\n", filename, strerror(errno));
362                 return 1;
363         }
364
365         if (fstat(fd, &sb) < 0) {
366                 log_err("fio: job file stat: %s\n", strerror(errno));
367                 close(fd);
368                 return 1;
369         }
370
371         buf = malloc(sb.st_size);
372
373         len = sb.st_size;
374         p = buf;
375         do {
376                 ret = read(fd, p, len);
377                 if (ret > 0) {
378                         len -= ret;
379                         if (!len)
380                                 break;
381                         p += ret;
382                         continue;
383                 } else if (!ret)
384                         break;
385                 else if (errno == EAGAIN || errno == EINTR)
386                         continue;
387         } while (1);
388
389         if (len) {
390                 log_err("fio: failed reading job file %s\n", filename);
391                 close(fd);
392                 free(buf);
393                 return 1;
394         }
395
396         ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size);
397         free(buf);
398         close(fd);
399         return ret;
400 }
401
402 int fio_clients_send_ini(const char *filename)
403 {
404         struct fio_client *client;
405         struct flist_head *entry, *tmp;
406
407         flist_for_each_safe(entry, tmp, &client_list) {
408                 client = flist_entry(entry, struct fio_client, list);
409
410                 if (fio_client_send_ini(client, filename))
411                         remove_client(client);
412         }
413
414         return !nr_clients;
415 }
416
417 static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
418 {
419         dst->max_val    = le64_to_cpu(src->max_val);
420         dst->min_val    = le64_to_cpu(src->min_val);
421         dst->samples    = le64_to_cpu(src->samples);
422
423         /*
424          * Floats arrive as IEEE 754 encoded uint64_t, convert back to double
425          */
426         dst->mean.u.f   = fio_uint64_to_double(le64_to_cpu(dst->mean.u.i));
427         dst->S.u.f      = fio_uint64_to_double(le64_to_cpu(dst->S.u.i));
428 }
429
430 static void convert_ts(struct thread_stat *dst, struct thread_stat *src)
431 {
432         int i, j;
433
434         dst->error      = le32_to_cpu(src->error);
435         dst->groupid    = le32_to_cpu(src->groupid);
436         dst->pid        = le32_to_cpu(src->pid);
437         dst->members    = le32_to_cpu(src->members);
438
439         for (i = 0; i < 2; i++) {
440                 convert_io_stat(&dst->clat_stat[i], &src->clat_stat[i]);
441                 convert_io_stat(&dst->slat_stat[i], &src->slat_stat[i]);
442                 convert_io_stat(&dst->lat_stat[i], &src->lat_stat[i]);
443                 convert_io_stat(&dst->bw_stat[i], &src->bw_stat[i]);
444         }
445
446         dst->usr_time           = le64_to_cpu(src->usr_time);
447         dst->sys_time           = le64_to_cpu(src->sys_time);
448         dst->ctx                = le64_to_cpu(src->ctx);
449         dst->minf               = le64_to_cpu(src->minf);
450         dst->majf               = le64_to_cpu(src->majf);
451         dst->clat_percentiles   = le64_to_cpu(src->clat_percentiles);
452
453         for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) {
454                 fio_fp64_t *fps = &src->percentile_list[i];
455                 fio_fp64_t *fpd = &dst->percentile_list[i];
456
457                 fpd->u.f = fio_uint64_to_double(le64_to_cpu(fps->u.i));
458         }
459
460         for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
461                 dst->io_u_map[i]        = le32_to_cpu(src->io_u_map[i]);
462                 dst->io_u_submit[i]     = le32_to_cpu(src->io_u_submit[i]);
463                 dst->io_u_complete[i]   = le32_to_cpu(src->io_u_complete[i]);
464         }
465
466         for (i = 0; i < FIO_IO_U_LAT_U_NR; i++) {
467                 dst->io_u_lat_u[i]      = le32_to_cpu(src->io_u_lat_u[i]);
468                 dst->io_u_lat_m[i]      = le32_to_cpu(src->io_u_lat_m[i]);
469         }
470
471         for (i = 0; i < 2; i++)
472                 for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
473                         dst->io_u_plat[i][j] = le32_to_cpu(src->io_u_plat[i][j]);
474
475         for (i = 0; i < 3; i++) {
476                 dst->total_io_u[i]      = le64_to_cpu(src->total_io_u[i]);
477                 dst->short_io_u[i]      = le64_to_cpu(src->short_io_u[i]);
478         }
479
480         dst->total_submit       = le64_to_cpu(src->total_submit);
481         dst->total_complete     = le64_to_cpu(src->total_complete);
482
483         for (i = 0; i < 2; i++) {
484                 dst->io_bytes[i]        = le64_to_cpu(src->io_bytes[i]);
485                 dst->runtime[i]         = le64_to_cpu(src->runtime[i]);
486         }
487
488         dst->total_run_time     = le64_to_cpu(src->total_run_time);
489         dst->continue_on_error  = le16_to_cpu(src->continue_on_error);
490         dst->total_err_count    = le64_to_cpu(src->total_err_count);
491         dst->first_error        = le32_to_cpu(src->first_error);
492         dst->kb_base            = le32_to_cpu(src->kb_base);
493 }
494
495 static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
496 {
497         int i;
498
499         for (i = 0; i < 2; i++) {
500                 dst->max_run[i]         = le64_to_cpu(src->max_run[i]);
501                 dst->min_run[i]         = le64_to_cpu(src->min_run[i]);
502                 dst->max_bw[i]          = le64_to_cpu(src->max_bw[i]);
503                 dst->min_bw[i]          = le64_to_cpu(src->min_bw[i]);
504                 dst->io_kb[i]           = le64_to_cpu(src->io_kb[i]);
505                 dst->agg[i]             = le64_to_cpu(src->agg[i]);
506         }
507
508         dst->kb_base    = le32_to_cpu(src->kb_base);
509         dst->groupid    = le32_to_cpu(src->groupid);
510 }
511
512 static void handle_ts(struct fio_net_cmd *cmd)
513 {
514         struct cmd_ts_pdu *p = (struct cmd_ts_pdu *) cmd->payload;
515
516         convert_ts(&p->ts, &p->ts);
517         convert_gs(&p->rs, &p->rs);
518
519         show_thread_status(&p->ts, &p->rs);
520 }
521
522 static void handle_gs(struct fio_net_cmd *cmd)
523 {
524         struct group_run_stats *gs = (struct group_run_stats *) cmd->payload;
525
526         convert_gs(gs, gs);
527         show_group_stats(gs);
528 }
529
530 static void convert_jobs_eta(struct jobs_eta *je)
531 {
532         int i;
533
534         je->nr_running          = le32_to_cpu(je->nr_running);
535         je->nr_ramp             = le32_to_cpu(je->nr_ramp);
536         je->nr_pending          = le32_to_cpu(je->nr_pending);
537         je->files_open          = le32_to_cpu(je->files_open);
538         je->m_rate              = le32_to_cpu(je->m_rate);
539         je->t_rate              = le32_to_cpu(je->t_rate);
540         je->m_iops              = le32_to_cpu(je->m_iops);
541         je->t_iops              = le32_to_cpu(je->t_iops);
542
543         for (i = 0; i < 2; i++) {
544                 je->rate[i]     = le32_to_cpu(je->rate[i]);
545                 je->iops[i]     = le32_to_cpu(je->iops[i]);
546         }
547
548         je->elapsed_sec         = le64_to_cpu(je->elapsed_sec);
549         je->eta_sec             = le64_to_cpu(je->eta_sec);
550 }
551
552 static void sum_jobs_eta(struct jobs_eta *je)
553 {
554         struct jobs_eta *dst = &client_etas;
555         int i;
556
557         dst->nr_running         += je->nr_running;
558         dst->nr_ramp            += je->nr_ramp;
559         dst->nr_pending         += je->nr_pending;
560         dst->files_open         += je->files_open;
561         dst->m_rate             += je->m_rate;
562         dst->t_rate             += je->t_rate;
563         dst->m_iops             += je->m_iops;
564         dst->t_iops             += je->t_iops;
565
566         for (i = 0; i < 2; i++) {
567                 dst->rate[i]    += je->rate[i];
568                 dst->iops[i]    += je->iops[i];
569         }
570
571         dst->elapsed_sec        += je->elapsed_sec;
572
573         if (je->eta_sec > dst->eta_sec)
574                 dst->eta_sec = je->eta_sec;
575 }
576
577 static void handle_eta(struct fio_net_cmd *cmd)
578 {
579         struct jobs_eta *je = (struct jobs_eta *) cmd->payload;
580
581         convert_jobs_eta(je);
582
583         if (nr_clients > 1) {
584                 sum_jobs_eta(je);
585                 received_etas++;
586                 if (received_etas == nr_clients) {
587                         received_etas = 0;
588                         display_thread_status(&client_etas);
589                         memset(&client_etas, 0, sizeof(client_etas));
590                 }
591         } else
592                 display_thread_status(je);
593 }
594
595 static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
596 {
597         struct cmd_probe_pdu *probe = (struct cmd_probe_pdu *) cmd->payload;
598         const char *os, *arch;
599
600         os = fio_get_os_string(probe->os);
601         if (!os)
602                 os = "unknown";
603
604         arch = fio_get_arch_string(probe->arch);
605         if (!arch)
606                 os = "unknown";
607
608         log_info("hostname=%s, be=%u, os=%s, arch=%s, fio=%u.%u.%u\n",
609                 probe->hostname, probe->bigendian, os, arch, probe->fio_major,
610                 probe->fio_minor, probe->fio_patch);
611
612         if (!client->name)
613                 client->name = strdup((char *) probe->hostname);
614 }
615
616 static int handle_client(struct fio_client *client)
617 {
618         struct fio_net_cmd *cmd;
619
620         dprint(FD_NET, "client: handle %s\n", client->hostname);
621
622         cmd = fio_net_recv_cmd(client->fd);
623         if (!cmd)
624                 return 0;
625
626         dprint(FD_NET, "client: got cmd op %d from %s\n",
627                                         cmd->opcode, client->hostname);
628
629         switch (cmd->opcode) {
630         case FIO_NET_CMD_QUIT:
631                 remove_client(client);
632                 free(cmd);
633                 break;
634         case FIO_NET_CMD_TEXT: {
635                 const char *buf = (const char *) cmd->payload;
636                 const char *name;
637                 int fio_unused ret;
638
639                 name = client->name ? client->name : client->hostname;
640
641                 if (!client->skip_newline)
642                         fprintf(f_out, "<%s> ", name);
643                 ret = fwrite(buf, cmd->pdu_len, 1, f_out);
644                 fflush(f_out);
645                 client->skip_newline = strchr(buf, '\n') == NULL;
646                 free(cmd);
647                 break;
648                 }
649         case FIO_NET_CMD_TS:
650                 handle_ts(cmd);
651                 free(cmd);
652                 break;
653         case FIO_NET_CMD_GS:
654                 handle_gs(cmd);
655                 free(cmd);
656                 break;
657         case FIO_NET_CMD_ETA:
658                 handle_eta(cmd);
659                 free(cmd);
660                 break;
661         case FIO_NET_CMD_PROBE:
662                 handle_probe(client, cmd);
663                 free(cmd);
664                 break;
665         case FIO_NET_CMD_START:
666                 client->state = Client_started;
667                 free(cmd);
668                 break;
669         case FIO_NET_CMD_STOP:
670                 client->state = Client_stopped;
671                 free(cmd);
672                 break;
673         default:
674                 log_err("fio: unknown client op: %d\n", cmd->opcode);
675                 free(cmd);
676                 break;
677         }
678
679         return 1;
680 }
681
682 int fio_handle_clients(void)
683 {
684         struct fio_client *client;
685         struct flist_head *entry;
686         struct pollfd *pfds;
687         int i, ret = 0;
688
689         pfds = malloc(nr_clients * sizeof(struct pollfd));
690
691         while (!exit_backend && nr_clients) {
692                 i = 0;
693                 flist_for_each(entry, &client_list) {
694                         client = flist_entry(entry, struct fio_client, list);
695
696                         pfds[i].fd = client->fd;
697                         pfds[i].events = POLLIN;
698                         i++;
699                 }
700
701                 assert(i == nr_clients);
702
703                 do {
704                         ret = poll(pfds, nr_clients, 100);
705                         if (ret < 0) {
706                                 if (errno == EINTR)
707                                         continue;
708                                 log_err("fio: poll clients: %s\n", strerror(errno));
709                                 break;
710                         } else if (!ret)
711                                 continue;
712                 } while (ret <= 0);
713
714                 for (i = 0; i < nr_clients; i++) {
715                         if (!(pfds[i].revents & POLLIN))
716                                 continue;
717
718                         client = find_client_by_fd(pfds[i].fd);
719                         if (!client) {
720                                 log_err("fio: unknown client fd %d\n", pfds[i].fd);
721                                 continue;
722                         }
723                         if (!handle_client(client)) {
724                                 log_info("client: host=%s disconnected\n",
725                                                 client->hostname);
726                                 remove_client(client);
727                         }
728                 }
729         }
730
731         free(pfds);
732         return 0;
733 }