11 #include <sys/socket.h>
13 #include <netinet/in.h>
14 #include <arpa/inet.h>
24 struct flist_head list;
25 struct flist_head hash_list;
26 struct sockaddr_in addr;
27 struct sockaddr_un addr_un;
44 static struct timeval eta_tv;
59 static FLIST_HEAD(client_list);
61 #define FIO_CLIENT_HASH_BITS 7
62 #define FIO_CLIENT_HASH_SZ (1 << FIO_CLIENT_HASH_BITS)
63 #define FIO_CLIENT_HASH_MASK (FIO_CLIENT_HASH_SZ - 1)
64 static struct flist_head client_hash[FIO_CLIENT_HASH_SZ];
66 static int handle_client(struct fio_client *client);
68 static void fio_client_add_hash(struct fio_client *client)
70 int bucket = hash_long(client->fd, FIO_CLIENT_HASH_BITS);
72 bucket &= FIO_CLIENT_HASH_MASK;
73 flist_add(&client->hash_list, &client_hash[bucket]);
76 static void fio_client_remove_hash(struct fio_client *client)
78 if (!flist_empty(&client->hash_list))
79 flist_del_init(&client->hash_list);
82 static void fio_init fio_client_hash_init(void)
86 for (i = 0; i < FIO_CLIENT_HASH_SZ; i++)
87 INIT_FLIST_HEAD(&client_hash[i]);
90 static struct fio_client *find_client_by_fd(int fd)
92 int bucket = hash_long(fd, FIO_CLIENT_HASH_BITS) & FIO_CLIENT_HASH_MASK;
93 struct fio_client *client;
94 struct flist_head *entry;
96 flist_for_each(entry, &client_hash[bucket]) {
97 client = flist_entry(entry, struct fio_client, hash_list);
106 static void remove_client(struct fio_client *client)
108 dprint(FD_NET, "client: removed <%s>\n", client->hostname);
109 flist_del(&client->list);
111 fio_client_remove_hash(client);
113 /* FIXME: check ->waiting_eta and handle it */
115 free(client->hostname);
125 static void __fio_client_add_cmd_option(struct fio_client *client,
130 index = client->argc++;
131 client->argv = realloc(client->argv, sizeof(char *) * client->argc);
132 client->argv[index] = strdup(opt);
133 dprint(FD_NET, "client: add cmd %d: %s\n", index, opt);
136 void fio_client_add_cmd_option(void *cookie, const char *opt)
138 struct fio_client *client = cookie;
143 __fio_client_add_cmd_option(client, opt);
146 int fio_client_add(const char *hostname, void **cookie)
148 struct fio_client *client;
150 client = malloc(sizeof(*client));
151 memset(client, 0, sizeof(*client));
153 INIT_FLIST_HEAD(&client->list);
154 INIT_FLIST_HEAD(&client->hash_list);
156 if (fio_server_parse_string(hostname, &client->hostname,
157 &client->is_sock, &client->port,
158 &client->addr.sin_addr))
163 __fio_client_add_cmd_option(client, "fio");
165 flist_add(&client->list, &client_list);
167 dprint(FD_NET, "client: added <%s>\n", client->hostname);
172 static int fio_client_connect_ip(struct fio_client *client)
176 client->addr.sin_family = AF_INET;
177 client->addr.sin_port = htons(client->port);
179 fd = socket(AF_INET, SOCK_STREAM, 0);
181 log_err("fio: socket: %s\n", strerror(errno));
185 if (connect(fd, (struct sockaddr *) &client->addr, sizeof(client->addr)) < 0) {
186 log_err("fio: connect: %s\n", strerror(errno));
187 log_err("fio: failed to connect to %s:%u\n", client->hostname,
196 static int fio_client_connect_sock(struct fio_client *client)
198 struct sockaddr_un *addr = &client->addr_un;
202 memset(addr, 0, sizeof(*addr));
203 addr->sun_family = AF_UNIX;
204 strcpy(addr->sun_path, client->hostname);
206 fd = socket(AF_UNIX, SOCK_STREAM, 0);
208 log_err("fio: socket: %s\n", strerror(errno));
212 len = sizeof(addr->sun_family) + strlen(addr->sun_path) + 1;
213 if (connect(fd, (struct sockaddr *) addr, len) < 0) {
214 log_err("fio: connect; %s\n", strerror(errno));
222 static int fio_client_connect(struct fio_client *client)
226 dprint(FD_NET, "client: connect to host %s\n", client->hostname);
229 fd = fio_client_connect_sock(client);
231 fd = fio_client_connect_ip(client);
237 fio_client_add_hash(client);
238 client->state = Client_connected;
242 void fio_clients_terminate(void)
244 struct flist_head *entry;
245 struct fio_client *client;
247 dprint(FD_NET, "client: terminate clients\n");
249 flist_for_each(entry, &client_list) {
250 client = flist_entry(entry, struct fio_client, list);
252 fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0);
256 static void sig_int(int sig)
258 dprint(FD_NET, "client: got signal %d\n", sig);
259 fio_clients_terminate();
262 static void client_signal_handler(void)
264 struct sigaction act;
266 memset(&act, 0, sizeof(act));
267 act.sa_handler = sig_int;
268 act.sa_flags = SA_RESTART;
269 sigaction(SIGINT, &act, NULL);
271 memset(&act, 0, sizeof(act));
272 act.sa_handler = sig_int;
273 act.sa_flags = SA_RESTART;
274 sigaction(SIGTERM, &act, NULL);
277 static void probe_client(struct fio_client *client)
279 dprint(FD_NET, "client: send probe\n");
281 fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_PROBE, 0);
282 handle_client(client);
285 static int send_client_cmd_line(struct fio_client *client)
287 struct cmd_single_line_pdu *cslp;
288 struct cmd_line_pdu *clp;
289 unsigned long offset;
294 dprint(FD_NET, "client: send cmdline %d\n", client->argc);
297 * Find out how much mem we need
299 for (i = 0, mem = 0; i < client->argc; i++)
300 mem += strlen(client->argv[i]) + 1;
303 * We need one cmd_line_pdu, and argc number of cmd_single_line_pdu
305 mem += sizeof(*clp) + (client->argc * sizeof(*cslp));
309 offset = sizeof(*clp);
311 for (i = 0; i < client->argc; i++) {
312 uint16_t arg_len = strlen(client->argv[i]) + 1;
315 strcpy((char *) cslp->text, client->argv[i]);
316 cslp->len = cpu_to_le16(arg_len);
317 offset += sizeof(*cslp) + arg_len;
320 clp->lines = cpu_to_le16(client->argc);
321 ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOBLINE, pdu, mem, 0);
326 int fio_clients_connect(void)
328 struct fio_client *client;
329 struct flist_head *entry, *tmp;
332 dprint(FD_NET, "client: connect all\n");
334 client_signal_handler();
336 flist_for_each_safe(entry, tmp, &client_list) {
337 client = flist_entry(entry, struct fio_client, list);
339 ret = fio_client_connect(client);
341 remove_client(client);
345 probe_client(client);
347 if (client->argc > 1)
348 send_client_cmd_line(client);
355 * Send file contents to server backend. We could use sendfile(), but to remain
356 * more portable lets just read/write the darn thing.
358 static int fio_client_send_ini(struct fio_client *client, const char *filename)
365 dprint(FD_NET, "send ini %s to %s\n", filename, client->hostname);
367 fd = open(filename, O_RDONLY);
369 log_err("fio: job file <%s> open: %s\n", filename, strerror(errno));
373 if (fstat(fd, &sb) < 0) {
374 log_err("fio: job file stat: %s\n", strerror(errno));
379 buf = malloc(sb.st_size);
384 ret = read(fd, p, len);
393 else if (errno == EAGAIN || errno == EINTR)
398 log_err("fio: failed reading job file %s\n", filename);
404 ret = fio_net_send_cmd(client->fd, FIO_NET_CMD_JOB, buf, sb.st_size, 0);
410 int fio_clients_send_ini(const char *filename)
412 struct fio_client *client;
413 struct flist_head *entry, *tmp;
415 flist_for_each_safe(entry, tmp, &client_list) {
416 client = flist_entry(entry, struct fio_client, list);
418 if (fio_client_send_ini(client, filename))
419 remove_client(client);
425 static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
427 dst->max_val = le64_to_cpu(src->max_val);
428 dst->min_val = le64_to_cpu(src->min_val);
429 dst->samples = le64_to_cpu(src->samples);
432 * Floats arrive as IEEE 754 encoded uint64_t, convert back to double
434 dst->mean.u.f = fio_uint64_to_double(le64_to_cpu(dst->mean.u.i));
435 dst->S.u.f = fio_uint64_to_double(le64_to_cpu(dst->S.u.i));
438 static void convert_ts(struct thread_stat *dst, struct thread_stat *src)
442 dst->error = le32_to_cpu(src->error);
443 dst->groupid = le32_to_cpu(src->groupid);
444 dst->pid = le32_to_cpu(src->pid);
445 dst->members = le32_to_cpu(src->members);
447 for (i = 0; i < 2; i++) {
448 convert_io_stat(&dst->clat_stat[i], &src->clat_stat[i]);
449 convert_io_stat(&dst->slat_stat[i], &src->slat_stat[i]);
450 convert_io_stat(&dst->lat_stat[i], &src->lat_stat[i]);
451 convert_io_stat(&dst->bw_stat[i], &src->bw_stat[i]);
454 dst->usr_time = le64_to_cpu(src->usr_time);
455 dst->sys_time = le64_to_cpu(src->sys_time);
456 dst->ctx = le64_to_cpu(src->ctx);
457 dst->minf = le64_to_cpu(src->minf);
458 dst->majf = le64_to_cpu(src->majf);
459 dst->clat_percentiles = le64_to_cpu(src->clat_percentiles);
461 for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) {
462 fio_fp64_t *fps = &src->percentile_list[i];
463 fio_fp64_t *fpd = &dst->percentile_list[i];
465 fpd->u.f = fio_uint64_to_double(le64_to_cpu(fps->u.i));
468 for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
469 dst->io_u_map[i] = le32_to_cpu(src->io_u_map[i]);
470 dst->io_u_submit[i] = le32_to_cpu(src->io_u_submit[i]);
471 dst->io_u_complete[i] = le32_to_cpu(src->io_u_complete[i]);
474 for (i = 0; i < FIO_IO_U_LAT_U_NR; i++) {
475 dst->io_u_lat_u[i] = le32_to_cpu(src->io_u_lat_u[i]);
476 dst->io_u_lat_m[i] = le32_to_cpu(src->io_u_lat_m[i]);
479 for (i = 0; i < 2; i++)
480 for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
481 dst->io_u_plat[i][j] = le32_to_cpu(src->io_u_plat[i][j]);
483 for (i = 0; i < 3; i++) {
484 dst->total_io_u[i] = le64_to_cpu(src->total_io_u[i]);
485 dst->short_io_u[i] = le64_to_cpu(src->short_io_u[i]);
488 dst->total_submit = le64_to_cpu(src->total_submit);
489 dst->total_complete = le64_to_cpu(src->total_complete);
491 for (i = 0; i < 2; i++) {
492 dst->io_bytes[i] = le64_to_cpu(src->io_bytes[i]);
493 dst->runtime[i] = le64_to_cpu(src->runtime[i]);
496 dst->total_run_time = le64_to_cpu(src->total_run_time);
497 dst->continue_on_error = le16_to_cpu(src->continue_on_error);
498 dst->total_err_count = le64_to_cpu(src->total_err_count);
499 dst->first_error = le32_to_cpu(src->first_error);
500 dst->kb_base = le32_to_cpu(src->kb_base);
503 static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
507 for (i = 0; i < 2; i++) {
508 dst->max_run[i] = le64_to_cpu(src->max_run[i]);
509 dst->min_run[i] = le64_to_cpu(src->min_run[i]);
510 dst->max_bw[i] = le64_to_cpu(src->max_bw[i]);
511 dst->min_bw[i] = le64_to_cpu(src->min_bw[i]);
512 dst->io_kb[i] = le64_to_cpu(src->io_kb[i]);
513 dst->agg[i] = le64_to_cpu(src->agg[i]);
516 dst->kb_base = le32_to_cpu(src->kb_base);
517 dst->groupid = le32_to_cpu(src->groupid);
520 static void handle_ts(struct fio_net_cmd *cmd)
522 struct cmd_ts_pdu *p = (struct cmd_ts_pdu *) cmd->payload;
524 convert_ts(&p->ts, &p->ts);
525 convert_gs(&p->rs, &p->rs);
527 show_thread_status(&p->ts, &p->rs);
530 static void handle_gs(struct fio_net_cmd *cmd)
532 struct group_run_stats *gs = (struct group_run_stats *) cmd->payload;
535 show_group_stats(gs);
538 static void convert_jobs_eta(struct jobs_eta *je)
542 je->nr_running = le32_to_cpu(je->nr_running);
543 je->nr_ramp = le32_to_cpu(je->nr_ramp);
544 je->nr_pending = le32_to_cpu(je->nr_pending);
545 je->files_open = le32_to_cpu(je->files_open);
546 je->m_rate = le32_to_cpu(je->m_rate);
547 je->t_rate = le32_to_cpu(je->t_rate);
548 je->m_iops = le32_to_cpu(je->m_iops);
549 je->t_iops = le32_to_cpu(je->t_iops);
551 for (i = 0; i < 2; i++) {
552 je->rate[i] = le32_to_cpu(je->rate[i]);
553 je->iops[i] = le32_to_cpu(je->iops[i]);
556 je->elapsed_sec = le64_to_cpu(je->elapsed_sec);
557 je->eta_sec = le64_to_cpu(je->eta_sec);
560 static void sum_jobs_eta(struct jobs_eta *dst, struct jobs_eta *je)
564 dst->nr_running += je->nr_running;
565 dst->nr_ramp += je->nr_ramp;
566 dst->nr_pending += je->nr_pending;
567 dst->files_open += je->files_open;
568 dst->m_rate += je->m_rate;
569 dst->t_rate += je->t_rate;
570 dst->m_iops += je->m_iops;
571 dst->t_iops += je->t_iops;
573 for (i = 0; i < 2; i++) {
574 dst->rate[i] += je->rate[i];
575 dst->iops[i] += je->iops[i];
578 dst->elapsed_sec += je->elapsed_sec;
580 if (je->eta_sec > dst->eta_sec)
581 dst->eta_sec = je->eta_sec;
584 static void handle_eta(struct fio_net_cmd *cmd)
586 struct jobs_eta *je = (struct jobs_eta *) cmd->payload;
587 struct client_eta *eta = (struct client_eta *) cmd->tag;
589 dprint(FD_NET, "client: got eta tag %p, %d\n", eta, eta->pending);
591 convert_jobs_eta(je);
592 sum_jobs_eta(&eta->eta, je);
594 if (!--eta->pending) {
595 display_thread_status(&eta->eta);
600 static void handle_probe(struct fio_client *client, struct fio_net_cmd *cmd)
602 struct cmd_probe_pdu *probe = (struct cmd_probe_pdu *) cmd->payload;
603 const char *os, *arch;
605 os = fio_get_os_string(probe->os);
609 arch = fio_get_arch_string(probe->arch);
613 log_info("hostname=%s, be=%u, os=%s, arch=%s, fio=%u.%u.%u\n",
614 probe->hostname, probe->bigendian, os, arch, probe->fio_major,
615 probe->fio_minor, probe->fio_patch);
618 client->name = strdup((char *) probe->hostname);
621 static int handle_client(struct fio_client *client)
623 struct fio_net_cmd *cmd;
625 dprint(FD_NET, "client: handle %s\n", client->hostname);
627 cmd = fio_net_recv_cmd(client->fd);
631 dprint(FD_NET, "client: got cmd op %d from %s\n",
632 cmd->opcode, client->hostname);
634 switch (cmd->opcode) {
635 case FIO_NET_CMD_QUIT:
636 remove_client(client);
639 case FIO_NET_CMD_TEXT: {
640 const char *buf = (const char *) cmd->payload;
644 name = client->name ? client->name : client->hostname;
646 if (!client->skip_newline)
647 fprintf(f_out, "<%s> ", name);
648 ret = fwrite(buf, cmd->pdu_len, 1, f_out);
650 client->skip_newline = strchr(buf, '\n') == NULL;
662 case FIO_NET_CMD_ETA:
666 case FIO_NET_CMD_PROBE:
667 handle_probe(client, cmd);
670 case FIO_NET_CMD_START:
671 client->state = Client_started;
674 case FIO_NET_CMD_STOP:
675 client->state = Client_stopped;
679 log_err("fio: unknown client op: %d\n", cmd->opcode);
687 static void request_client_etas(void)
689 struct fio_client *client;
690 struct flist_head *entry;
691 struct client_eta *eta;
693 dprint(FD_NET, "client: request eta (%d)\n", nr_clients);
696 * We need to do something more clever about checking status
697 * of command being send, client haven't sent previous ETA
701 eta = malloc(sizeof(*eta));
702 memset(&eta->eta, 0, sizeof(eta->eta));
703 eta->pending = nr_clients;
705 flist_for_each(entry, &client_list) {
706 client = flist_entry(entry, struct fio_client, list);
708 client->waiting_eta = 1;
709 fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_SEND_ETA,
713 dprint(FD_NET, "client: requested eta tag %p\n", eta);
716 int fio_handle_clients(void)
718 struct fio_client *client;
719 struct flist_head *entry;
723 gettimeofday(&eta_tv, NULL);
725 pfds = malloc(nr_clients * sizeof(struct pollfd));
727 while (!exit_backend && nr_clients) {
729 flist_for_each(entry, &client_list) {
730 client = flist_entry(entry, struct fio_client, list);
732 pfds[i].fd = client->fd;
733 pfds[i].events = POLLIN;
737 assert(i == nr_clients);
742 gettimeofday(&tv, NULL);
743 if (mtime_since(&eta_tv, &tv) >= 900) {
744 request_client_etas();
745 memcpy(&eta_tv, &tv, sizeof(tv));
748 ret = poll(pfds, nr_clients, 100);
752 log_err("fio: poll clients: %s\n", strerror(errno));
758 for (i = 0; i < nr_clients; i++) {
759 if (!(pfds[i].revents & POLLIN))
762 client = find_client_by_fd(pfds[i].fd);
764 log_err("fio: unknown client fd %d\n", pfds[i].fd);
767 if (!handle_client(client)) {
768 log_info("client: host=%s disconnected\n",
770 remove_client(client);