From 132159a5a062cabfe963b3d57e82a80741bf5506 Mon Sep 17 00:00:00 2001 From: Jens Axboe Date: Fri, 30 Sep 2011 15:01:32 -0600 Subject: [PATCH] Add start of client, start of real protocol Signed-off-by: Jens Axboe --- Makefile | 2 +- client.c | 124 ++++++++++++++++++++++++ init.c | 49 +++++++++- server.c | 281 +++++++++++++++++++++++++++++++++++++++++++++++++------ server.h | 64 +++++++++++++ 5 files changed, 488 insertions(+), 32 deletions(-) create mode 100644 client.c diff --git a/Makefile b/Makefile index a4b042b2..7c581927 100644 --- a/Makefile +++ b/Makefile @@ -14,7 +14,7 @@ SOURCE = gettime.c fio.c ioengines.c init.c stat.c log.c time.c filesetup.c \ rbtree.c smalloc.c filehash.c profile.c debug.c lib/rand.c \ lib/num2str.c $(wildcard crc/*.c) engines/cpu.c \ engines/mmap.c engines/sync.c engines/null.c engines/net.c \ - memalign.c server.c + memalign.c server.c client.c ifeq ($(UNAME), Linux) SOURCE += diskutil.c fifo.c blktrace.c helpers.c cgroup.c trim.c \ diff --git a/client.c b/client.c new file mode 100644 index 00000000..c94ccf3f --- /dev/null +++ b/client.c @@ -0,0 +1,124 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "fio.h" +#include "server.h" +#include "crc/crc32.h" + +int fio_client_fd = -1; + +int fio_client_connect(const char *host) +{ + struct sockaddr_in addr; + int fd; + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(fio_net_port); + + if (inet_aton(host, &addr.sin_addr) != 1) { + struct hostent *hent; + + hent = gethostbyname(host); + if (!hent) { + log_err("fio: gethostbyname: %s\n", strerror(errno)); + return 1; + } + + memcpy(&addr.sin_addr, hent->h_addr, 4); + } + + fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + log_err("fio: socket: %s\n", strerror(errno)); + return 1; + } + + if (connect(fd, (struct sockaddr *) &addr, sizeof(addr)) < 0) { + log_err("fio: connect: %s\n", strerror(errno)); + return 1; + } + + fio_client_fd = fd; + return 0; +} + +static int send_file_buf(char *buf, off_t size) +{ + struct fio_net_cmd *cmd; + int ret; + + cmd = malloc(sizeof(*cmd) + size); + + fio_init_net_cmd(cmd); + cmd->opcode = cpu_to_le16(FIO_NET_CMD_JOB_END); + cmd->pdu_len = cpu_to_le32(size); + + memcpy(&cmd->payload, buf, size); + + fio_net_cmd_crc(cmd); + + printf("cmd crc %x, pdu crc %x\n", cmd->cmd_crc32, cmd->pdu_crc32); + printf("job %x\n", cmd->opcode); + + ret = fio_send_data(fio_client_fd, cmd, sizeof(*cmd) + size); + free(cmd); + return ret; +} + +/* + * Send file contents to server backend. We could use sendfile(), but to remain + * more portable lets just read/write the darn thing. + */ +int fio_client_send_ini(const char *filename) +{ + struct stat sb; + char *p, *buf; + off_t len; + int fd, ret; + + fd = open(filename, O_RDONLY); + if (fd < 0) { + log_err("fio: job file open: %s\n", strerror(errno)); + return 1; + } + + if (fstat(fd, &sb) < 0) { + log_err("fio: job file stat: %s\n", strerror(errno)); + return 1; + } + + buf = malloc(sb.st_size); + + len = sb.st_size; + p = buf; + do { + ret = read(fd, p, len); + if (ret > 0) { + len -= ret; + if (!len) + break; + p += ret; + continue; + } else if (!ret) + break; + else if (errno == EAGAIN || errno == EINTR) + continue; + } while (1); + + ret = send_file_buf(buf, sb.st_size); + free(buf); + return ret; +} diff --git a/init.c b/init.c index ce164507..088e814e 100644 --- a/init.c +++ b/init.c @@ -46,6 +46,8 @@ char *exec_profile = NULL; int warnings_fatal = 0; int terse_version = 2; int is_backend = 0; +int is_client = 0; +char *client; int write_bw_log = 0; int read_only = 0; @@ -160,6 +162,16 @@ static struct option l_opts[FIO_NR_OPTIONS] = { .has_arg = no_argument, .val = 'S', }, + { + .name = (char *) "net-port", + .has_arg = required_argument, + .val = 'P', + }, + { + .name = (char *) "client", + .has_arg = required_argument, + .val = 'C', + }, { .name = NULL, }, @@ -1079,6 +1091,9 @@ static void usage(const char *name) " (def 1024)\n"); printf("\t--warnings-fatal Fio parser warnings are fatal\n"); printf("\t--max-jobs\tMaximum number of threads/processes to support\n"); + printf("\t--server\tStart a backend fio server\n"); + printf("\t--client=hostname Talk to remove backend fio server at hostname\n"); + printf("\t--net-port=port\tUse specified port for client/server connection\n"); printf("\nFio was written by Jens Axboe "); printf("\n Jens Axboe \n"); } @@ -1304,8 +1319,27 @@ static int parse_cmd_line(int argc, char *argv[]) } break; case 'S': + if (is_client) { + log_err("fio: can't be both client and server\n"); + do_exit++; + exit_val = 1; + break; + } is_backend = 1; break; + case 'P': + fio_net_port = atoi(optarg); + break; + case 'C': + if (is_backend) { + log_err("fio: can't be both client and server\n"); + do_exit++; + exit_val = 1; + break; + } + is_client = 1; + client = strdup(optarg); + break; default: do_exit++; exit_val = 1; @@ -1316,6 +1350,12 @@ static int parse_cmd_line(int argc, char *argv[]) if (do_exit) exit(exit_val); + if (is_client && fio_client_connect(client)) { + do_exit++; + exit_val = 1; + return 1; + } + if (is_backend) return fio_server(); @@ -1354,8 +1394,13 @@ int parse_options(int argc, char *argv[]) for (i = 0; i < job_files; i++) { if (fill_def_thread()) return 1; - if (parse_jobs_ini(ini_file[i], 0, i)) - return 1; + if (is_client) { + if (fio_client_send_ini(ini_file[i])) + return 1; + } else { + if (parse_jobs_ini(ini_file[i], 0, i)) + return 1; + } free(ini_file[i]); } diff --git a/server.c b/server.c index 5ef0fe01..f393873f 100644 --- a/server.c +++ b/server.c @@ -2,33 +2,275 @@ #include #include #include -#include -#include #include #include #include -#include #include #include -#include #include -#include #include #include #include #include "fio.h" +#include "server.h" +#include "crc/crc32.h" -static int net_port = 8765; +int fio_net_port = 8765; int exit_backend = 0; +static char *job_buf; +static unsigned int job_cur_len; +static unsigned int job_max_len; + +int fio_send_data(int sk, const void *p, unsigned int len) +{ + do { + int ret = send(sk, p, len, 0); + + if (ret > 0) { + len -= ret; + if (!len) + break; + p += ret; + continue; + } else if (!ret) + break; + else if (errno == EAGAIN || errno == EINTR) + continue; + } while (!exit_backend); + + if (!len) + return 0; + + return 1; +} + +int fio_recv_data(int sk, void *p, unsigned int len) +{ + do { + int ret = recv(sk, p, len, MSG_WAITALL); + + if (ret > 0) { + len -= ret; + if (!len) + break; + p += ret; + continue; + } else if (!ret) + break; + else if (errno == EAGAIN || errno == EINTR) + continue; + } while (!exit_backend); + + if (!len) + return 0; + + return -1; +} + +static int verify_convert_cmd(struct fio_net_cmd *cmd) +{ + uint32_t crc; + + cmd->cmd_crc32 = le32_to_cpu(cmd->cmd_crc32); + cmd->pdu_crc32 = le32_to_cpu(cmd->pdu_crc32); + + crc = crc32(cmd, sizeof(*cmd) - 2 * sizeof(uint32_t)); + if (crc != cmd->cmd_crc32) { + log_err("fio: server bad crc on command (got %x, wanted %x)\n", + cmd->cmd_crc32, crc); + return 1; + } + + cmd->version = le16_to_cpu(cmd->version); + cmd->opcode = le16_to_cpu(cmd->opcode); + cmd->flags = le32_to_cpu(cmd->flags); + cmd->serial = le64_to_cpu(cmd->serial); + cmd->pdu_len = le32_to_cpu(cmd->pdu_len); + + switch (cmd->version) { + case FIO_SERVER_VER1: + break; + default: + log_err("fio: bad server cmd version %d\n", cmd->version); + return 1; + } + + if (cmd->pdu_len > FIO_SERVER_MAX_PDU) { + log_err("fio: command payload too large: %u\n", cmd->pdu_len); + return 1; + } + + return 0; +} + +static struct fio_net_cmd *read_command(int sk) +{ + struct fio_net_cmd cmd, *ret = NULL; + uint32_t crc; + + if (fio_recv_data(sk, &cmd, sizeof(cmd))) + return NULL; + + /* We have a command, verify it and swap if need be */ + if (verify_convert_cmd(&cmd)) + return NULL; + + /* Command checks out, alloc real command and fill in */ + ret = malloc(sizeof(cmd) + cmd.pdu_len); + memcpy(ret, &cmd, sizeof(cmd)); + + if (!ret->pdu_len) + return ret; + + /* There's payload, get it */ + if (fio_recv_data(sk, (void *) ret + sizeof(*ret), ret->pdu_len)) { + free(ret); + return NULL; + } + + /* Verify payload crc */ + crc = crc32(ret->payload, ret->pdu_len); + if (crc != ret->pdu_crc32) { + log_err("fio: server bad crc on payload (got %x, wanted %x)\n", + ret->pdu_crc32, crc); + free(ret); + return NULL; + } + + + return ret; +} + +void fio_net_cmd_crc(struct fio_net_cmd *cmd) +{ + uint32_t pdu_len; + + cmd->cmd_crc32 = cpu_to_le32(crc32(cmd, + sizeof(*cmd) - 2 * sizeof(uint32_t))); + + pdu_len = le32_to_cpu(cmd->pdu_len); + if (pdu_len) + cmd->pdu_crc32 = cpu_to_le32(crc32(cmd->payload, pdu_len)); +} + +static int send_simple_command(int sk, uint16_t opcode, uint64_t serial) +{ + struct fio_net_cmd cmd = { + .version = cpu_to_le16(FIO_SERVER_VER1), + .opcode = cpu_to_le16(opcode), + .serial = cpu_to_le16(serial), + }; + + fio_net_cmd_crc(&cmd); + + return fio_send_data(sk, &cmd, sizeof(cmd)); +} + +/* + * Send an ack for this command + */ +static int ack_command(int sk, struct fio_net_cmd *cmd) +{ + return send_simple_command(sk, FIO_NET_CMD_ACK, cmd->serial); +} + +#if 0 +static int nak_command(int sk, struct fio_net_cmd *cmd) +{ + return send_simple_command(sk, FIO_NET_CMD_NAK, cmd->serial); +} +#endif + +static int handle_cur_job(struct fio_net_cmd *cmd, int done) +{ + unsigned int left = job_max_len - job_cur_len; + int ret = 0; + + if (left < cmd->pdu_len) { + job_buf = realloc(job_buf, job_max_len + 2 * cmd->pdu_len); + job_max_len += 2 * cmd->pdu_len; + } + + memcpy(job_buf + job_cur_len, cmd->payload, cmd->pdu_len); + job_cur_len += cmd->pdu_len; + + if (done) { + parse_jobs_ini(job_buf, 1, 0); + ret = exec_run(); + reset_fio_state(); + free(job_buf); + job_buf = NULL; + job_cur_len = job_max_len = 0; + } + + return ret; +} + +static int handle_command(struct fio_net_cmd *cmd) +{ + int ret; + + switch (cmd->opcode) { + case FIO_NET_CMD_QUIT: + exit_backend = 1; + return 1; + case FIO_NET_CMD_ACK: + return 0; + case FIO_NET_CMD_NAK: + return 1; + case FIO_NET_CMD_JOB: + ret = handle_cur_job(cmd, 0); + break; + case FIO_NET_CMD_JOB_END: + ret = handle_cur_job(cmd, 1); + break; + default: + log_err("fio: unknown opcode: %d\n", cmd->opcode); + ret = 1; + } + + return ret; +} + +static int handle_connection(int sk) +{ + struct fio_net_cmd *cmd = NULL; + int ret = 0; + + /* read forever */ + while (!exit_backend) { + cmd = read_command(sk); + if (!cmd) { + ret = 1; + break; + } + + ret = ack_command(sk, cmd); + if (ret) + break; + + ret = handle_command(cmd); + if (ret) + break; + + free(cmd); + } + + if (cmd) + free(cmd); + + return ret; +} + static int accept_loop(int listen_sk) { struct sockaddr addr; unsigned int len = sizeof(addr); struct pollfd pfd; - int ret, sk, flags; + int ret, sk, flags, exitval = 0; flags = fcntl(listen_sk, F_GETFL); flags |= O_NONBLOCK; @@ -59,26 +301,7 @@ again: return -1; } - /* read forever */ - while (!exit_backend) { - char buf[131072]; - - ret = recv(sk, buf, 4096, 0); - if (ret > 0) { - if (!strncmp("FIO_QUIT", buf, 8)) { - exit_backend = 1; - break; - } - parse_jobs_ini(buf, 1, 0); - exec_run(); - reset_fio_state(); - break; - } else if (!ret) - break; - if (errno == EAGAIN || errno == EINTR) - continue; - break; - } + exitval = handle_connection(sk); close(sk); @@ -86,7 +309,7 @@ again: goto again; out: - return 0; + return exitval; } int fio_server(void) @@ -116,7 +339,7 @@ int fio_server(void) saddr_in.sin_family = AF_INET; saddr_in.sin_addr.s_addr = htonl(INADDR_ANY); - saddr_in.sin_port = htons(net_port); + saddr_in.sin_port = htons(fio_net_port); if (bind(sk, (struct sockaddr *) &saddr_in, sizeof(saddr_in)) < 0) { log_err("fio: bind: %s\n", strerror(errno)); diff --git a/server.h b/server.h index aed61b2c..51a84176 100644 --- a/server.h +++ b/server.h @@ -1,7 +1,71 @@ #ifndef FIO_SERVER_H #define FIO_SERVER_H +#include +#include + +/* + * On-wire encoding is little endian + */ +struct fio_net_cmd { + uint16_t version; /* protocol version */ + uint16_t opcode; /* command opcode */ + uint32_t flags; /* modifier flags */ + uint64_t serial; /* serial number */ + uint32_t pad; + uint32_t pdu_len; /* length of post-cmd layload */ + uint32_t cmd_crc32; /* cmd checksum */ + uint32_t pdu_crc32; /* payload checksum */ + uint8_t payload[0]; /* payload */ +}; + +enum { + FIO_SERVER_VER = 1, + FIO_SERVER_VER1 = 1, + + FIO_SERVER_MAX_PDU = 4096, + + FIO_NET_CMD_QUIT = 1, + FIO_NET_CMD_JOB = 2, + FIO_NET_CMD_JOB_END = 3, + FIO_NET_CMD_ACK = 4, + FIO_NET_CMD_NAK = 5, +}; + extern int fio_server(void); + +extern int fio_client_connect(const char *); +extern int fio_client_send_ini(const char *); + +extern int fio_recv_data(int sk, void *p, unsigned int len); +extern int fio_send_data(int sk, const void *p, unsigned int len); +extern void fio_net_cmd_crc(struct fio_net_cmd *); + extern int exit_backend; +extern int fio_net_port; + +#if __BYTE_ORDER == __LITTLE_ENDIAN +#define le16_to_cpu(x) (x) +#define le32_to_cpu(x) (x) +#define le64_to_cpu(x) (x) +#define cpu_to_le16(x) (x) +#define cpu_to_le32(x) (x) +#define cpu_to_le64(x) (x) +#elif __BYTE_ORDER == __BIG_ENDIAN +#define le16_to_cpu(x) __bswap_16(x) +#define le32_to_cpu(x) __bswap_32(x) +#define le64_to_cpu(x) __bswap_64(x) +#define cpu_to_le16(x) __bswap_16(x) +#define cpu_to_le32(x) __bswap_32(x) +#define cpu_to_le64(x) __bswap_64(x) +#else +#error "Endianness not detected" +#endif + +static inline void fio_init_net_cmd(struct fio_net_cmd *cmd) +{ + memset(cmd, 0, sizeof(*cmd)); + cmd->version = cpu_to_le16(FIO_SERVER_VER1); +} #endif -- 2.25.1