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