server: make client connections fork off
Instead of handling connections inline from the server, fork
them off before running them. This makes the able to handle
the idle loop command send/reply directly, without being
invoked from the idle loop.
New connections are forked off automatically as well, leaving
the server free to accept new connections. Not sure we want that,
or at least it should be an option to return BUSY on attempted
new connections.
Signed-off-by: Jens Axboe <axboe@kernel.dk>
diff --git a/backend.c b/backend.c
index 92c4648..e65af52 100644
--- a/backend.c
+++ b/backend.c
@@ -969,6 +969,9 @@
dprint(FD_PROCESS, "jobs pid=%d started\n", (int) td->pid);
+ if (is_backend)
+ fio_server_send_start(td);
+
INIT_FLIST_HEAD(&td->io_u_freelist);
INIT_FLIST_HEAD(&td->io_u_busylist);
INIT_FLIST_HEAD(&td->io_u_requeues);
@@ -1555,21 +1558,13 @@
reap_threads(&nr_running, &t_rate, &m_rate);
- if (todo) {
- if (is_backend)
- fio_server_idle_loop();
- else
- usleep(100000);
- }
+ if (todo)
+ usleep(100000);
}
while (nr_running) {
reap_threads(&nr_running, &t_rate, &m_rate);
-
- if (is_backend)
- fio_server_idle_loop();
- else
- usleep(10000);
+ usleep(10000);
}
update_io_ticks();
diff --git a/client.c b/client.c
index 6230a66..b553b05 100644
--- a/client.c
+++ b/client.c
@@ -373,7 +373,7 @@
void fio_client_terminate(struct fio_client *client)
{
- fio_net_send_simple_cmd(client->fd, FIO_NET_CMD_QUIT, 0, NULL);
+ fio_net_send_quit(client->fd);
}
void fio_clients_terminate(void)
@@ -1118,7 +1118,8 @@
convert_stop(cmd);
client->state = Client_stopped;
- client->error = pdu->error;
+ client->error = le32_to_cpu(pdu->error);
+ client->signal = le32_to_cpu(pdu->signal);
ops->stop(client, cmd);
free(cmd);
break;
diff --git a/client.h b/client.h
index 3022816..fd60b46 100644
--- a/client.h
+++ b/client.h
@@ -43,6 +43,7 @@
int disk_stats_shown;
unsigned int jobs;
int error;
+ int signal;
int ipv6;
int sent_job;
uint32_t type;
diff --git a/server.c b/server.c
index 899b230..d5733c4 100644
--- a/server.c
+++ b/server.c
@@ -34,9 +34,22 @@
static char *bind_sock;
static struct sockaddr_in saddr_in;
static struct sockaddr_in6 saddr_in6;
-static int first_cmd_check;
static int use_ipv6;
+struct fio_fork_item {
+ struct flist_head list;
+ int exitval;
+ int signal;
+ int exited;
+ pid_t pid;
+};
+
+/* Created on fork on new connection */
+static FLIST_HEAD(conn_list);
+
+/* Created on job fork from connection */
+static FLIST_HEAD(job_list);
+
static const char *fio_server_ops[FIO_NET_CMD_NR] = {
"",
"QUIT",
@@ -375,26 +388,125 @@
return 0;
}
-static int fio_server_send_quit_cmd(void)
+int fio_net_send_quit(int sk)
{
dprint(FD_NET, "server: sending quit\n");
+
return fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_QUIT, 0, NULL);
}
+int fio_net_send_stop(int sk, int error, int signal)
+{
+ struct cmd_end_pdu epdu;
+
+ dprint(FD_NET, "server: sending stop (%d, %d)\n", error, signal);
+
+ epdu.error = __cpu_to_le32(error);
+ epdu.signal = __cpu_to_le32(signal);
+ return fio_net_send_cmd(server_fd, FIO_NET_CMD_STOP, &epdu, sizeof(epdu), 0);
+}
+
+static void fio_server_add_fork_item(pid_t pid, struct flist_head *list)
+{
+ struct fio_fork_item *ffi;
+
+ ffi = malloc(sizeof(*ffi));
+ ffi->exitval = 0;
+ ffi->signal = 0;
+ ffi->exited = 0;
+ ffi->pid = pid;
+ flist_add_tail(&ffi->list, list);
+}
+
+static void fio_server_add_conn_pid(pid_t pid)
+{
+ dprint(FD_NET, "server: forked off connection job (pid=%u)\n", pid);
+ fio_server_add_fork_item(pid, &conn_list);
+}
+
+static void fio_server_add_job_pid(pid_t pid)
+{
+ dprint(FD_NET, "server: forked off job job (pid=%u)\n", pid);
+ fio_server_add_fork_item(pid, &job_list);
+}
+
+static void fio_server_check_fork_item(struct fio_fork_item *ffi)
+{
+ int ret, status;
+
+ ret = waitpid(ffi->pid, &status, WNOHANG);
+ if (ret < 0) {
+ if (errno == ECHILD) {
+ log_err("fio: connection pid %u disappeared\n", ffi->pid);
+ ffi->exited = 1;
+ } else
+ log_err("fio: waitpid: %s\n", strerror(errno));
+ } else if (ret == ffi->pid) {
+ if (WIFSIGNALED(status)) {
+ ffi->signal = WTERMSIG(status);
+ ffi->exited = 1;
+ }
+ if (WIFEXITED(status)) {
+ if (WEXITSTATUS(status))
+ ffi->exitval = WEXITSTATUS(status);
+ ffi->exited = 1;
+ }
+ }
+}
+
+static void fio_server_fork_item_done(struct fio_fork_item *ffi)
+{
+ dprint(FD_NET, "pid %u exited, sig=%u, exitval=%d\n", ffi->pid, ffi->signal, ffi->exitval);
+
+ /*
+ * Fold STOP and QUIT...
+ */
+ fio_net_send_stop(server_fd, ffi->exitval, ffi->signal);
+ fio_net_send_quit(server_fd);
+ flist_del(&ffi->list);
+ free(ffi);
+}
+
+static void fio_server_check_fork_items(struct flist_head *list, int bla)
+{
+ struct flist_head *entry, *tmp;
+ struct fio_fork_item *ffi;
+
+ flist_for_each_safe(entry, tmp, list) {
+ ffi = flist_entry(entry, struct fio_fork_item, list);
+
+ fio_server_check_fork_item(ffi);
+
+ if (ffi->exited)
+ fio_server_fork_item_done(ffi);
+ }
+}
+
+static void fio_server_check_jobs(void)
+{
+ fio_server_check_fork_items(&job_list, 0);
+}
+
+static void fio_server_check_conns(void)
+{
+ fio_server_check_fork_items(&conn_list, 1);
+}
+
static int handle_run_cmd(struct fio_net_cmd *cmd)
{
- struct cmd_end_pdu epdu;
+ pid_t pid;
int ret;
+ set_genesis_time();
+
+ pid = fork();
+ if (pid) {
+ fio_server_add_job_pid(pid);
+ return 0;
+ }
+
ret = fio_backend();
-
- epdu.error = ret;
- fio_net_send_cmd(server_fd, FIO_NET_CMD_STOP, &epdu, sizeof(epdu), 0);
-
- fio_server_send_quit_cmd();
- reset_fio_state();
- first_cmd_check = 0;
- return ret;
+ _exit(ret);
}
static int handle_job_cmd(struct fio_net_cmd *cmd)
@@ -407,7 +519,7 @@
pdu->client_type = le32_to_cpu(pdu->client_type);
if (parse_jobs_ini(buf, 1, 0, pdu->client_type)) {
- fio_server_send_quit_cmd();
+ fio_net_send_quit(server_fd);
return -1;
}
@@ -443,7 +555,7 @@
}
if (parse_cmd_line(clp->lines, argv, clp->client_type)) {
- fio_server_send_quit_cmd();
+ fio_net_send_quit(server_fd);
free(argv);
return -1;
}
@@ -558,11 +670,15 @@
return ret;
}
-static int handle_connection(int sk, int block)
+static int handle_connection(int sk)
{
struct fio_net_cmd *cmd = NULL;
int ret = 0;
+ reset_fio_state();
+ INIT_FLIST_HEAD(&job_list);
+ server_fd = sk;
+
/* read forever */
while (!exit_backend) {
struct pollfd pfd = {
@@ -579,8 +695,7 @@
log_err("fio: poll: %s\n", strerror(errno));
break;
} else if (!ret) {
- if (!block)
- return 0;
+ fio_server_check_jobs();
continue;
}
@@ -592,6 +707,8 @@
}
} while (!exit_backend);
+ fio_server_check_jobs();
+
if (ret < 0)
break;
@@ -612,17 +729,8 @@
if (cmd)
free(cmd);
- return ret;
-}
-
-void fio_server_idle_loop(void)
-{
- if (!first_cmd_check) {
- fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_SERVER_START, 0, NULL);
- first_cmd_check = 1;
- }
- if (server_fd != -1)
- handle_connection(server_fd, 0);
+ close(sk);
+ _exit(ret);
}
static int accept_loop(int listen_sk)
@@ -630,52 +738,59 @@
struct sockaddr_in addr;
fio_socklen_t len = sizeof(addr);
struct pollfd pfd;
- int ret, sk, flags, exitval = 0;
+ int ret = 0, sk, flags, exitval = 0;
dprint(FD_NET, "server enter accept loop\n");
flags = fcntl(listen_sk, F_GETFL);
flags |= O_NONBLOCK;
fcntl(listen_sk, F_SETFL, flags);
-again:
- pfd.fd = listen_sk;
- pfd.events = POLLIN;
- do {
- ret = poll(&pfd, 1, 100);
- if (ret < 0) {
- if (errno == EINTR)
+
+ while (!exit_backend) {
+ pid_t pid;
+
+ pfd.fd = listen_sk;
+ pfd.events = POLLIN;
+ do {
+ ret = poll(&pfd, 1, 100);
+ if (ret < 0) {
+ if (errno == EINTR)
+ break;
+ log_err("fio: poll: %s\n", strerror(errno));
break;
- log_err("fio: poll: %s\n", strerror(errno));
- goto out;
- } else if (!ret)
- continue;
+ } else if (!ret) {
+ fio_server_check_conns();
+ continue;
+ }
- if (pfd.revents & POLLIN)
+ if (pfd.revents & POLLIN)
+ break;
+ } while (!exit_backend);
+
+ fio_server_check_conns();
+
+ if (exit_backend || ret < 0)
break;
- } while (!exit_backend);
- if (exit_backend)
- goto out;
+ sk = accept(listen_sk, (struct sockaddr *) &addr, &len);
+ if (sk < 0) {
+ log_err("fio: accept: %s\n", strerror(errno));
+ return -1;
+ }
- sk = accept(listen_sk, (struct sockaddr *) &addr, &len);
- if (sk < 0) {
- log_err("fio: accept: %s\n", strerror(errno));
- return -1;
+ dprint(FD_NET, "server: connect from %s\n", inet_ntoa(addr.sin_addr));
+
+ pid = fork();
+ if (pid) {
+ close(sk);
+ fio_server_add_conn_pid(pid);
+ continue;
+ }
+
+ /* exits */
+ handle_connection(sk);
}
- dprint(FD_NET, "server: connect from %s\n", inet_ntoa(addr.sin_addr));
-
- server_fd = sk;
-
- exitval = handle_connection(sk, 1);
-
- server_fd = -1;
- close(sk);
-
- if (!exit_backend)
- goto again;
-
-out:
return exitval;
}
@@ -993,6 +1108,13 @@
fio_net_send_cmd(server_fd, FIO_NET_CMD_ADD_JOB, &pdu, sizeof(pdu), 0);
}
+void fio_server_send_start(struct thread_data *td)
+{
+ assert(server_fd != -1);
+
+ fio_net_send_simple_cmd(server_fd, FIO_NET_CMD_SERVER_START, 0, NULL);
+}
+
static int fio_init_server_ip(void)
{
struct sockaddr *addr;
diff --git a/server.h b/server.h
index 3c66ecb..422bfbf 100644
--- a/server.h
+++ b/server.h
@@ -38,7 +38,7 @@
};
enum {
- FIO_SERVER_VER = 13,
+ FIO_SERVER_VER = 14,
FIO_SERVER_MAX_FRAGMENT_PDU = 1024,
@@ -116,6 +116,7 @@
struct cmd_end_pdu {
uint32_t error;
+ uint32_t signal;
};
struct cmd_add_job_pdu {
@@ -155,7 +156,6 @@
extern void fio_server_send_ts(struct thread_stat *, struct group_run_stats *);
extern void fio_server_send_gs(struct group_run_stats *);
extern void fio_server_send_du(void);
-extern void fio_server_idle_loop(void);
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);
@@ -165,6 +165,9 @@
extern int fio_send_iolog(struct thread_data *, struct io_log *, const char *);
extern void fio_server_send_add_job(struct thread_data *);
+extern void fio_server_send_start(struct thread_data *);
+extern int fio_net_send_stop(int sk, int error, int signal);
+extern int fio_net_send_quit(int sk);
extern int exit_backend;
extern int fio_net_port;