static void handle_stop(struct fio_client *client, struct fio_net_cmd *cmd);
static void handle_start(struct fio_client *client, struct fio_net_cmd *cmd);
+static void convert_text(struct fio_net_cmd *cmd);
+
struct client_ops fio_client_ops = {
.text = handle_text,
.disk_util = handle_du,
return 1;
}
+static void fio_drain_client_text(struct fio_client *client)
+{
+ do {
+ struct fio_net_cmd *cmd;
+
+ cmd = fio_net_recv_cmd(client->fd, false);
+ if (!cmd)
+ break;
+
+ if (cmd->opcode != FIO_NET_CMD_TEXT) {
+ free(cmd);
+ continue;
+ }
+
+ convert_text(cmd);
+ client->ops->text(client, cmd);
+ } while (1);
+}
+
static void remove_client(struct fio_client *client)
{
assert(client->refs);
dprint(FD_NET, "client: removed <%s>\n", client->hostname);
+ fio_drain_client_text(client);
+
if (!flist_empty(&client->list))
flist_del_init(&client->list);
dprint(FD_NET, "client: handle %s\n", client->hostname);
- cmd = fio_net_recv_cmd(client->fd);
+ cmd = fio_net_recv_cmd(client->fd, true);
if (!cmd)
return 0;
return fio_sendv_data(sk, &iov, 1);
}
-static int fio_recv_data(int sk, void *p, unsigned int len)
+static int fio_recv_data(int sk, void *p, unsigned int len, bool wait)
{
+ int flags;
+
+ if (wait)
+ flags = MSG_WAITALL;
+ else
+ flags = MSG_DONTWAIT;
+
do {
- int ret = recv(sk, p, len, MSG_WAITALL);
+ int ret = recv(sk, p, len, flags);
if (ret > 0) {
len -= ret;
continue;
} else if (!ret)
break;
- else if (errno == EAGAIN || errno == EINTR)
- continue;
- else
+ else if (errno == EAGAIN || errno == EINTR) {
+ if (wait)
+ continue;
+ break;
+ } else
break;
} while (!exit_backend);
/*
* Read (and defragment, if necessary) incoming commands
*/
-struct fio_net_cmd *fio_net_recv_cmd(int sk)
+struct fio_net_cmd *fio_net_recv_cmd(int sk, bool wait)
{
struct fio_net_cmd cmd, *tmp, *cmdret = NULL;
size_t cmd_size = 0, pdu_offset = 0;
void *pdu = NULL;
do {
- ret = fio_recv_data(sk, &cmd, sizeof(cmd));
+ ret = fio_recv_data(sk, &cmd, sizeof(cmd), wait);
if (ret)
break;
/* There's payload, get it */
pdu = (void *) cmdret->payload + pdu_offset;
- ret = fio_recv_data(sk, pdu, cmd.pdu_len);
+ ret = fio_recv_data(sk, pdu, cmd.pdu_len, wait);
if (ret)
break;
if (ret < 0)
break;
- cmd = fio_net_recv_cmd(sk_out->sk);
+ cmd = fio_net_recv_cmd(sk_out->sk, true);
if (!cmd) {
ret = -1;
break;
extern void fio_server_send_job_options(struct flist_head *, unsigned int);
extern int fio_server_get_verify_state(const char *, int, void **, int *);
-extern struct fio_net_cmd *fio_net_recv_cmd(int sk);
+extern struct fio_net_cmd *fio_net_recv_cmd(int sk, bool wait);
extern int fio_send_iolog(struct thread_data *, struct io_log *, const char *);
extern void fio_server_send_add_job(struct thread_data *);