#include <netdb.h>
#include <syslog.h>
#include <signal.h>
+#include <zlib.h>
#include "fio.h"
#include "server.h"
"START",
"STOP",
"DISK_UTIL",
- "RUN",
+ "SERVER_START",
"ADD_JOB",
+ "CMD_RUN"
};
const char *fio_server_op(unsigned int op)
int fio_send_data(int sk, const void *p, unsigned int len)
{
- assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_PDU);
+ assert(len <= sizeof(struct fio_net_cmd) + FIO_SERVER_MAX_FRAGMENT_PDU);
do {
int ret = send(sk, p, len, 0);
if (!len)
return 0;
+ if (errno)
+ return -errno;
+
return 1;
}
return 1;
}
- if (cmd->pdu_len > FIO_SERVER_MAX_PDU) {
+ if (cmd->pdu_len > FIO_SERVER_MAX_FRAGMENT_PDU) {
log_err("fio: command payload too large: %u\n", cmd->pdu_len);
return 1;
}
buf[pdu->buf_len ] = '\0';
} else if (cmdret->opcode == FIO_NET_CMD_JOB) {
- char *buf = (char *) cmdret->payload;
+ struct cmd_job_pdu *pdu = (struct cmd_job_pdu *) cmdret->payload;
+ char *buf = (char *) pdu->buf;
+ int len = le32_to_cpu(pdu->buf_len);
- buf[cmdret->pdu_len ] = '\0';
+ buf[len] = '\0';
}
}
return cmdret;
}
-void fio_net_cmd_crc(struct fio_net_cmd *cmd)
+void fio_net_cmd_crc_pdu(struct fio_net_cmd *cmd, void *pdu)
{
uint32_t pdu_len;
cmd->cmd_crc16 = __cpu_to_le16(fio_crc16(cmd, FIO_NET_CMD_CRC_SZ));
pdu_len = le32_to_cpu(cmd->pdu_len);
- if (pdu_len)
- cmd->pdu_crc16 = __cpu_to_le16(fio_crc16(cmd->payload, pdu_len));
+ cmd->pdu_crc16 = __cpu_to_le16(fio_crc16(pdu, pdu_len));
+}
+
+void fio_net_cmd_crc(struct fio_net_cmd *cmd)
+{
+ fio_net_cmd_crc_pdu(cmd, cmd->payload);
}
int fio_net_send_cmd(int fd, uint16_t opcode, const void *buf, off_t size,
do {
this_len = size;
- if (this_len > FIO_SERVER_MAX_PDU)
- this_len = FIO_SERVER_MAX_PDU;
+ if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
+ this_len = FIO_SERVER_MAX_FRAGMENT_PDU;
if (!cmd || cur_len < sizeof(*cmd) + this_len) {
if (cmd)
return fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_QUIT, 0, NULL);
}
-static int handle_job_cmd(struct fio_net_cmd *cmd)
+static int handle_run_cmd(struct fio_net_cmd *cmd)
{
- char *buf = (char *) cmd->payload;
- struct cmd_start_pdu spdu;
struct cmd_end_pdu epdu;
int ret;
- if (parse_jobs_ini(buf, 1, 0)) {
- fio_server_send_quit_cmd();
- return -1;
- }
-
- spdu.jobs = cpu_to_le32(thread_number);
- fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0);
-
ret = fio_backend();
epdu.error = ret;
return ret;
}
+static int handle_job_cmd(struct fio_net_cmd *cmd)
+{
+ struct cmd_job_pdu *pdu = (struct cmd_job_pdu *) cmd->payload;
+ void *buf = pdu->buf;
+ struct cmd_start_pdu spdu;
+
+ pdu->buf_len = le32_to_cpu(pdu->buf_len);
+ pdu->client_type = le32_to_cpu(pdu->client_type);
+
+ if (parse_jobs_ini(buf, 1, 0, pdu->client_type)) {
+ fio_server_send_quit_cmd();
+ return -1;
+ }
+
+ spdu.jobs = cpu_to_le32(thread_number);
+ fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0);
+ return 0;
+}
+
static int handle_jobline_cmd(struct fio_net_cmd *cmd)
{
void *pdu = cmd->payload;
struct cmd_single_line_pdu *cslp;
struct cmd_line_pdu *clp;
unsigned long offset;
+ struct cmd_start_pdu spdu;
char **argv;
- int ret, i;
+ int i;
clp = pdu;
clp->lines = le16_to_cpu(clp->lines);
+ clp->client_type = le16_to_cpu(clp->client_type);
argv = malloc(clp->lines * sizeof(char *));
offset = sizeof(*clp);
dprint(FD_NET, "server: %d: %s\n", i, argv[i]);
}
- if (parse_cmd_line(clp->lines, argv)) {
+ if (parse_cmd_line(clp->lines, argv, clp->client_type)) {
fio_server_send_quit_cmd();
free(argv);
return -1;
free(argv);
- fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_START, 0, NULL);
-
- ret = fio_backend();
- fio_server_send_quit_cmd();
- reset_fio_state();
- first_cmd_check = 0;
- return ret;
+ spdu.jobs = cpu_to_le32(thread_number);
+ fio_net_send_cmd(server_fd, FIO_NET_CMD_START, &spdu, sizeof(spdu), 0);
+ return 0;
}
static int handle_probe_cmd(struct fio_net_cmd *cmd)
je->iops[i] = cpu_to_le32(je->iops[i]);
}
- je->elapsed_sec = cpu_to_le32(je->nr_running);
+ je->elapsed_sec = cpu_to_le64(je->elapsed_sec);
je->eta_sec = cpu_to_le64(je->eta_sec);
+ je->nr_threads = cpu_to_le32(je->nr_threads);
fio_net_send_cmd(server_fd, FIO_NET_CMD_ETA, je, size, cmd->tag);
free(je);
case FIO_NET_CMD_SEND_ETA:
ret = handle_send_eta_cmd(cmd);
break;
+ case FIO_NET_CMD_RUN:
+ ret = handle_run_cmd(cmd);
+ break;
default:
log_err("fio: unknown opcode: %s\n",fio_server_op(cmd->opcode));
ret = 1;
void fio_server_idle_loop(void)
{
if (!first_cmd_check) {
- fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_RUN, 0, NULL);
+ fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_SERVER_START, 0, NULL);
first_cmd_check = 1;
}
if (server_fd != -1)
}
}
-void fio_server_send_add_job(struct thread_options *o, const char *ioengine)
+int fio_send_iolog(struct thread_data *td, struct io_log *log, const char *name)
{
- struct cmd_add_job_pdu pdu;
+ struct cmd_iolog_pdu pdu;
+ struct fio_net_cmd cmd;
+ z_stream stream;
+ void *out_pdu;
int i;
- strcpy((char *) pdu.jobname, o->name);
- strcpy((char *) pdu.ioengine, ioengine);
+ pdu.nr_samples = __cpu_to_le32(log->nr_samples);
+ pdu.log_type = cpu_to_le32(log->log_type);
+ strcpy((char *) pdu.name, name);
- pdu.iodepth = cpu_to_le32(o->iodepth);
- pdu.rw = cpu_to_le32(o->td_ddir);
+ for (i = 0; i < log->nr_samples; i++) {
+ struct io_sample *s = &log->log[i];
- for (i = 0; i < 2; i++) {
- pdu.min_bs[i] = cpu_to_le32(o->min_bs[i]);
- pdu.max_bs[i] = cpu_to_le32(o->max_bs[i]);
+ s->time = cpu_to_le64(s->time);
+ s->val = cpu_to_le64(s->val);
+ s->ddir = cpu_to_le32(s->ddir);
+ s->bs = cpu_to_le32(s->bs);
}
- pdu.numjobs = cpu_to_le32(o->numjobs);
- pdu.group_reporting = cpu_to_le32(o->group_reporting);
+ /*
+ * Dirty - since the log is potentially huge, compress it into
+ * FIO_SERVER_MAX_FRAGMENT_PDU chunks and let the receiving
+ * side defragment it.
+ */
+ out_pdu = malloc(FIO_SERVER_MAX_FRAGMENT_PDU);
+
+ stream.zalloc = Z_NULL;
+ stream.zfree = Z_NULL;
+ stream.opaque = Z_NULL;
+
+ if (deflateInit(&stream, Z_DEFAULT_COMPRESSION) != Z_OK) {
+ free(out_pdu);
+ return 1;
+ }
+
+ /*
+ * Send header first, it's not compressed.
+ */
+ __fio_init_net_cmd(&cmd, FIO_NET_CMD_IOLOG, sizeof(pdu), 0);
+ cmd.flags = __cpu_to_le32(FIO_NET_CMD_F_MORE);
+ fio_net_cmd_crc_pdu(&cmd, &pdu);
+ fio_send_data(server_fd, &cmd, sizeof(cmd));
+ fio_send_data(server_fd, &pdu, sizeof(pdu));
+
+ stream.next_in = (void *) log->log;
+ stream.avail_in = log->nr_samples * sizeof(struct io_sample);
+
+ do {
+ unsigned int this_len;
+
+ stream.avail_out = FIO_SERVER_MAX_FRAGMENT_PDU;
+ stream.next_out = out_pdu;
+ assert(deflate(&stream, Z_FINISH) == Z_OK);
+
+ this_len = FIO_SERVER_MAX_FRAGMENT_PDU - stream.avail_out;
+
+ __fio_init_net_cmd(&cmd, FIO_NET_CMD_IOLOG, this_len, 0);
+
+ if (stream.avail_in)
+ cmd.flags = __cpu_to_le32(FIO_NET_CMD_F_MORE);
+
+ fio_net_cmd_crc_pdu(&cmd, out_pdu);
+
+ fio_send_data(server_fd, &cmd, sizeof(cmd));
+ fio_send_data(server_fd, out_pdu, this_len);
+ } while (stream.avail_in);
+
+ free(out_pdu);
+ deflateEnd(&stream);
+ return 0;
+}
+
+void fio_server_send_add_job(struct thread_options *o, const char *ioengine)
+{
+ struct cmd_add_job_pdu pdu;
+
+ memset(&pdu, 0, sizeof(pdu));
+ convert_thread_options_to_net(&pdu.top, o);
fio_net_send_cmd(server_fd, FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), 0);
}
int ret;
#if defined(WIN32)
- WSADATA wsd;
- WSAStartup(MAKEWORD(2,2), &wsd);
+ WSADATA wsd;
+ WSAStartup(MAKEWORD(2,2), &wsd);
#endif
if (!pidfile)
projects / fio.git / blobdiff