projects / fio.git / blame_incremental
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (non-incremental) | history | HEAD
t/io_uring: link with libaio when necessary
[fio.git] / server.c
... / ...
CommitLineData
1#include <stdio.h>
2#include <stdlib.h>
3#include <unistd.h>
4#include <errno.h>
5#include <poll.h>
6#include <sys/types.h>
7#include <sys/wait.h>
8#include <sys/socket.h>
9#include <sys/stat.h>
10#include <sys/un.h>
11#include <sys/uio.h>
12#include <netinet/in.h>
13#include <arpa/inet.h>
14#include <netdb.h>
15#include <syslog.h>
16#include <signal.h>
17#ifdef CONFIG_ZLIB
18#include <zlib.h>
19#endif
20
21#include "fio.h"
22#include "options.h"
23#include "server.h"
24#include "crc/crc16.h"
25#include "lib/ieee754.h"
26#include "verify-state.h"
27#include "smalloc.h"
28
29int fio_net_port = FIO_NET_PORT;
30
31bool exit_backend = false;
32
33enum {
34 SK_F_FREE = 1,
35 SK_F_COPY = 2,
36 SK_F_SIMPLE = 4,
37 SK_F_VEC = 8,
38 SK_F_INLINE = 16,
39};
40
41struct sk_entry {
42 struct flist_head list; /* link on sk_out->list */
43 int flags; /* SK_F_* */
44 int opcode; /* Actual command fields */
45 void *buf;
46 off_t size;
47 uint64_t tag;
48 struct flist_head next; /* Other sk_entry's, if linked command */
49};
50
51static char *fio_server_arg;
52static char *bind_sock;
53static struct sockaddr_in saddr_in;
54static struct sockaddr_in6 saddr_in6;
55static int use_ipv6;
56#ifdef CONFIG_ZLIB
57static unsigned int has_zlib = 1;
58#else
59static unsigned int has_zlib = 0;
60#endif
61static unsigned int use_zlib;
62static char me[128];
63
64static pthread_key_t sk_out_key;
65
66struct fio_fork_item {
67 struct flist_head list;
68 int exitval;
69 int signal;
70 int exited;
71 pid_t pid;
72};
73
74struct cmd_reply {
75 struct fio_sem lock;
76 void *data;
77 size_t size;
78 int error;
79};
80
81static const char *fio_server_ops[FIO_NET_CMD_NR] = {
82 "",
83 "QUIT",
84 "EXIT",
85 "JOB",
86 "JOBLINE",
87 "TEXT",
88 "TS",
89 "GS",
90 "SEND_ETA",
91 "ETA",
92 "PROBE",
93 "START",
94 "STOP",
95 "DISK_UTIL",
96 "SERVER_START",
97 "ADD_JOB",
98 "RUN",
99 "IOLOG",
100 "UPDATE_JOB",
101 "LOAD_FILE",
102 "VTRIGGER",
103 "SENDFILE",
104 "JOB_OPT",
105};
106
107static void sk_lock(struct sk_out *sk_out)
108{
109 fio_sem_down(&sk_out->lock);
110}
111
112static void sk_unlock(struct sk_out *sk_out)
113{
114 fio_sem_up(&sk_out->lock);
115}
116
117void sk_out_assign(struct sk_out *sk_out)
118{
119 if (!sk_out)
120 return;
121
122 sk_lock(sk_out);
123 sk_out->refs++;
124 sk_unlock(sk_out);
125 pthread_setspecific(sk_out_key, sk_out);
126}
127
128static void sk_out_free(struct sk_out *sk_out)
129{
130 __fio_sem_remove(&sk_out->lock);
131 __fio_sem_remove(&sk_out->wait);
132 __fio_sem_remove(&sk_out->xmit);
133 sfree(sk_out);
134}
135
136static int __sk_out_drop(struct sk_out *sk_out)
137{
138 if (sk_out) {
139 int refs;
140
141 sk_lock(sk_out);
142 assert(sk_out->refs != 0);
143 refs = --sk_out->refs;
144 sk_unlock(sk_out);
145
146 if (!refs) {
147 sk_out_free(sk_out);
148 pthread_setspecific(sk_out_key, NULL);
149 return 0;
150 }
151 }
152
153 return 1;
154}
155
156void sk_out_drop(void)
157{
158 struct sk_out *sk_out;
159
160 sk_out = pthread_getspecific(sk_out_key);
161 __sk_out_drop(sk_out);
162}
163
164static void __fio_init_net_cmd(struct fio_net_cmd *cmd, uint16_t opcode,
165 uint32_t pdu_len, uint64_t tag)
166{
167 memset(cmd, 0, sizeof(*cmd));
168
169 cmd->version = __cpu_to_le16(FIO_SERVER_VER);
170 cmd->opcode = cpu_to_le16(opcode);
171 cmd->tag = cpu_to_le64(tag);
172 cmd->pdu_len = cpu_to_le32(pdu_len);
173}
174
175
176static void fio_init_net_cmd(struct fio_net_cmd *cmd, uint16_t opcode,
177 const void *pdu, uint32_t pdu_len, uint64_t tag)
178{
179 __fio_init_net_cmd(cmd, opcode, pdu_len, tag);
180
181 if (pdu)
182 memcpy(&cmd->payload, pdu, pdu_len);
183}
184
185const char *fio_server_op(unsigned int op)
186{
187 static char buf[32];
188
189 if (op < FIO_NET_CMD_NR)
190 return fio_server_ops[op];
191
192 sprintf(buf, "UNKNOWN/%d", op);
193 return buf;
194}
195
196static ssize_t iov_total_len(const struct iovec *iov, int count)
197{
198 ssize_t ret = 0;
199
200 while (count--) {
201 ret += iov->iov_len;
202 iov++;
203 }
204
205 return ret;
206}
207
208static int fio_sendv_data(int sk, struct iovec *iov, int count)
209{
210 ssize_t total_len = iov_total_len(iov, count);
211 ssize_t ret;
212
213 do {
214 ret = writev(sk, iov, count);
215 if (ret > 0) {
216 total_len -= ret;
217 if (!total_len)
218 break;
219
220 while (ret) {
221 if (ret >= iov->iov_len) {
222 ret -= iov->iov_len;
223 iov++;
224 continue;
225 }
226 iov->iov_base += ret;
227 iov->iov_len -= ret;
228 ret = 0;
229 }
230 } else if (!ret)
231 break;
232 else if (errno == EAGAIN || errno == EINTR)
233 continue;
234 else
235 break;
236 } while (!exit_backend);
237
238 if (!total_len)
239 return 0;
240
241 return 1;
242}
243
244static int fio_send_data(int sk, const void *p, unsigned int len)
245{
246 struct iovec iov = { .iov_base = (void *) p, .iov_len = len };
247
248 assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_FRAGMENT_PDU);
249
250 return fio_sendv_data(sk, &iov, 1);
251}
252
253static int fio_recv_data(int sk, void *buf, unsigned int len, bool wait)
254{
255 int flags;
256 char *p = buf;
257
258 if (wait)
259 flags = MSG_WAITALL;
260 else
261 flags = OS_MSG_DONTWAIT;
262
263 do {
264 int ret = recv(sk, p, len, flags);
265
266 if (ret > 0) {
267 len -= ret;
268 if (!len)
269 break;
270 p += ret;
271 continue;
272 } else if (!ret)
273 break;
274 else if (errno == EAGAIN || errno == EINTR) {
275 if (wait)
276 continue;
277 break;
278 } else
279 break;
280 } while (!exit_backend);
281
282 if (!len)
283 return 0;
284
285 return -1;
286}
287
288static int verify_convert_cmd(struct fio_net_cmd *cmd)
289{
290 uint16_t crc;
291
292 cmd->cmd_crc16 = le16_to_cpu(cmd->cmd_crc16);
293 cmd->pdu_crc16 = le16_to_cpu(cmd->pdu_crc16);
294
295 crc = fio_crc16(cmd, FIO_NET_CMD_CRC_SZ);
296 if (crc != cmd->cmd_crc16) {
297 log_err("fio: server bad crc on command (got %x, wanted %x)\n",
298 cmd->cmd_crc16, crc);
299 fprintf(f_err, "fio: server bad crc on command (got %x, wanted %x)\n",
300 cmd->cmd_crc16, crc);
301 return 1;
302 }
303
304 cmd->version = le16_to_cpu(cmd->version);
305 cmd->opcode = le16_to_cpu(cmd->opcode);
306 cmd->flags = le32_to_cpu(cmd->flags);
307 cmd->tag = le64_to_cpu(cmd->tag);
308 cmd->pdu_len = le32_to_cpu(cmd->pdu_len);
309
310 switch (cmd->version) {
311 case FIO_SERVER_VER:
312 break;
313 default:
314 log_err("fio: bad server cmd version %d\n", cmd->version);
315 fprintf(f_err, "fio: client/server version mismatch (%d != %d)\n",
316 cmd->version, FIO_SERVER_VER);
317 return 1;
318 }
319
320 if (cmd->pdu_len > FIO_SERVER_MAX_FRAGMENT_PDU) {
321 log_err("fio: command payload too large: %u\n", cmd->pdu_len);
322 return 1;
323 }
324
325 return 0;
326}
327
328/*
329 * Read (and defragment, if necessary) incoming commands
330 */
331struct fio_net_cmd *fio_net_recv_cmd(int sk, bool wait)
332{
333 struct fio_net_cmd cmd, *tmp, *cmdret = NULL;
334 size_t cmd_size = 0, pdu_offset = 0;
335 uint16_t crc;
336 int ret, first = 1;
337 void *pdu = NULL;
338
339 do {
340 ret = fio_recv_data(sk, &cmd, sizeof(cmd), wait);
341 if (ret)
342 break;
343
344 /* We have a command, verify it and swap if need be */
345 ret = verify_convert_cmd(&cmd);
346 if (ret)
347 break;
348
349 if (first) {
350 /* if this is text, add room for \0 at the end */
351 cmd_size = sizeof(cmd) + cmd.pdu_len + 1;
352 assert(!cmdret);
353 } else
354 cmd_size += cmd.pdu_len;
355
356 if (cmd_size / 1024 > FIO_SERVER_MAX_CMD_MB * 1024) {
357 log_err("fio: cmd+pdu too large (%llu)\n", (unsigned long long) cmd_size);
358 ret = 1;
359 break;
360 }
361
362 tmp = realloc(cmdret, cmd_size);
363 if (!tmp) {
364 log_err("fio: server failed allocating cmd\n");
365 ret = 1;
366 break;
367 }
368 cmdret = tmp;
369
370 if (first)
371 memcpy(cmdret, &cmd, sizeof(cmd));
372 else if (cmdret->opcode != cmd.opcode) {
373 log_err("fio: fragment opcode mismatch (%d != %d)\n",
374 cmdret->opcode, cmd.opcode);
375 ret = 1;
376 break;
377 }
378
379 if (!cmd.pdu_len)
380 break;
381
382 /* There's payload, get it */
383 pdu = (char *) cmdret->payload + pdu_offset;
384 ret = fio_recv_data(sk, pdu, cmd.pdu_len, wait);
385 if (ret)
386 break;
387
388 /* Verify payload crc */
389 crc = fio_crc16(pdu, cmd.pdu_len);
390 if (crc != cmd.pdu_crc16) {
391 log_err("fio: server bad crc on payload ");
392 log_err("(got %x, wanted %x)\n", cmd.pdu_crc16, crc);
393 ret = 1;
394 break;
395 }
396
397 pdu_offset += cmd.pdu_len;
398 if (!first)
399 cmdret->pdu_len += cmd.pdu_len;
400 first = 0;
401 } while (cmd.flags & FIO_NET_CMD_F_MORE);
402
403 if (ret) {
404 free(cmdret);
405 cmdret = NULL;
406 } else if (cmdret) {
407 /* zero-terminate text input */
408 if (cmdret->pdu_len) {
409 if (cmdret->opcode == FIO_NET_CMD_TEXT) {
410 struct cmd_text_pdu *__pdu = (struct cmd_text_pdu *) cmdret->payload;
411 char *buf = (char *) __pdu->buf;
412 int len = le32_to_cpu(__pdu->buf_len);
413
414 buf[len] = '\0';
415 } else if (cmdret->opcode == FIO_NET_CMD_JOB) {
416 struct cmd_job_pdu *__pdu = (struct cmd_job_pdu *) cmdret->payload;
417 char *buf = (char *) __pdu->buf;
418 int len = le32_to_cpu(__pdu->buf_len);
419
420 buf[len] = '\0';
421 }
422 }
423
424 /* frag flag is internal */
425 cmdret->flags &= ~FIO_NET_CMD_F_MORE;
426 }
427
428 return cmdret;
429}
430
431static void add_reply(uint64_t tag, struct flist_head *list)
432{
433 struct fio_net_cmd_reply *reply;
434
435 reply = (struct fio_net_cmd_reply *) (uintptr_t) tag;
436 flist_add_tail(&reply->list, list);
437}
438
439static uint64_t alloc_reply(uint64_t tag, uint16_t opcode)
440{
441 struct fio_net_cmd_reply *reply;
442
443 reply = calloc(1, sizeof(*reply));
444 INIT_FLIST_HEAD(&reply->list);
445 fio_gettime(&reply->ts, NULL);
446 reply->saved_tag = tag;
447 reply->opcode = opcode;
448
449 return (uintptr_t) reply;
450}
451
452static void free_reply(uint64_t tag)
453{
454 struct fio_net_cmd_reply *reply;
455
456 reply = (struct fio_net_cmd_reply *) (uintptr_t) tag;
457 free(reply);
458}
459
460static void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, const void *pdu)
461{
462 uint32_t pdu_len;
463
464 cmd->cmd_crc16 = __cpu_to_le16(fio_crc16(cmd, FIO_NET_CMD_CRC_SZ));
465
466 pdu_len = le32_to_cpu(cmd->pdu_len);
467 cmd->pdu_crc16 = __cpu_to_le16(fio_crc16(pdu, pdu_len));
468}
469
470static void fio_net_cmd_crc(struct fio_net_cmd *cmd)
471{
472 fio_net_cmd_crc_pdu(cmd, cmd->payload);
473}
474
475int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size,
476 uint64_t *tagptr, struct flist_head *list)
477{
478 struct fio_net_cmd *cmd = NULL;
479 size_t this_len, cur_len = 0;
480 uint64_t tag;
481 int ret;
482
483 if (list) {
484 assert(tagptr);
485 tag = *tagptr = alloc_reply(*tagptr, opcode);
486 } else
487 tag = tagptr ? *tagptr : 0;
488
489 do {
490 this_len = size;
491 if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
492 this_len = FIO_SERVER_MAX_FRAGMENT_PDU;
493
494 if (!cmd || cur_len < sizeof(*cmd) + this_len) {
495 if (cmd)
496 free(cmd);
497
498 cur_len = sizeof(*cmd) + this_len;
499 cmd = malloc(cur_len);
500 }
501
502 fio_init_net_cmd(cmd, opcode, buf, this_len, tag);
503
504 if (this_len < size)
505 cmd->flags = __cpu_to_le32(FIO_NET_CMD_F_MORE);
506
507 fio_net_cmd_crc(cmd);
508
509 ret = fio_send_data(fd, cmd, sizeof(*cmd) + this_len);
510 size -= this_len;
511 buf += this_len;
512 } while (!ret && size);
513
514 if (list) {
515 if (ret)
516 free_reply(tag);
517 else
518 add_reply(tag, list);
519 }
520
521 if (cmd)
522 free(cmd);
523
524 return ret;
525}
526
527static struct sk_entry *fio_net_prep_cmd(uint16_t opcode, void *buf,
528 size_t size, uint64_t *tagptr,
529 int flags)
530{
531 struct sk_entry *entry;
532
533 entry = smalloc(sizeof(*entry));
534 if (!entry)
535 return NULL;
536
537 INIT_FLIST_HEAD(&entry->next);
538 entry->opcode = opcode;
539 if (flags & SK_F_COPY) {
540 entry->buf = smalloc(size);
541 memcpy(entry->buf, buf, size);
542 } else
543 entry->buf = buf;
544
545 entry->size = size;
546 if (tagptr)
547 entry->tag = *tagptr;
548 else
549 entry->tag = 0;
550 entry->flags = flags;
551 return entry;
552}
553
554static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry);
555
556static void fio_net_queue_entry(struct sk_entry *entry)
557{
558 struct sk_out *sk_out = pthread_getspecific(sk_out_key);
559
560 if (entry->flags & SK_F_INLINE)
561 handle_sk_entry(sk_out, entry);
562 else {
563 sk_lock(sk_out);
564 flist_add_tail(&entry->list, &sk_out->list);
565 sk_unlock(sk_out);
566
567 fio_sem_up(&sk_out->wait);
568 }
569}
570
571static int fio_net_queue_cmd(uint16_t opcode, void *buf, off_t size,
572 uint64_t *tagptr, int flags)
573{
574 struct sk_entry *entry;
575
576 entry = fio_net_prep_cmd(opcode, buf, size, tagptr, flags);
577 if (entry) {
578 fio_net_queue_entry(entry);
579 return 0;
580 }
581
582 return 1;
583}
584
585static int fio_net_send_simple_stack_cmd(int sk, uint16_t opcode, uint64_t tag)
586{
587 struct fio_net_cmd cmd;
588
589 fio_init_net_cmd(&cmd, opcode, NULL, 0, tag);
590 fio_net_cmd_crc(&cmd);
591
592 return fio_send_data(sk, &cmd, sizeof(cmd));
593}
594
595/*
596 * If 'list' is non-NULL, then allocate and store the sent command for
597 * later verification.
598 */
599int fio_net_send_simple_cmd(int sk, uint16_t opcode, uint64_t tag,
600 struct flist_head *list)
601{
602 int ret;
603
604 if (list)
605 tag = alloc_reply(tag, opcode);
606
607 ret = fio_net_send_simple_stack_cmd(sk, opcode, tag);
608 if (ret) {
609 if (list)
610 free_reply(tag);
611
612 return ret;
613 }
614
615 if (list)
616 add_reply(tag, list);
617
618 return 0;
619}
620
621static int fio_net_queue_quit(void)
622{
623 dprint(FD_NET, "server: sending quit\n");
624
625 return fio_net_queue_cmd(FIO_NET_CMD_QUIT, NULL, 0, NULL, SK_F_SIMPLE);
626}
627
628int fio_net_send_quit(int sk)
629{
630 dprint(FD_NET, "server: sending quit\n");
631
632 return fio_net_send_simple_cmd(sk, FIO_NET_CMD_QUIT, 0, NULL);
633}
634
635static int fio_net_send_ack(struct fio_net_cmd *cmd, int error, int signal)
636{
637 struct cmd_end_pdu epdu;
638 uint64_t tag = 0;
639
640 if (cmd)
641 tag = cmd->tag;
642
643 epdu.error = __cpu_to_le32(error);
644 epdu.signal = __cpu_to_le32(signal);
645 return fio_net_queue_cmd(FIO_NET_CMD_STOP, &epdu, sizeof(epdu), &tag, SK_F_COPY);
646}
647
648static int fio_net_queue_stop(int error, int signal)
649{
650 dprint(FD_NET, "server: sending stop (%d, %d)\n", error, signal);
651 return fio_net_send_ack(NULL, error, signal);
652}
653
654static void fio_server_add_fork_item(pid_t pid, struct flist_head *list)
655{
656 struct fio_fork_item *ffi;
657
658 ffi = malloc(sizeof(*ffi));
659 ffi->exitval = 0;
660 ffi->signal = 0;
661 ffi->exited = 0;
662 ffi->pid = pid;
663 flist_add_tail(&ffi->list, list);
664}
665
666static void fio_server_add_conn_pid(struct flist_head *conn_list, pid_t pid)
667{
668 dprint(FD_NET, "server: forked off connection job (pid=%u)\n", (int) pid);
669 fio_server_add_fork_item(pid, conn_list);
670}
671
672static void fio_server_add_job_pid(struct flist_head *job_list, pid_t pid)
673{
674 dprint(FD_NET, "server: forked off job job (pid=%u)\n", (int) pid);
675 fio_server_add_fork_item(pid, job_list);
676}
677
678static void fio_server_check_fork_item(struct fio_fork_item *ffi)
679{
680 int ret, status;
681
682 ret = waitpid(ffi->pid, &status, WNOHANG);
683 if (ret < 0) {
684 if (errno == ECHILD) {
685 log_err("fio: connection pid %u disappeared\n", (int) ffi->pid);
686 ffi->exited = 1;
687 } else
688 log_err("fio: waitpid: %s\n", strerror(errno));
689 } else if (ret == ffi->pid) {
690 if (WIFSIGNALED(status)) {
691 ffi->signal = WTERMSIG(status);
692 ffi->exited = 1;
693 }
694 if (WIFEXITED(status)) {
695 if (WEXITSTATUS(status))
696 ffi->exitval = WEXITSTATUS(status);
697 ffi->exited = 1;
698 }
699 }
700}
701
702static void fio_server_fork_item_done(struct fio_fork_item *ffi, bool stop)
703{
704 dprint(FD_NET, "pid %u exited, sig=%u, exitval=%d\n", (int) ffi->pid, ffi->signal, ffi->exitval);
705
706 /*
707 * Fold STOP and QUIT...
708 */
709 if (stop) {
710 fio_net_queue_stop(ffi->exitval, ffi->signal);
711 fio_net_queue_quit();
712 }
713
714 flist_del(&ffi->list);
715 free(ffi);
716}
717
718static void fio_server_check_fork_items(struct flist_head *list, bool stop)
719{
720 struct flist_head *entry, *tmp;
721 struct fio_fork_item *ffi;
722
723 flist_for_each_safe(entry, tmp, list) {
724 ffi = flist_entry(entry, struct fio_fork_item, list);
725
726 fio_server_check_fork_item(ffi);
727
728 if (ffi->exited)
729 fio_server_fork_item_done(ffi, stop);
730 }
731}
732
733static void fio_server_check_jobs(struct flist_head *job_list)
734{
735 fio_server_check_fork_items(job_list, true);
736}
737
738static void fio_server_check_conns(struct flist_head *conn_list)
739{
740 fio_server_check_fork_items(conn_list, false);
741}
742
743static int handle_load_file_cmd(struct fio_net_cmd *cmd)
744{
745 struct cmd_load_file_pdu *pdu = (struct cmd_load_file_pdu *) cmd->payload;
746 void *file_name = pdu->file;
747 struct cmd_start_pdu spdu;
748
749 dprint(FD_NET, "server: loading local file %s\n", (char *) file_name);
750
751 pdu->name_len = le16_to_cpu(pdu->name_len);
752 pdu->client_type = le16_to_cpu(pdu->client_type);
753
754 if (parse_jobs_ini(file_name, 0, 0, pdu->client_type)) {
755 fio_net_queue_quit();
756 return -1;
757 }
758
759 spdu.jobs = cpu_to_le32(thread_number);
760 spdu.stat_outputs = cpu_to_le32(stat_number);
761 fio_net_queue_cmd(FIO_NET_CMD_START, &spdu, sizeof(spdu), NULL, SK_F_COPY);
762 return 0;
763}
764
765static int handle_run_cmd(struct sk_out *sk_out, struct flist_head *job_list,
766 struct fio_net_cmd *cmd)
767{
768 pid_t pid;
769 int ret;
770
771 sk_out_assign(sk_out);
772
773 fio_time_init();
774 set_genesis_time();
775
776 pid = fork();
777 if (pid) {
778 fio_server_add_job_pid(job_list, pid);
779 return 0;
780 }
781
782 ret = fio_backend(sk_out);
783 free_threads_shm();
784 sk_out_drop();
785 _exit(ret);
786}
787
788static int handle_job_cmd(struct fio_net_cmd *cmd)
789{
790 struct cmd_job_pdu *pdu = (struct cmd_job_pdu *) cmd->payload;
791 void *buf = pdu->buf;
792 struct cmd_start_pdu spdu;
793
794 pdu->buf_len = le32_to_cpu(pdu->buf_len);
795 pdu->client_type = le32_to_cpu(pdu->client_type);
796
797 if (parse_jobs_ini(buf, 1, 0, pdu->client_type)) {
798 fio_net_queue_quit();
799 return -1;
800 }
801
802 spdu.jobs = cpu_to_le32(thread_number);
803 spdu.stat_outputs = cpu_to_le32(stat_number);
804
805 fio_net_queue_cmd(FIO_NET_CMD_START, &spdu, sizeof(spdu), NULL, SK_F_COPY);
806 return 0;
807}
808
809static int handle_jobline_cmd(struct fio_net_cmd *cmd)
810{
811 void *pdu = cmd->payload;
812 struct cmd_single_line_pdu *cslp;
813 struct cmd_line_pdu *clp;
814 unsigned long offset;
815 struct cmd_start_pdu spdu;
816 char **argv;
817 int i;
818
819 clp = pdu;
820 clp->lines = le16_to_cpu(clp->lines);
821 clp->client_type = le16_to_cpu(clp->client_type);
822 argv = malloc(clp->lines * sizeof(char *));
823 offset = sizeof(*clp);
824
825 dprint(FD_NET, "server: %d command line args\n", clp->lines);
826
827 for (i = 0; i < clp->lines; i++) {
828 cslp = pdu + offset;
829 argv[i] = (char *) cslp->text;
830
831 offset += sizeof(*cslp) + le16_to_cpu(cslp->len);
832 dprint(FD_NET, "server: %d: %s\n", i, argv[i]);
833 }
834
835 if (parse_cmd_line(clp->lines, argv, clp->client_type)) {
836 fio_net_queue_quit();
837 free(argv);
838 return -1;
839 }
840
841 free(argv);
842
843 spdu.jobs = cpu_to_le32(thread_number);
844 spdu.stat_outputs = cpu_to_le32(stat_number);
845
846 fio_net_queue_cmd(FIO_NET_CMD_START, &spdu, sizeof(spdu), NULL, SK_F_COPY);
847 return 0;
848}
849
850static int handle_probe_cmd(struct fio_net_cmd *cmd)
851{
852 struct cmd_client_probe_pdu *pdu = (struct cmd_client_probe_pdu *) cmd->payload;
853 uint64_t tag = cmd->tag;
854 struct cmd_probe_reply_pdu probe = {
855#ifdef CONFIG_BIG_ENDIAN
856 .bigendian = 1,
857#endif
858 .os = FIO_OS,
859 .arch = FIO_ARCH,
860 .bpp = sizeof(void *),
861 .cpus = __cpu_to_le32(cpus_online()),
862 };
863
864 dprint(FD_NET, "server: sending probe reply\n");
865
866 strcpy(me, (char *) pdu->server);
867
868 gethostname((char *) probe.hostname, sizeof(probe.hostname));
869 snprintf((char *) probe.fio_version, sizeof(probe.fio_version), "%s",
870 fio_version_string);
871
872 /*
873 * If the client supports compression and we do too, then enable it
874 */
875 if (has_zlib && le64_to_cpu(pdu->flags) & FIO_PROBE_FLAG_ZLIB) {
876 probe.flags = __cpu_to_le64(FIO_PROBE_FLAG_ZLIB);
877 use_zlib = 1;
878 } else {
879 probe.flags = 0;
880 use_zlib = 0;
881 }
882
883 return fio_net_queue_cmd(FIO_NET_CMD_PROBE, &probe, sizeof(probe), &tag, SK_F_COPY);
884}
885
886static int handle_send_eta_cmd(struct fio_net_cmd *cmd)
887{
888 struct jobs_eta *je;
889 uint64_t tag = cmd->tag;
890 size_t size;
891 int i;
892
893 dprint(FD_NET, "server sending status\n");
894
895 /*
896 * Fake ETA return if we don't have a local one, otherwise the client
897 * will end up timing out waiting for a response to the ETA request
898 */
899 je = get_jobs_eta(true, &size);
900 if (!je) {
901 size = sizeof(*je);
902 je = calloc(1, size);
903 } else {
904 je->nr_running = cpu_to_le32(je->nr_running);
905 je->nr_ramp = cpu_to_le32(je->nr_ramp);
906 je->nr_pending = cpu_to_le32(je->nr_pending);
907 je->nr_setting_up = cpu_to_le32(je->nr_setting_up);
908 je->files_open = cpu_to_le32(je->files_open);
909
910 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
911 je->m_rate[i] = cpu_to_le64(je->m_rate[i]);
912 je->t_rate[i] = cpu_to_le64(je->t_rate[i]);
913 je->m_iops[i] = cpu_to_le32(je->m_iops[i]);
914 je->t_iops[i] = cpu_to_le32(je->t_iops[i]);
915 je->rate[i] = cpu_to_le64(je->rate[i]);
916 je->iops[i] = cpu_to_le32(je->iops[i]);
917 }
918
919 je->elapsed_sec = cpu_to_le64(je->elapsed_sec);
920 je->eta_sec = cpu_to_le64(je->eta_sec);
921 je->nr_threads = cpu_to_le32(je->nr_threads);
922 je->is_pow2 = cpu_to_le32(je->is_pow2);
923 je->unit_base = cpu_to_le32(je->unit_base);
924 }
925
926 fio_net_queue_cmd(FIO_NET_CMD_ETA, je, size, &tag, SK_F_FREE);
927 return 0;
928}
929
930static int send_update_job_reply(uint64_t __tag, int error)
931{
932 uint64_t tag = __tag;
933 uint32_t pdu_error;
934
935 pdu_error = __cpu_to_le32(error);
936 return fio_net_queue_cmd(FIO_NET_CMD_UPDATE_JOB, &pdu_error, sizeof(pdu_error), &tag, SK_F_COPY);
937}
938
939static int handle_update_job_cmd(struct fio_net_cmd *cmd)
940{
941 struct cmd_add_job_pdu *pdu = (struct cmd_add_job_pdu *) cmd->payload;
942 struct thread_data *td;
943 uint32_t tnumber;
944
945 tnumber = le32_to_cpu(pdu->thread_number);
946
947 dprint(FD_NET, "server: updating options for job %u\n", tnumber);
948
949 if (!tnumber || tnumber > thread_number) {
950 send_update_job_reply(cmd->tag, ENODEV);
951 return 0;
952 }
953
954 td = tnumber_to_td(tnumber);
955 convert_thread_options_to_cpu(&td->o, &pdu->top);
956 send_update_job_reply(cmd->tag, 0);
957 return 0;
958}
959
960static int handle_trigger_cmd(struct fio_net_cmd *cmd, struct flist_head *job_list)
961{
962 struct cmd_vtrigger_pdu *pdu = (struct cmd_vtrigger_pdu *) cmd->payload;
963 char *buf = (char *) pdu->cmd;
964 struct all_io_list *rep;
965 size_t sz;
966
967 pdu->len = le16_to_cpu(pdu->len);
968 buf[pdu->len] = '\0';
969
970 rep = get_all_io_list(IO_LIST_ALL, &sz);
971 if (!rep) {
972 struct all_io_list state;
973
974 state.threads = cpu_to_le64((uint64_t) 0);
975 fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, &state, sizeof(state), NULL, SK_F_COPY | SK_F_INLINE);
976 } else
977 fio_net_queue_cmd(FIO_NET_CMD_VTRIGGER, rep, sz, NULL, SK_F_FREE | SK_F_INLINE);
978
979 fio_terminate_threads(TERMINATE_ALL, TERMINATE_ALL);
980 fio_server_check_jobs(job_list);
981 exec_trigger(buf);
982 return 0;
983}
984
985static int handle_command(struct sk_out *sk_out, struct flist_head *job_list,
986 struct fio_net_cmd *cmd)
987{
988 int ret;
989
990 dprint(FD_NET, "server: got op [%s], pdu=%u, tag=%llx\n",
991 fio_server_op(cmd->opcode), cmd->pdu_len,
992 (unsigned long long) cmd->tag);
993
994 switch (cmd->opcode) {
995 case FIO_NET_CMD_QUIT:
996 fio_terminate_threads(TERMINATE_ALL, TERMINATE_ALL);
997 ret = 0;
998 break;
999 case FIO_NET_CMD_EXIT:
1000 exit_backend = true;
1001 return -1;
1002 case FIO_NET_CMD_LOAD_FILE:
1003 ret = handle_load_file_cmd(cmd);
1004 break;
1005 case FIO_NET_CMD_JOB:
1006 ret = handle_job_cmd(cmd);
1007 break;
1008 case FIO_NET_CMD_JOBLINE:
1009 ret = handle_jobline_cmd(cmd);
1010 break;
1011 case FIO_NET_CMD_PROBE:
1012 ret = handle_probe_cmd(cmd);
1013 break;
1014 case FIO_NET_CMD_SEND_ETA:
1015 ret = handle_send_eta_cmd(cmd);
1016 break;
1017 case FIO_NET_CMD_RUN:
1018 ret = handle_run_cmd(sk_out, job_list, cmd);
1019 break;
1020 case FIO_NET_CMD_UPDATE_JOB:
1021 ret = handle_update_job_cmd(cmd);
1022 break;
1023 case FIO_NET_CMD_VTRIGGER:
1024 ret = handle_trigger_cmd(cmd, job_list);
1025 break;
1026 case FIO_NET_CMD_SENDFILE: {
1027 struct cmd_sendfile_reply *in;
1028 struct cmd_reply *rep;
1029
1030 rep = (struct cmd_reply *) (uintptr_t) cmd->tag;
1031
1032 in = (struct cmd_sendfile_reply *) cmd->payload;
1033 in->size = le32_to_cpu(in->size);
1034 in->error = le32_to_cpu(in->error);
1035 if (in->error) {
1036 ret = 1;
1037 rep->error = in->error;
1038 } else {
1039 ret = 0;
1040 rep->data = smalloc(in->size);
1041 if (!rep->data) {
1042 ret = 1;
1043 rep->error = ENOMEM;
1044 } else {
1045 rep->size = in->size;
1046 memcpy(rep->data, in->data, in->size);
1047 }
1048 }
1049 fio_sem_up(&rep->lock);
1050 break;
1051 }
1052 default:
1053 log_err("fio: unknown opcode: %s\n", fio_server_op(cmd->opcode));
1054 ret = 1;
1055 }
1056
1057 return ret;
1058}
1059
1060/*
1061 * Send a command with a separate PDU, not inlined in the command
1062 */
1063static int fio_send_cmd_ext_pdu(int sk, uint16_t opcode, const void *buf,
1064 off_t size, uint64_t tag, uint32_t flags)
1065{
1066 struct fio_net_cmd cmd;
1067 struct iovec iov[2];
1068 size_t this_len;
1069 int ret;
1070
1071 iov[0].iov_base = (void *) &cmd;
1072 iov[0].iov_len = sizeof(cmd);
1073
1074 do {
1075 uint32_t this_flags = flags;
1076
1077 this_len = size;
1078 if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
1079 this_len = FIO_SERVER_MAX_FRAGMENT_PDU;
1080
1081 if (this_len < size)
1082 this_flags |= FIO_NET_CMD_F_MORE;
1083
1084 __fio_init_net_cmd(&cmd, opcode, this_len, tag);
1085 cmd.flags = __cpu_to_le32(this_flags);
1086 fio_net_cmd_crc_pdu(&cmd, buf);
1087
1088 iov[1].iov_base = (void *) buf;
1089 iov[1].iov_len = this_len;
1090
1091 ret = fio_sendv_data(sk, iov, 2);
1092 size -= this_len;
1093 buf += this_len;
1094 } while (!ret && size);
1095
1096 return ret;
1097}
1098
1099static void finish_entry(struct sk_entry *entry)
1100{
1101 if (entry->flags & SK_F_FREE)
1102 free(entry->buf);
1103 else if (entry->flags & SK_F_COPY)
1104 sfree(entry->buf);
1105
1106 sfree(entry);
1107}
1108
1109static void entry_set_flags(struct sk_entry *entry, struct flist_head *list,
1110 unsigned int *flags)
1111{
1112 if (!flist_empty(list))
1113 *flags = FIO_NET_CMD_F_MORE;
1114 else
1115 *flags = 0;
1116}
1117
1118static int send_vec_entry(struct sk_out *sk_out, struct sk_entry *first)
1119{
1120 unsigned int flags;
1121 int ret;
1122
1123 entry_set_flags(first, &first->next, &flags);
1124
1125 ret = fio_send_cmd_ext_pdu(sk_out->sk, first->opcode, first->buf,
1126 first->size, first->tag, flags);
1127
1128 while (!flist_empty(&first->next)) {
1129 struct sk_entry *next;
1130
1131 next = flist_first_entry(&first->next, struct sk_entry, list);
1132 flist_del_init(&next->list);
1133
1134 entry_set_flags(next, &first->next, &flags);
1135
1136 ret += fio_send_cmd_ext_pdu(sk_out->sk, next->opcode, next->buf,
1137 next->size, next->tag, flags);
1138 finish_entry(next);
1139 }
1140
1141 return ret;
1142}
1143
1144static int handle_sk_entry(struct sk_out *sk_out, struct sk_entry *entry)
1145{
1146 int ret;
1147
1148 fio_sem_down(&sk_out->xmit);
1149
1150 if (entry->flags & SK_F_VEC)
1151 ret = send_vec_entry(sk_out, entry);
1152 else if (entry->flags & SK_F_SIMPLE) {
1153 ret = fio_net_send_simple_cmd(sk_out->sk, entry->opcode,
1154 entry->tag, NULL);
1155 } else {
1156 ret = fio_net_send_cmd(sk_out->sk, entry->opcode, entry->buf,
1157 entry->size, &entry->tag, NULL);
1158 }
1159
1160 fio_sem_up(&sk_out->xmit);
1161
1162 if (ret)
1163 log_err("fio: failed handling cmd %s\n", fio_server_op(entry->opcode));
1164
1165 finish_entry(entry);
1166 return ret;
1167}
1168
1169static int handle_xmits(struct sk_out *sk_out)
1170{
1171 struct sk_entry *entry;
1172 FLIST_HEAD(list);
1173 int ret = 0;
1174
1175 sk_lock(sk_out);
1176 if (flist_empty(&sk_out->list)) {
1177 sk_unlock(sk_out);
1178 return 0;
1179 }
1180
1181 flist_splice_init(&sk_out->list, &list);
1182 sk_unlock(sk_out);
1183
1184 while (!flist_empty(&list)) {
1185 entry = flist_entry(list.next, struct sk_entry, list);
1186 flist_del(&entry->list);
1187 ret += handle_sk_entry(sk_out, entry);
1188 }
1189
1190 return ret;
1191}
1192
1193static int handle_connection(struct sk_out *sk_out)
1194{
1195 struct fio_net_cmd *cmd = NULL;
1196 FLIST_HEAD(job_list);
1197 int ret = 0;
1198
1199 reset_fio_state();
1200
1201 /* read forever */
1202 while (!exit_backend) {
1203 struct pollfd pfd = {
1204 .fd = sk_out->sk,
1205 .events = POLLIN,
1206 };
1207
1208 do {
1209 int timeout = 1000;
1210
1211 if (!flist_empty(&job_list))
1212 timeout = 100;
1213
1214 handle_xmits(sk_out);
1215
1216 ret = poll(&pfd, 1, 0);
1217 if (ret < 0) {
1218 if (errno == EINTR)
1219 break;
1220 log_err("fio: poll: %s\n", strerror(errno));
1221 break;
1222 } else if (!ret) {
1223 fio_server_check_jobs(&job_list);
1224 fio_sem_down_timeout(&sk_out->wait, timeout);
1225 continue;
1226 }
1227
1228 if (pfd.revents & POLLIN)
1229 break;
1230 if (pfd.revents & (POLLERR|POLLHUP)) {
1231 ret = 1;
1232 break;
1233 }
1234 } while (!exit_backend);
1235
1236 fio_server_check_jobs(&job_list);
1237
1238 if (ret < 0)
1239 break;
1240
1241 cmd = fio_net_recv_cmd(sk_out->sk, true);
1242 if (!cmd) {
1243 ret = -1;
1244 break;
1245 }
1246
1247 ret = handle_command(sk_out, &job_list, cmd);
1248 if (ret)
1249 break;
1250
1251 free(cmd);
1252 cmd = NULL;
1253 }
1254
1255 if (cmd)
1256 free(cmd);
1257
1258 handle_xmits(sk_out);
1259
1260 close(sk_out->sk);
1261 sk_out->sk = -1;
1262 __sk_out_drop(sk_out);
1263 _exit(ret);
1264}
1265
1266/* get the address on this host bound by the input socket,
1267 * whether it is ipv6 or ipv4 */
1268
1269static int get_my_addr_str(int sk)
1270{
1271 struct sockaddr_in6 myaddr6 = { 0, };
1272 struct sockaddr_in myaddr4 = { 0, };
1273 struct sockaddr *sockaddr_p;
1274 char *net_addr;
1275 socklen_t len;
1276 int ret;
1277
1278 if (use_ipv6) {
1279 len = sizeof(myaddr6);
1280 sockaddr_p = (struct sockaddr * )&myaddr6;
1281 net_addr = (char * )&myaddr6.sin6_addr;
1282 } else {
1283 len = sizeof(myaddr4);
1284 sockaddr_p = (struct sockaddr * )&myaddr4;
1285 net_addr = (char * )&myaddr4.sin_addr;
1286 }
1287
1288 ret = getsockname(sk, sockaddr_p, &len);
1289 if (ret) {
1290 log_err("fio: getsockname: %s\n", strerror(errno));
1291 return -1;
1292 }
1293
1294 if (!inet_ntop(use_ipv6?AF_INET6:AF_INET, net_addr, client_sockaddr_str, INET6_ADDRSTRLEN - 1)) {
1295 log_err("inet_ntop: failed to convert addr to string\n");
1296 return -1;
1297 }
1298
1299 dprint(FD_NET, "fio server bound to addr %s\n", client_sockaddr_str);
1300 return 0;
1301}
1302
1303static int accept_loop(int listen_sk)
1304{
1305 struct sockaddr_in addr;
1306 struct sockaddr_in6 addr6;
1307 socklen_t len = use_ipv6 ? sizeof(addr6) : sizeof(addr);
1308 struct pollfd pfd;
1309 int ret = 0, sk, exitval = 0;
1310 FLIST_HEAD(conn_list);
1311
1312 dprint(FD_NET, "server enter accept loop\n");
1313
1314 fio_set_fd_nonblocking(listen_sk, "server");
1315
1316 while (!exit_backend) {
1317 struct sk_out *sk_out;
1318 const char *from;
1319 char buf[64];
1320 pid_t pid;
1321
1322 pfd.fd = listen_sk;
1323 pfd.events = POLLIN;
1324 do {
1325 int timeout = 1000;
1326
1327 if (!flist_empty(&conn_list))
1328 timeout = 100;
1329
1330 ret = poll(&pfd, 1, timeout);
1331 if (ret < 0) {
1332 if (errno == EINTR)
1333 break;
1334 log_err("fio: poll: %s\n", strerror(errno));
1335 break;
1336 } else if (!ret) {
1337 fio_server_check_conns(&conn_list);
1338 continue;
1339 }
1340
1341 if (pfd.revents & POLLIN)
1342 break;
1343 } while (!exit_backend);
1344
1345 fio_server_check_conns(&conn_list);
1346
1347 if (exit_backend || ret < 0)
1348 break;
1349
1350 if (use_ipv6)
1351 sk = accept(listen_sk, (struct sockaddr *) &addr6, &len);
1352 else
1353 sk = accept(listen_sk, (struct sockaddr *) &addr, &len);
1354
1355 if (sk < 0) {
1356 log_err("fio: accept: %s\n", strerror(errno));
1357 return -1;
1358 }
1359
1360 if (use_ipv6)
1361 from = inet_ntop(AF_INET6, (struct sockaddr *) &addr6.sin6_addr, buf, sizeof(buf));
1362 else
1363 from = inet_ntop(AF_INET, (struct sockaddr *) &addr.sin_addr, buf, sizeof(buf));
1364
1365 dprint(FD_NET, "server: connect from %s\n", from);
1366
1367 sk_out = scalloc(1, sizeof(*sk_out));
1368 if (!sk_out) {
1369 close(sk);
1370 return -1;
1371 }
1372
1373 sk_out->sk = sk;
1374 INIT_FLIST_HEAD(&sk_out->list);
1375 __fio_sem_init(&sk_out->lock, FIO_SEM_UNLOCKED);
1376 __fio_sem_init(&sk_out->wait, FIO_SEM_LOCKED);
1377 __fio_sem_init(&sk_out->xmit, FIO_SEM_UNLOCKED);
1378
1379 pid = fork();
1380 if (pid) {
1381 close(sk);
1382 fio_server_add_conn_pid(&conn_list, pid);
1383 continue;
1384 }
1385
1386 /* if error, it's already logged, non-fatal */
1387 get_my_addr_str(sk);
1388
1389 /*
1390 * Assign sk_out here, it'll be dropped in handle_connection()
1391 * since that function calls _exit() when done
1392 */
1393 sk_out_assign(sk_out);
1394 handle_connection(sk_out);
1395 }
1396
1397 return exitval;
1398}
1399
1400int fio_server_text_output(int level, const char *buf, size_t len)
1401{
1402 struct sk_out *sk_out = pthread_getspecific(sk_out_key);
1403 struct cmd_text_pdu *pdu;
1404 unsigned int tlen;
1405 struct timeval tv;
1406
1407 if (!sk_out || sk_out->sk == -1)
1408 return -1;
1409
1410 tlen = sizeof(*pdu) + len;
1411 pdu = malloc(tlen);
1412
1413 pdu->level = __cpu_to_le32(level);
1414 pdu->buf_len = __cpu_to_le32(len);
1415
1416 gettimeofday(&tv, NULL);
1417 pdu->log_sec = __cpu_to_le64(tv.tv_sec);
1418 pdu->log_usec = __cpu_to_le64(tv.tv_usec);
1419
1420 memcpy(pdu->buf, buf, len);
1421
1422 fio_net_queue_cmd(FIO_NET_CMD_TEXT, pdu, tlen, NULL, SK_F_COPY);
1423 free(pdu);
1424 return len;
1425}
1426
1427static void convert_io_stat(struct io_stat *dst, struct io_stat *src)
1428{
1429 dst->max_val = cpu_to_le64(src->max_val);
1430 dst->min_val = cpu_to_le64(src->min_val);
1431 dst->samples = cpu_to_le64(src->samples);
1432
1433 /*
1434 * Encode to IEEE 754 for network transfer
1435 */
1436 dst->mean.u.i = cpu_to_le64(fio_double_to_uint64(src->mean.u.f));
1437 dst->S.u.i = cpu_to_le64(fio_double_to_uint64(src->S.u.f));
1438}
1439
1440static void convert_gs(struct group_run_stats *dst, struct group_run_stats *src)
1441{
1442 int i;
1443
1444 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
1445 dst->max_run[i] = cpu_to_le64(src->max_run[i]);
1446 dst->min_run[i] = cpu_to_le64(src->min_run[i]);
1447 dst->max_bw[i] = cpu_to_le64(src->max_bw[i]);
1448 dst->min_bw[i] = cpu_to_le64(src->min_bw[i]);
1449 dst->iobytes[i] = cpu_to_le64(src->iobytes[i]);
1450 dst->agg[i] = cpu_to_le64(src->agg[i]);
1451 }
1452
1453 dst->kb_base = cpu_to_le32(src->kb_base);
1454 dst->unit_base = cpu_to_le32(src->unit_base);
1455 dst->groupid = cpu_to_le32(src->groupid);
1456 dst->unified_rw_rep = cpu_to_le32(src->unified_rw_rep);
1457 dst->sig_figs = cpu_to_le32(src->sig_figs);
1458}
1459
1460/*
1461 * Send a CMD_TS, which packs struct thread_stat and group_run_stats
1462 * into a single payload.
1463 */
1464void fio_server_send_ts(struct thread_stat *ts, struct group_run_stats *rs)
1465{
1466 struct cmd_ts_pdu p;
1467 int i, j, k;
1468 void *ss_buf;
1469 uint64_t *ss_iops, *ss_bw;
1470
1471 dprint(FD_NET, "server sending end stats\n");
1472
1473 memset(&p, 0, sizeof(p));
1474
1475 snprintf(p.ts.name, sizeof(p.ts.name), "%s", ts->name);
1476 snprintf(p.ts.verror, sizeof(p.ts.verror), "%s", ts->verror);
1477 snprintf(p.ts.description, sizeof(p.ts.description), "%s",
1478 ts->description);
1479
1480 p.ts.error = cpu_to_le32(ts->error);
1481 p.ts.thread_number = cpu_to_le32(ts->thread_number);
1482 p.ts.groupid = cpu_to_le32(ts->groupid);
1483 p.ts.pid = cpu_to_le32(ts->pid);
1484 p.ts.members = cpu_to_le32(ts->members);
1485 p.ts.unified_rw_rep = cpu_to_le32(ts->unified_rw_rep);
1486
1487 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
1488 convert_io_stat(&p.ts.clat_stat[i], &ts->clat_stat[i]);
1489 convert_io_stat(&p.ts.slat_stat[i], &ts->slat_stat[i]);
1490 convert_io_stat(&p.ts.lat_stat[i], &ts->lat_stat[i]);
1491 convert_io_stat(&p.ts.bw_stat[i], &ts->bw_stat[i]);
1492 convert_io_stat(&p.ts.iops_stat[i], &ts->iops_stat[i]);
1493 }
1494 convert_io_stat(&p.ts.sync_stat, &ts->sync_stat);
1495
1496 p.ts.usr_time = cpu_to_le64(ts->usr_time);
1497 p.ts.sys_time = cpu_to_le64(ts->sys_time);
1498 p.ts.ctx = cpu_to_le64(ts->ctx);
1499 p.ts.minf = cpu_to_le64(ts->minf);
1500 p.ts.majf = cpu_to_le64(ts->majf);
1501 p.ts.clat_percentiles = cpu_to_le32(ts->clat_percentiles);
1502 p.ts.lat_percentiles = cpu_to_le32(ts->lat_percentiles);
1503 p.ts.slat_percentiles = cpu_to_le32(ts->slat_percentiles);
1504 p.ts.percentile_precision = cpu_to_le64(ts->percentile_precision);
1505
1506 for (i = 0; i < FIO_IO_U_LIST_MAX_LEN; i++) {
1507 fio_fp64_t *src = &ts->percentile_list[i];
1508 fio_fp64_t *dst = &p.ts.percentile_list[i];
1509
1510 dst->u.i = cpu_to_le64(fio_double_to_uint64(src->u.f));
1511 }
1512
1513 for (i = 0; i < FIO_IO_U_MAP_NR; i++) {
1514 p.ts.io_u_map[i] = cpu_to_le64(ts->io_u_map[i]);
1515 p.ts.io_u_submit[i] = cpu_to_le64(ts->io_u_submit[i]);
1516 p.ts.io_u_complete[i] = cpu_to_le64(ts->io_u_complete[i]);
1517 }
1518
1519 for (i = 0; i < FIO_IO_U_LAT_N_NR; i++)
1520 p.ts.io_u_lat_n[i] = cpu_to_le64(ts->io_u_lat_n[i]);
1521 for (i = 0; i < FIO_IO_U_LAT_U_NR; i++)
1522 p.ts.io_u_lat_u[i] = cpu_to_le64(ts->io_u_lat_u[i]);
1523 for (i = 0; i < FIO_IO_U_LAT_M_NR; i++)
1524 p.ts.io_u_lat_m[i] = cpu_to_le64(ts->io_u_lat_m[i]);
1525
1526 for (i = 0; i < FIO_LAT_CNT; i++)
1527 for (j = 0; j < DDIR_RWDIR_CNT; j++)
1528 for (k = 0; k < FIO_IO_U_PLAT_NR; k++)
1529 p.ts.io_u_plat[i][j][k] = cpu_to_le64(ts->io_u_plat[i][j][k]);
1530
1531 for (j = 0; j < FIO_IO_U_PLAT_NR; j++)
1532 p.ts.io_u_sync_plat[j] = cpu_to_le64(ts->io_u_sync_plat[j]);
1533
1534 for (i = 0; i < DDIR_RWDIR_SYNC_CNT; i++)
1535 p.ts.total_io_u[i] = cpu_to_le64(ts->total_io_u[i]);
1536
1537 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
1538 p.ts.short_io_u[i] = cpu_to_le64(ts->short_io_u[i]);
1539 p.ts.drop_io_u[i] = cpu_to_le64(ts->drop_io_u[i]);
1540 }
1541
1542 p.ts.total_submit = cpu_to_le64(ts->total_submit);
1543 p.ts.total_complete = cpu_to_le64(ts->total_complete);
1544 p.ts.nr_zone_resets = cpu_to_le64(ts->nr_zone_resets);
1545
1546 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
1547 p.ts.io_bytes[i] = cpu_to_le64(ts->io_bytes[i]);
1548 p.ts.runtime[i] = cpu_to_le64(ts->runtime[i]);
1549 }
1550
1551 p.ts.total_run_time = cpu_to_le64(ts->total_run_time);
1552 p.ts.continue_on_error = cpu_to_le16(ts->continue_on_error);
1553 p.ts.total_err_count = cpu_to_le64(ts->total_err_count);
1554 p.ts.first_error = cpu_to_le32(ts->first_error);
1555 p.ts.kb_base = cpu_to_le32(ts->kb_base);
1556 p.ts.unit_base = cpu_to_le32(ts->unit_base);
1557
1558 p.ts.latency_depth = cpu_to_le32(ts->latency_depth);
1559 p.ts.latency_target = cpu_to_le64(ts->latency_target);
1560 p.ts.latency_window = cpu_to_le64(ts->latency_window);
1561 p.ts.latency_percentile.u.i = cpu_to_le64(fio_double_to_uint64(ts->latency_percentile.u.f));
1562
1563 p.ts.sig_figs = cpu_to_le32(ts->sig_figs);
1564
1565 p.ts.nr_block_infos = cpu_to_le64(ts->nr_block_infos);
1566 for (i = 0; i < p.ts.nr_block_infos; i++)
1567 p.ts.block_infos[i] = cpu_to_le32(ts->block_infos[i]);
1568
1569 p.ts.ss_dur = cpu_to_le64(ts->ss_dur);
1570 p.ts.ss_state = cpu_to_le32(ts->ss_state);
1571 p.ts.ss_head = cpu_to_le32(ts->ss_head);
1572 p.ts.ss_limit.u.i = cpu_to_le64(fio_double_to_uint64(ts->ss_limit.u.f));
1573 p.ts.ss_slope.u.i = cpu_to_le64(fio_double_to_uint64(ts->ss_slope.u.f));
1574 p.ts.ss_deviation.u.i = cpu_to_le64(fio_double_to_uint64(ts->ss_deviation.u.f));
1575 p.ts.ss_criterion.u.i = cpu_to_le64(fio_double_to_uint64(ts->ss_criterion.u.f));
1576
1577 p.ts.cachehit = cpu_to_le64(ts->cachehit);
1578 p.ts.cachemiss = cpu_to_le64(ts->cachemiss);
1579
1580 for (i = 0; i < DDIR_RWDIR_CNT; i++) {
1581 for (j = 0; j < FIO_IO_U_PLAT_NR; j++) {
1582 p.ts.io_u_plat_high_prio[i][j] = cpu_to_le64(ts->io_u_plat_high_prio[i][j]);
1583 p.ts.io_u_plat_low_prio[i][j] = cpu_to_le64(ts->io_u_plat_low_prio[i][j]);
1584 }
1585 convert_io_stat(&p.ts.clat_high_prio_stat[i], &ts->clat_high_prio_stat[i]);
1586 convert_io_stat(&p.ts.clat_low_prio_stat[i], &ts->clat_low_prio_stat[i]);
1587 }
1588
1589 convert_gs(&p.rs, rs);
1590
1591 dprint(FD_NET, "ts->ss_state = %d\n", ts->ss_state);
1592 if (ts->ss_state & FIO_SS_DATA) {
1593 dprint(FD_NET, "server sending steadystate ring buffers\n");
1594
1595 ss_buf = malloc(sizeof(p) + 2*ts->ss_dur*sizeof(uint64_t));
1596
1597 memcpy(ss_buf, &p, sizeof(p));
1598
1599 ss_iops = (uint64_t *) ((struct cmd_ts_pdu *)ss_buf + 1);
1600 ss_bw = ss_iops + (int) ts->ss_dur;
1601 for (i = 0; i < ts->ss_dur; i++) {
1602 ss_iops[i] = cpu_to_le64(ts->ss_iops_data[i]);
1603 ss_bw[i] = cpu_to_le64(ts->ss_bw_data[i]);
1604 }
1605
1606 fio_net_queue_cmd(FIO_NET_CMD_TS, ss_buf, sizeof(p) + 2*ts->ss_dur*sizeof(uint64_t), NULL, SK_F_COPY);
1607
1608 free(ss_buf);
1609 }
1610 else
1611 fio_net_queue_cmd(FIO_NET_CMD_TS, &p, sizeof(p), NULL, SK_F_COPY);
1612}
1613
1614void fio_server_send_gs(struct group_run_stats *rs)
1615{
1616 struct group_run_stats gs;
1617
1618 dprint(FD_NET, "server sending group run stats\n");
1619
1620 convert_gs(&gs, rs);
1621 fio_net_queue_cmd(FIO_NET_CMD_GS, &gs, sizeof(gs), NULL, SK_F_COPY);
1622}
1623
1624void fio_server_send_job_options(struct flist_head *opt_list,
1625 unsigned int gid)
1626{
1627 struct cmd_job_option pdu;
1628 struct flist_head *entry;
1629
1630 if (flist_empty(opt_list))
1631 return;
1632
1633 flist_for_each(entry, opt_list) {
1634 struct print_option *p;
1635 size_t len;
1636
1637 p = flist_entry(entry, struct print_option, list);
1638 memset(&pdu, 0, sizeof(pdu));
1639
1640 if (gid == -1U) {
1641 pdu.global = __cpu_to_le16(1);
1642 pdu.groupid = 0;
1643 } else {
1644 pdu.global = 0;
1645 pdu.groupid = cpu_to_le32(gid);
1646 }
1647 len = strlen(p->name);
1648 if (len >= sizeof(pdu.name)) {
1649 len = sizeof(pdu.name) - 1;
1650 pdu.truncated = __cpu_to_le16(1);
1651 }
1652 memcpy(pdu.name, p->name, len);
1653 if (p->value) {
1654 len = strlen(p->value);
1655 if (len >= sizeof(pdu.value)) {
1656 len = sizeof(pdu.value) - 1;
1657 pdu.truncated = __cpu_to_le16(1);
1658 }
1659 memcpy(pdu.value, p->value, len);
1660 }
1661 fio_net_queue_cmd(FIO_NET_CMD_JOB_OPT, &pdu, sizeof(pdu), NULL, SK_F_COPY);
1662 }
1663}
1664
1665static void convert_agg(struct disk_util_agg *dst, struct disk_util_agg *src)
1666{
1667 int i;
1668
1669 for (i = 0; i < 2; i++) {
1670 dst->ios[i] = cpu_to_le64(src->ios[i]);
1671 dst->merges[i] = cpu_to_le64(src->merges[i]);
1672 dst->sectors[i] = cpu_to_le64(src->sectors[i]);
1673 dst->ticks[i] = cpu_to_le64(src->ticks[i]);
1674 }
1675
1676 dst->io_ticks = cpu_to_le64(src->io_ticks);
1677 dst->time_in_queue = cpu_to_le64(src->time_in_queue);
1678 dst->slavecount = cpu_to_le32(src->slavecount);
1679 dst->max_util.u.i = cpu_to_le64(fio_double_to_uint64(src->max_util.u.f));
1680}
1681
1682static void convert_dus(struct disk_util_stat *dst, struct disk_util_stat *src)
1683{
1684 int i;
1685
1686 snprintf((char *) dst->name, sizeof(dst->name), "%s", src->name);
1687
1688 for (i = 0; i < 2; i++) {
1689 dst->s.ios[i] = cpu_to_le64(src->s.ios[i]);
1690 dst->s.merges[i] = cpu_to_le64(src->s.merges[i]);
1691 dst->s.sectors[i] = cpu_to_le64(src->s.sectors[i]);
1692 dst->s.ticks[i] = cpu_to_le64(src->s.ticks[i]);
1693 }
1694
1695 dst->s.io_ticks = cpu_to_le64(src->s.io_ticks);
1696 dst->s.time_in_queue = cpu_to_le64(src->s.time_in_queue);
1697 dst->s.msec = cpu_to_le64(src->s.msec);
1698}
1699
1700void fio_server_send_du(void)
1701{
1702 struct disk_util *du;
1703 struct flist_head *entry;
1704 struct cmd_du_pdu pdu;
1705
1706 dprint(FD_NET, "server: sending disk_util %d\n", !flist_empty(&disk_list));
1707
1708 memset(&pdu, 0, sizeof(pdu));
1709
1710 flist_for_each(entry, &disk_list) {
1711 du = flist_entry(entry, struct disk_util, list);
1712
1713 convert_dus(&pdu.dus, &du->dus);
1714 convert_agg(&pdu.agg, &du->agg);
1715
1716 fio_net_queue_cmd(FIO_NET_CMD_DU, &pdu, sizeof(pdu), NULL, SK_F_COPY);
1717 }
1718}
1719
1720#ifdef CONFIG_ZLIB
1721
1722static inline void __fio_net_prep_tail(z_stream *stream, void *out_pdu,
1723 struct sk_entry **last_entry,
1724 struct sk_entry *first)
1725{
1726 unsigned int this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream->avail_out;
1727
1728 *last_entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, out_pdu, this_len,
1729 NULL, SK_F_VEC | SK_F_INLINE | SK_F_FREE);
1730 if (*last_entry)
1731 flist_add_tail(&(*last_entry)->list, &first->next);
1732}
1733
1734/*
1735 * Deflates the next input given, creating as many new packets in the
1736 * linked list as necessary.
1737 */
1738static int __deflate_pdu_buffer(void *next_in, unsigned int next_sz, void **out_pdu,
1739 struct sk_entry **last_entry, z_stream *stream,
1740 struct sk_entry *first)
1741{
1742 int ret;
1743
1744 stream->next_in = next_in;
1745 stream->avail_in = next_sz;
1746 do {
1747 if (!stream->avail_out) {
1748 __fio_net_prep_tail(stream, *out_pdu, last_entry, first);
1749 if (*last_entry == NULL)
1750 return 1;
1751
1752 *out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
1753
1754 stream->avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
1755 stream->next_out = *out_pdu;
1756 }
1757
1758 ret = deflate(stream, Z_BLOCK);
1759
1760 if (ret < 0) {
1761 free(*out_pdu);
1762 return 1;
1763 }
1764 } while (stream->avail_in);
1765
1766 return 0;
1767}
1768
1769static int __fio_append_iolog_gz_hist(struct sk_entry *first, struct io_log *log,
1770 struct io_logs *cur_log, z_stream *stream)
1771{
1772 struct sk_entry *entry;
1773 void *out_pdu;
1774 int ret, i, j;
1775 int sample_sz = log_entry_sz(log);
1776
1777 out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
1778 stream->avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
1779 stream->next_out = out_pdu;
1780
1781 for (i = 0; i < cur_log->nr_samples; i++) {
1782 struct io_sample *s;
1783 struct io_u_plat_entry *cur_plat_entry, *prev_plat_entry;
1784 uint64_t *cur_plat, *prev_plat;
1785
1786 s = get_sample(log, cur_log, i);
1787 ret = __deflate_pdu_buffer(s, sample_sz, &out_pdu, &entry, stream, first);
1788 if (ret)
1789 return ret;
1790
1791 /* Do the subtraction on server side so that client doesn't have to
1792 * reconstruct our linked list from packets.
1793 */
1794 cur_plat_entry = s->data.plat_entry;
1795 prev_plat_entry = flist_first_entry(&cur_plat_entry->list, struct io_u_plat_entry, list);
1796 cur_plat = cur_plat_entry->io_u_plat;
1797 prev_plat = prev_plat_entry->io_u_plat;
1798
1799 for (j = 0; j < FIO_IO_U_PLAT_NR; j++) {
1800 cur_plat[j] -= prev_plat[j];
1801 }
1802
1803 flist_del(&prev_plat_entry->list);
1804 free(prev_plat_entry);
1805
1806 ret = __deflate_pdu_buffer(cur_plat_entry, sizeof(*cur_plat_entry),
1807 &out_pdu, &entry, stream, first);
1808
1809 if (ret)
1810 return ret;
1811 }
1812
1813 __fio_net_prep_tail(stream, out_pdu, &entry, first);
1814 return entry == NULL;
1815}
1816
1817static int __fio_append_iolog_gz(struct sk_entry *first, struct io_log *log,
1818 struct io_logs *cur_log, z_stream *stream)
1819{
1820 unsigned int this_len;
1821 void *out_pdu;
1822 int ret;
1823
1824 if (log->log_type == IO_LOG_TYPE_HIST)
1825 return __fio_append_iolog_gz_hist(first, log, cur_log, stream);
1826
1827 stream->next_in = (void *) cur_log->log;
1828 stream->avail_in = cur_log->nr_samples * log_entry_sz(log);
1829
1830 do {
1831 struct sk_entry *entry;
1832
1833 /*
1834 * Dirty - since the log is potentially huge, compress it into
1835 * FIO_SERVER_MAX_FRAGMENT_PDU chunks and let the receiving
1836 * side defragment it.
1837 */
1838 out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
1839
1840 stream->avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
1841 stream->next_out = out_pdu;
1842 ret = deflate(stream, Z_BLOCK);
1843 /* may be Z_OK, or Z_STREAM_END */
1844 if (ret < 0) {
1845 free(out_pdu);
1846 return 1;
1847 }
1848
1849 this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream->avail_out;
1850
1851 entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, out_pdu, this_len,
1852 NULL, SK_F_VEC | SK_F_INLINE | SK_F_FREE);
1853 if (!entry) {
1854 free(out_pdu);
1855 return 1;
1856 }
1857 flist_add_tail(&entry->list, &first->next);
1858 } while (stream->avail_in);
1859
1860 return 0;
1861}
1862
1863static int fio_append_iolog_gz(struct sk_entry *first, struct io_log *log)
1864{
1865 z_stream stream = {
1866 .zalloc = Z_NULL,
1867 .zfree = Z_NULL,
1868 .opaque = Z_NULL,
1869 };
1870 int ret = 0;
1871
1872 if (deflateInit(&stream, Z_DEFAULT_COMPRESSION) != Z_OK)
1873 return 1;
1874
1875 while (!flist_empty(&log->io_logs)) {
1876 struct io_logs *cur_log;
1877
1878 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
1879 flist_del_init(&cur_log->list);
1880
1881 ret = __fio_append_iolog_gz(first, log, cur_log, &stream);
1882 if (ret)
1883 break;
1884 }
1885
1886 ret = deflate(&stream, Z_FINISH);
1887
1888 while (ret != Z_STREAM_END) {
1889 struct sk_entry *entry;
1890 unsigned int this_len;
1891 void *out_pdu;
1892
1893 out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
1894 stream.avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
1895 stream.next_out = out_pdu;
1896
1897 ret = deflate(&stream, Z_FINISH);
1898 /* may be Z_OK, or Z_STREAM_END */
1899 if (ret < 0) {
1900 free(out_pdu);
1901 break;
1902 }
1903
1904 this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream.avail_out;
1905
1906 entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, out_pdu, this_len,
1907 NULL, SK_F_VEC | SK_F_INLINE | SK_F_FREE);
1908 if (!entry) {
1909 free(out_pdu);
1910 break;
1911 }
1912 flist_add_tail(&entry->list, &first->next);
1913 }
1914
1915 ret = deflateEnd(&stream);
1916 if (ret == Z_OK)
1917 return 0;
1918
1919 return 1;
1920}
1921#else
1922static int fio_append_iolog_gz(struct sk_entry *first, struct io_log *log)
1923{
1924 return 1;
1925}
1926#endif
1927
1928static int fio_append_gz_chunks(struct sk_entry *first, struct io_log *log)
1929{
1930 struct sk_entry *entry;
1931 struct flist_head *node;
1932 int ret = 0;
1933
1934 pthread_mutex_lock(&log->chunk_lock);
1935 flist_for_each(node, &log->chunk_list) {
1936 struct iolog_compress *c;
1937
1938 c = flist_entry(node, struct iolog_compress, list);
1939 entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, c->buf, c->len,
1940 NULL, SK_F_VEC | SK_F_INLINE);
1941 if (!entry) {
1942 ret = 1;
1943 break;
1944 }
1945 flist_add_tail(&entry->list, &first->next);
1946 }
1947 pthread_mutex_unlock(&log->chunk_lock);
1948 return ret;
1949}
1950
1951static int fio_append_text_log(struct sk_entry *first, struct io_log *log)
1952{
1953 struct sk_entry *entry;
1954 int ret = 0;
1955
1956 while (!flist_empty(&log->io_logs)) {
1957 struct io_logs *cur_log;
1958 size_t size;
1959
1960 cur_log = flist_first_entry(&log->io_logs, struct io_logs, list);
1961 flist_del_init(&cur_log->list);
1962
1963 size = cur_log->nr_samples * log_entry_sz(log);
1964
1965 entry = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, cur_log->log, size,
1966 NULL, SK_F_VEC | SK_F_INLINE);
1967 if (!entry) {
1968 ret = 1;
1969 break;
1970 }
1971 flist_add_tail(&entry->list, &first->next);
1972 }
1973
1974 return ret;
1975}
1976
1977int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
1978{
1979 struct cmd_iolog_pdu pdu = {
1980 .nr_samples = cpu_to_le64(iolog_nr_samples(log)),
1981 .thread_number = cpu_to_le32(td->thread_number),
1982 .log_type = cpu_to_le32(log->log_type),
1983 .log_hist_coarseness = cpu_to_le32(log->hist_coarseness),
1984 };
1985 struct sk_entry *first;
1986 struct flist_head *entry;
1987 int ret = 0;
1988
1989 if (!flist_empty(&log->chunk_list))
1990 pdu.compressed = __cpu_to_le32(STORE_COMPRESSED);
1991 else if (use_zlib)
1992 pdu.compressed = __cpu_to_le32(XMIT_COMPRESSED);
1993 else
1994 pdu.compressed = 0;
1995
1996 snprintf((char *) pdu.name, sizeof(pdu.name), "%s", name);
1997
1998 /*
1999 * We can't do this for a pre-compressed log, but for that case,
2000 * log->nr_samples is zero anyway.
2001 */
2002 flist_for_each(entry, &log->io_logs) {
2003 struct io_logs *cur_log;
2004 int i;
2005
2006 cur_log = flist_entry(entry, struct io_logs, list);
2007
2008 for (i = 0; i < cur_log->nr_samples; i++) {
2009 struct io_sample *s = get_sample(log, cur_log, i);
2010
2011 s->time = cpu_to_le64(s->time);
2012 s->data.val = cpu_to_le64(s->data.val);
2013 s->__ddir = __cpu_to_le32(s->__ddir);
2014 s->bs = cpu_to_le64(s->bs);
2015
2016 if (log->log_offset) {
2017 struct io_sample_offset *so = (void *) s;
2018
2019 so->offset = cpu_to_le64(so->offset);
2020 }
2021 }
2022 }
2023
2024 /*
2025 * Assemble header entry first
2026 */
2027 first = fio_net_prep_cmd(FIO_NET_CMD_IOLOG, &pdu, sizeof(pdu), NULL, SK_F_VEC | SK_F_INLINE | SK_F_COPY);
2028 if (!first)
2029 return 1;
2030
2031 /*
2032 * Now append actual log entries. If log compression was enabled on
2033 * the job, just send out the compressed chunks directly. If we
2034 * have a plain log, compress if we can, then send. Otherwise, send
2035 * the plain text output.
2036 */
2037 if (!flist_empty(&log->chunk_list))
2038 ret = fio_append_gz_chunks(first, log);
2039 else if (use_zlib)
2040 ret = fio_append_iolog_gz(first, log);
2041 else
2042 ret = fio_append_text_log(first, log);
2043
2044 fio_net_queue_entry(first);
2045 return ret;
2046}
2047
2048void fio_server_send_add_job(struct thread_data *td)
2049{
2050 struct cmd_add_job_pdu pdu = {
2051 .thread_number = cpu_to_le32(td->thread_number),
2052 .groupid = cpu_to_le32(td->groupid),
2053 };
2054
2055 convert_thread_options_to_net(&pdu.top, &td->o);
2056
2057 fio_net_queue_cmd(FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), NULL,
2058 SK_F_COPY);
2059}
2060
2061void fio_server_send_start(struct thread_data *td)
2062{
2063 struct sk_out *sk_out = pthread_getspecific(sk_out_key);
2064
2065 assert(sk_out->sk != -1);
2066
2067 fio_net_queue_cmd(FIO_NET_CMD_SERVER_START, NULL, 0, NULL, SK_F_SIMPLE);
2068}
2069
2070int fio_server_get_verify_state(const char *name, int threadnumber,
2071 void **datap)
2072{
2073 struct thread_io_list *s;
2074 struct cmd_sendfile out;
2075 struct cmd_reply *rep;
2076 uint64_t tag;
2077 void *data;
2078 int ret;
2079
2080 dprint(FD_NET, "server: request verify state\n");
2081
2082 rep = smalloc(sizeof(*rep));
2083 if (!rep)
2084 return ENOMEM;
2085
2086 __fio_sem_init(&rep->lock, FIO_SEM_LOCKED);
2087 rep->data = NULL;
2088 rep->error = 0;
2089
2090 verify_state_gen_name((char *) out.path, sizeof(out.path), name, me,
2091 threadnumber);
2092 tag = (uint64_t) (uintptr_t) rep;
2093 fio_net_queue_cmd(FIO_NET_CMD_SENDFILE, &out, sizeof(out), &tag,
2094 SK_F_COPY);
2095
2096 /*
2097 * Wait for the backend to receive the reply
2098 */
2099 if (fio_sem_down_timeout(&rep->lock, 10000)) {
2100 log_err("fio: timed out waiting for reply\n");
2101 ret = ETIMEDOUT;
2102 goto fail;
2103 }
2104
2105 if (rep->error) {
2106 log_err("fio: failure on receiving state file %s: %s\n",
2107 out.path, strerror(rep->error));
2108 ret = rep->error;
2109fail:
2110 *datap = NULL;
2111 sfree(rep);
2112 fio_net_queue_quit();
2113 return ret;
2114 }
2115
2116 /*
2117 * The format is verify_state_hdr, then thread_io_list. Verify
2118 * the header, and the thread_io_list checksum
2119 */
2120 s = rep->data + sizeof(struct verify_state_hdr);
2121 if (verify_state_hdr(rep->data, s)) {
2122 ret = EILSEQ;
2123 goto fail;
2124 }
2125
2126 /*
2127 * Don't need the header from now, copy just the thread_io_list
2128 */
2129 ret = 0;
2130 rep->size -= sizeof(struct verify_state_hdr);
2131 data = malloc(rep->size);
2132 memcpy(data, s, rep->size);
2133 *datap = data;
2134
2135 sfree(rep->data);
2136 __fio_sem_remove(&rep->lock);
2137 sfree(rep);
2138 return ret;
2139}
2140
2141static int fio_init_server_ip(void)
2142{
2143 struct sockaddr *addr;
2144 socklen_t socklen;
2145 char buf[80];
2146 const char *str;
2147 int sk, opt;
2148
2149 if (use_ipv6)
2150 sk = socket(AF_INET6, SOCK_STREAM, 0);
2151 else
2152 sk = socket(AF_INET, SOCK_STREAM, 0);
2153
2154 if (sk < 0) {
2155 log_err("fio: socket: %s\n", strerror(errno));
2156 return -1;
2157 }
2158
2159 opt = 1;
2160 if (setsockopt(sk, SOL_SOCKET, SO_REUSEADDR, (void *)&opt, sizeof(opt)) < 0) {
2161 log_err("fio: setsockopt(REUSEADDR): %s\n", strerror(errno));
2162 close(sk);
2163 return -1;
2164 }
2165#ifdef SO_REUSEPORT
2166 /*
2167 * Not fatal if fails, so just ignore it if that happens
2168 */
2169 if (setsockopt(sk, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt))) {
2170 }
2171#endif
2172
2173 if (use_ipv6) {
2174 void *src = &saddr_in6.sin6_addr;
2175
2176 addr = (struct sockaddr *) &saddr_in6;
2177 socklen = sizeof(saddr_in6);
2178 saddr_in6.sin6_family = AF_INET6;
2179 str = inet_ntop(AF_INET6, src, buf, sizeof(buf));
2180 } else {
2181 void *src = &saddr_in.sin_addr;
2182
2183 addr = (struct sockaddr *) &saddr_in;
2184 socklen = sizeof(saddr_in);
2185 saddr_in.sin_family = AF_INET;
2186 str = inet_ntop(AF_INET, src, buf, sizeof(buf));
2187 }
2188
2189 if (bind(sk, addr, socklen) < 0) {
2190 log_err("fio: bind: %s\n", strerror(errno));
2191 log_info("fio: failed with IPv%c %s\n", use_ipv6 ? '6' : '4', str);
2192 close(sk);
2193 return -1;
2194 }
2195
2196 return sk;
2197}
2198
2199static int fio_init_server_sock(void)
2200{
2201 struct sockaddr_un addr;
2202 socklen_t len;
2203 mode_t mode;
2204 int sk;
2205
2206 sk = socket(AF_UNIX, SOCK_STREAM, 0);
2207 if (sk < 0) {
2208 log_err("fio: socket: %s\n", strerror(errno));
2209 return -1;
2210 }
2211
2212 mode = umask(000);
2213
2214 addr.sun_family = AF_UNIX;
2215 snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", bind_sock);
2216
2217 len = sizeof(addr.sun_family) + strlen(bind_sock) + 1;
2218
2219 if (bind(sk, (struct sockaddr *) &addr, len) < 0) {
2220 log_err("fio: bind: %s\n", strerror(errno));
2221 close(sk);
2222 return -1;
2223 }
2224
2225 umask(mode);
2226 return sk;
2227}
2228
2229static int fio_init_server_connection(void)
2230{
2231 char bind_str[128];
2232 int sk;
2233
2234 dprint(FD_NET, "starting server\n");
2235
2236 if (!bind_sock)
2237 sk = fio_init_server_ip();
2238 else
2239 sk = fio_init_server_sock();
2240
2241 if (sk < 0)
2242 return sk;
2243
2244 memset(bind_str, 0, sizeof(bind_str));
2245
2246 if (!bind_sock) {
2247 char *p, port[16];
2248 void *src;
2249 int af;
2250
2251 if (use_ipv6) {
2252 af = AF_INET6;
2253 src = &saddr_in6.sin6_addr;
2254 } else {
2255 af = AF_INET;
2256 src = &saddr_in.sin_addr;
2257 }
2258
2259 p = (char *) inet_ntop(af, src, bind_str, sizeof(bind_str));
2260
2261 sprintf(port, ",%u", fio_net_port);
2262 if (p)
2263 strcat(p, port);
2264 else
2265 snprintf(bind_str, sizeof(bind_str), "%s", port);
2266 } else
2267 snprintf(bind_str, sizeof(bind_str), "%s", bind_sock);
2268
2269 log_info("fio: server listening on %s\n", bind_str);
2270
2271 if (listen(sk, 4) < 0) {
2272 log_err("fio: listen: %s\n", strerror(errno));
2273 close(sk);
2274 return -1;
2275 }
2276
2277 return sk;
2278}
2279
2280int fio_server_parse_host(const char *host, int ipv6, struct in_addr *inp,
2281 struct in6_addr *inp6)
2282
2283{
2284 int ret = 0;
2285
2286 if (ipv6)
2287 ret = inet_pton(AF_INET6, host, inp6);
2288 else
2289 ret = inet_pton(AF_INET, host, inp);
2290
2291 if (ret != 1) {
2292 struct addrinfo *res, hints = {
2293 .ai_family = ipv6 ? AF_INET6 : AF_INET,
2294 .ai_socktype = SOCK_STREAM,
2295 };
2296
2297 ret = getaddrinfo(host, NULL, &hints, &res);
2298 if (ret) {
2299 log_err("fio: failed to resolve <%s> (%s)\n", host,
2300 gai_strerror(ret));
2301 return 1;
2302 }
2303
2304 if (ipv6)
2305 memcpy(inp6, &((struct sockaddr_in6 *) res->ai_addr)->sin6_addr, sizeof(*inp6));
2306 else
2307 memcpy(inp, &((struct sockaddr_in *) res->ai_addr)->sin_addr, sizeof(*inp));
2308
2309 ret = 1;
2310 freeaddrinfo(res);
2311 }
2312
2313 return !(ret == 1);
2314}
2315
2316/*
2317 * Parse a host/ip/port string. Reads from 'str'.
2318 *
2319 * Outputs:
2320 *
2321 * For IPv4:
2322 * *ptr is the host, *port is the port, inp is the destination.
2323 * For IPv6:
2324 * *ptr is the host, *port is the port, inp6 is the dest, and *ipv6 is 1.
2325 * For local domain sockets:
2326 * *ptr is the filename, *is_sock is 1.
2327 */
2328int fio_server_parse_string(const char *str, char **ptr, bool *is_sock,
2329 int *port, struct in_addr *inp,
2330 struct in6_addr *inp6, int *ipv6)
2331{
2332 const char *host = str;
2333 char *portp;
2334 int lport = 0;
2335
2336 *ptr = NULL;
2337 *is_sock = false;
2338 *port = fio_net_port;
2339 *ipv6 = 0;
2340
2341 if (!strncmp(str, "sock:", 5)) {
2342 *ptr = strdup(str + 5);
2343 *is_sock = true;
2344
2345 return 0;
2346 }
2347
2348 /*
2349 * Is it ip:<ip or host>:port
2350 */
2351 if (!strncmp(host, "ip:", 3))
2352 host += 3;
2353 else if (!strncmp(host, "ip4:", 4))
2354 host += 4;
2355 else if (!strncmp(host, "ip6:", 4)) {
2356 host += 4;
2357 *ipv6 = 1;
2358 } else if (host[0] == ':') {
2359 /* String is :port */
2360 host++;
2361 lport = atoi(host);
2362 if (!lport || lport > 65535) {
2363 log_err("fio: bad server port %u\n", lport);
2364 return 1;
2365 }
2366 /* no hostname given, we are done */
2367 *port = lport;
2368 return 0;
2369 }
2370
2371 /*
2372 * If no port seen yet, check if there's a last ',' at the end
2373 */
2374 if (!lport) {
2375 portp = strchr(host, ',');
2376 if (portp) {
2377 *portp = '\0';
2378 portp++;
2379 lport = atoi(portp);
2380 if (!lport || lport > 65535) {
2381 log_err("fio: bad server port %u\n", lport);
2382 return 1;
2383 }
2384 }
2385 }
2386
2387 if (lport)
2388 *port = lport;
2389
2390 if (!strlen(host))
2391 return 0;
2392
2393 *ptr = strdup(host);
2394
2395 if (fio_server_parse_host(*ptr, *ipv6, inp, inp6)) {
2396 free(*ptr);
2397 *ptr = NULL;
2398 return 1;
2399 }
2400
2401 if (*port == 0)
2402 *port = fio_net_port;
2403
2404 return 0;
2405}
2406
2407/*
2408 * Server arg should be one of:
2409 *
2410 * sock:/path/to/socket
2411 * ip:1.2.3.4
2412 * 1.2.3.4
2413 *
2414 * Where sock uses unix domain sockets, and ip binds the server to
2415 * a specific interface. If no arguments are given to the server, it
2416 * uses IP and binds to 0.0.0.0.
2417 *
2418 */
2419static int fio_handle_server_arg(void)
2420{
2421 int port = fio_net_port;
2422 bool is_sock;
2423 int ret = 0;
2424
2425 saddr_in.sin_addr.s_addr = htonl(INADDR_ANY);
2426
2427 if (!fio_server_arg)
2428 goto out;
2429
2430 ret = fio_server_parse_string(fio_server_arg, &bind_sock, &is_sock,
2431 &port, &saddr_in.sin_addr,
2432 &saddr_in6.sin6_addr, &use_ipv6);
2433
2434 if (!is_sock && bind_sock) {
2435 free(bind_sock);
2436 bind_sock = NULL;
2437 }
2438
2439out:
2440 fio_net_port = port;
2441 saddr_in.sin_port = htons(port);
2442 saddr_in6.sin6_port = htons(port);
2443 return ret;
2444}
2445
2446static void sig_int(int sig)
2447{
2448 if (bind_sock)
2449 unlink(bind_sock);
2450}
2451
2452static void set_sig_handlers(void)
2453{
2454 struct sigaction act = {
2455 .sa_handler = sig_int,
2456 .sa_flags = SA_RESTART,
2457 };
2458
2459 sigaction(SIGINT, &act, NULL);
2460
2461 /* Windows uses SIGBREAK as a quit signal from other applications */
2462#ifdef WIN32
2463 sigaction(SIGBREAK, &act, NULL);
2464#endif
2465}
2466
2467void fio_server_destroy_sk_key(void)
2468{
2469 pthread_key_delete(sk_out_key);
2470}
2471
2472int fio_server_create_sk_key(void)
2473{
2474 if (pthread_key_create(&sk_out_key, NULL)) {
2475 log_err("fio: can't create sk_out backend key\n");
2476 return 1;
2477 }
2478
2479 pthread_setspecific(sk_out_key, NULL);
2480 return 0;
2481}
2482
2483static int fio_server(void)
2484{
2485 int sk, ret;
2486
2487 dprint(FD_NET, "starting server\n");
2488
2489 if (fio_handle_server_arg())
2490 return -1;
2491
2492 sk = fio_init_server_connection();
2493 if (sk < 0)
2494 return -1;
2495
2496 set_sig_handlers();
2497
2498 ret = accept_loop(sk);
2499
2500 close(sk);
2501
2502 if (fio_server_arg) {
2503 free(fio_server_arg);
2504 fio_server_arg = NULL;
2505 }
2506 if (bind_sock)
2507 free(bind_sock);
2508
2509 return ret;
2510}
2511
2512void fio_server_got_signal(int signal)
2513{
2514 struct sk_out *sk_out = pthread_getspecific(sk_out_key);
2515
2516 assert(sk_out);
2517
2518 if (signal == SIGPIPE)
2519 sk_out->sk = -1;
2520 else {
2521 log_info("\nfio: terminating on signal %d\n", signal);
2522 exit_backend = true;
2523 }
2524}
2525
2526static int check_existing_pidfile(const char *pidfile)
2527{
2528 struct stat sb;
2529 char buf[16];
2530 pid_t pid;
2531 FILE *f;
2532
2533 if (stat(pidfile, &sb))
2534 return 0;
2535
2536 f = fopen(pidfile, "r");
2537 if (!f)
2538 return 0;
2539
2540 if (fread(buf, sb.st_size, 1, f) <= 0) {
2541 fclose(f);
2542 return 1;
2543 }
2544 fclose(f);
2545
2546 pid = atoi(buf);
2547 if (kill(pid, SIGCONT) < 0)
2548 return errno != ESRCH;
2549
2550 return 1;
2551}
2552
2553static int write_pid(pid_t pid, const char *pidfile)
2554{
2555 FILE *fpid;
2556
2557 fpid = fopen(pidfile, "w");
2558 if (!fpid) {
2559 log_err("fio: failed opening pid file %s\n", pidfile);
2560 return 1;
2561 }
2562
2563 fprintf(fpid, "%u\n", (unsigned int) pid);
2564 fclose(fpid);
2565 return 0;
2566}
2567
2568/*
2569 * If pidfile is specified, background us.
2570 */
2571int fio_start_server(char *pidfile)
2572{
2573 FILE *file;
2574 pid_t pid;
2575 int ret;
2576
2577#if defined(WIN32)
2578 WSADATA wsd;
2579 WSAStartup(MAKEWORD(2, 2), &wsd);
2580#endif
2581
2582 if (!pidfile)
2583 return fio_server();
2584
2585 if (check_existing_pidfile(pidfile)) {
2586 log_err("fio: pidfile %s exists and server appears alive\n",
2587 pidfile);
2588 free(pidfile);
2589 return -1;
2590 }
2591
2592 pid = fork();
2593 if (pid < 0) {
2594 log_err("fio: failed server fork: %s\n", strerror(errno));
2595 free(pidfile);
2596 return -1;
2597 } else if (pid) {
2598 ret = write_pid(pid, pidfile);
2599 free(pidfile);
2600 _exit(ret);
2601 }
2602
2603 setsid();
2604 openlog("fio", LOG_NDELAY|LOG_NOWAIT|LOG_PID, LOG_USER);
2605 log_syslog = true;
2606
2607 file = freopen("/dev/null", "r", stdin);
2608 if (!file)
2609 perror("freopen");
2610
2611 file = freopen("/dev/null", "w", stdout);
2612 if (!file)
2613 perror("freopen");
2614
2615 file = freopen("/dev/null", "w", stderr);
2616 if (!file)
2617 perror("freopen");
2618
2619 f_out = NULL;
2620 f_err = NULL;
2621
2622 ret = fio_server();
2623
2624 fclose(stdin);
2625 fclose(stdout);
2626 fclose(stderr);
2627
2628 closelog();
2629 unlink(pidfile);
2630 free(pidfile);
2631 return ret;
2632}
2633
2634void fio_server_set_arg(const char *arg)
2635{
2636 fio_server_arg = strdup(arg);
2637}
fio - Flexible IO Tester