server: split vectored commands into fragments

[fio.git] / server.c

diff --git a/server.c b/server.c
index eccd9d37e80f9b48026c9d5435111c84c55e15ca..750717adb913ee8d41b66d5fdc96b2b913eae531 100644 (file)
--- a/server.c
+++ b/server.c
@@ -262,10 +262,17 @@ static int fio_send_data(int sk, const void *p, unsigned int len)
        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 = OS_MSG_DONTWAIT;
+
        do {
-               int ret = recv(sk, p, len, MSG_WAITALL);
+               int ret = recv(sk, p, len, flags);
 
                if (ret > 0) {
                        len -= ret;
@@ -275,9 +282,11 @@ static int fio_recv_data(int sk, void *p, unsigned int len)
                        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);
 
@@ -326,7 +335,7 @@ static int verify_convert_cmd(struct fio_net_cmd *cmd)
 /*
  * 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;
@@ -335,7 +344,7 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk)
        void *pdu = NULL;
 
        do {
-               ret = fio_recv_data(sk, &cmd, sizeof(cmd));
+               ret = fio_recv_data(sk, &cmd, sizeof(cmd), wait);
                if (ret)
                        break;
 
@@ -379,7 +388,7 @@ struct fio_net_cmd *fio_net_recv_cmd(int sk)
 
                /* 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;
 
@@ -1053,17 +1062,35 @@ static int fio_send_cmd_ext_pdu(int sk, uint16_t opcode, const void *buf,
 {
        struct fio_net_cmd cmd;
        struct iovec iov[2];
+       size_t this_len;
+       int ret;
 
        iov[0].iov_base = (void *) &cmd;
        iov[0].iov_len = sizeof(cmd);
-       iov[1].iov_base = (void *) buf;
-       iov[1].iov_len = size;
 
-       __fio_init_net_cmd(&cmd, opcode, size, tag);
-       cmd.flags = __cpu_to_le32(flags);
-       fio_net_cmd_crc_pdu(&cmd, buf);
+       do {
+               uint32_t this_flags = flags;
+
+               this_len = size;
+               if (this_len > FIO_SERVER_MAX_FRAGMENT_PDU)
+                       this_len = FIO_SERVER_MAX_FRAGMENT_PDU;
+
+               if (this_len < size)
+                       this_flags |= FIO_NET_CMD_F_MORE;
+
+               __fio_init_net_cmd(&cmd, opcode, this_len, tag);
+               cmd.flags = __cpu_to_le32(this_flags);
+               fio_net_cmd_crc_pdu(&cmd, buf);
 
-       return fio_sendv_data(sk, iov, 2);
+               iov[1].iov_base = (void *) buf;
+               iov[1].iov_len = this_len;
+
+               ret = fio_sendv_data(sk, iov, 2);
+               size -= this_len;
+               buf += this_len;
+       } while (!ret && size);
+
+       return ret;
 }
 
 static void finish_entry(struct sk_entry *entry)
@@ -1209,7 +1236,7 @@ static int handle_connection(struct sk_out *sk_out)
                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;
@@ -2186,16 +2213,25 @@ static void set_sig_handlers(void)
        sigaction(SIGINT, &act, NULL);
 }
 
-static int fio_server(void)
+void fio_server_destroy_sk_key(void)
 {
-       int sk, ret;
+       pthread_key_delete(sk_out_key);
+}
 
+int fio_server_create_sk_key(void)
+{
        if (pthread_key_create(&sk_out_key, NULL)) {
                log_err("fio: can't create sk_out backend key\n");
-               return -1;
+               return 1;
        }
 
        pthread_setspecific(sk_out_key, NULL);
+       return 0;
+}
+
+static int fio_server(void)
+{
+       int sk, ret;
 
        dprint(FD_NET, "starting server\n");