First snapshot of FIO for Windows

[fio.git] / engines / net.c

diff --git a/engines/net.c b/engines/net.c
index fd01089caa8cbc6eda75ec676225447dc4293ac8..b594e0a50d27561da11b4fe64a3947e142a2ff0c 100644 (file)
--- a/engines/net.c
+++ b/engines/net.c
@@ -28,6 +28,16 @@ struct netio_data {
        struct sockaddr_in addr;
 };
 
+struct udp_close_msg {
+       uint32_t magic;
+       uint32_t cmd;
+};
+
+enum {
+       FIO_LINK_CLOSE = 0x89,
+       FIO_LINK_CLOSE_MAGIC = 0x6c696e6b,
+};
+
 /*
  * Return -1 for error and 'nr events' for a positive number
  * of events
@@ -43,7 +53,7 @@ static int poll_wait(struct thread_data *td, int fd, short events)
                ret = poll(&pfd, 1, -1);
                if (ret < 0) {
                        if (errno == EINTR)
-                               continue;
+                               break;
 
                        td_verror(td, errno, "poll");
                        return -1;
@@ -55,8 +65,6 @@ static int poll_wait(struct thread_data *td, int fd, short events)
 
        if (pfd.revents & events)
                return 1;
-       else if (td->terminate)
-               return 1;
 
        return -1;
 }
@@ -215,47 +223,87 @@ static int fio_netio_splice_out(struct thread_data *td, struct io_u *io_u)
 static int fio_netio_send(struct thread_data *td, struct io_u *io_u)
 {
        struct netio_data *nd = td->io_ops->data;
-       int ret, flags = 0;
+       int ret, flags = OS_MSG_DONTWAIT;
 
-       ret = poll_wait(td, io_u->file->fd, POLLOUT);
-       if (ret <= 0)
-               return ret;
+       do {
+               if (nd->net_protocol == IPPROTO_UDP) {
+                       struct sockaddr *to = (struct sockaddr *) &nd->addr;
 
-       /*
-        * if we are going to write more, set MSG_MORE
-        */
+                       ret = sendto(io_u->file->fd, io_u->xfer_buf,
+                                       io_u->xfer_buflen, flags, to,
+                                       sizeof(*to));
+               } else {
+                       /*
+                        * if we are going to write more, set MSG_MORE
+                        */
 #ifdef MSG_MORE
-       if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen < td->o.size)
-               flags = MSG_MORE;
+                       if (td->this_io_bytes[DDIR_WRITE] + io_u->xfer_buflen <
+                           td->o.size)
+                               flags |= MSG_MORE;
 #endif
+                       ret = send(io_u->file->fd, io_u->xfer_buf,
+                                       io_u->xfer_buflen, flags);
+               }
+               if (ret > 0)
+                       break;
 
-       if (nd->net_protocol == IPPROTO_UDP) {
-               return sendto(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen,
-                               flags, &nd->addr, sizeof(nd->addr));
-       } else {
-               return send(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen,
-                               flags);
-       }
+               ret = poll_wait(td, io_u->file->fd, POLLOUT);
+               if (ret <= 0)
+                       break;
+
+               flags &= ~OS_MSG_DONTWAIT;
+       } while (1);
+
+       return ret;
+}
+
+static int is_udp_close(struct io_u *io_u, int len)
+{
+       struct udp_close_msg *msg;
+
+       if (len != sizeof(struct udp_close_msg))
+               return 0;
+
+       msg = io_u->xfer_buf;
+       if (ntohl(msg->magic) != FIO_LINK_CLOSE_MAGIC)
+               return 0;
+       if (ntohl(msg->cmd) != FIO_LINK_CLOSE)
+               return 0;
+
+       return 1;
 }
 
 static int fio_netio_recv(struct thread_data *td, struct io_u *io_u)
 {
        struct netio_data *nd = td->io_ops->data;
-       int ret, flags = MSG_WAITALL;
-
-       ret = poll_wait(td, io_u->file->fd, POLLIN);
-       if (ret <= 0)
-               return ret;
+       int ret, flags = OS_MSG_DONTWAIT;
+
+       do {
+               if (nd->net_protocol == IPPROTO_UDP) {
+                       socklen_t len = sizeof(nd->addr);
+                       struct sockaddr *from = (struct sockaddr *) &nd->addr;
+
+                       ret = recvfrom(io_u->file->fd, io_u->xfer_buf,
+                                       io_u->xfer_buflen, flags, from, &len);
+                       if (is_udp_close(io_u, ret)) {
+                               td->done = 1;
+                               return 0;
+                       }
+               } else {
+                       ret = recv(io_u->file->fd, io_u->xfer_buf,
+                                       io_u->xfer_buflen, flags);
+               }
+               if (ret > 0)
+                       break;
 
-       if (nd->net_protocol == IPPROTO_UDP) {
-               socklen_t len = sizeof(nd->addr);
+               ret = poll_wait(td, io_u->file->fd, POLLIN);
+               if (ret <= 0)
+                       break;
+               flags &= ~OS_MSG_DONTWAIT;
+               flags |= MSG_WAITALL;
+       } while (1);
 
-               return recvfrom(io_u->file->fd, io_u->xfer_buf,
-                               io_u->xfer_buflen, 0, &nd->addr, &len);
-       } else {
-               return recv(io_u->file->fd, io_u->xfer_buf, io_u->xfer_buflen,
-                               flags);
-       }
+       return ret;
 }
 
 static int fio_netio_queue(struct thread_data *td, struct io_u *io_u)
@@ -358,6 +406,36 @@ static int fio_netio_open_file(struct thread_data *td, struct fio_file *f)
                return fio_netio_connect(td, f);
 }
 
+static void fio_netio_udp_close(struct thread_data *td, struct fio_file *f)
+{
+       struct netio_data *nd = td->io_ops->data;
+       struct udp_close_msg msg;
+       struct sockaddr *to = (struct sockaddr *) &nd->addr;
+       int ret;
+
+       msg.magic = htonl(FIO_LINK_CLOSE_MAGIC);
+       msg.cmd = htonl(FIO_LINK_CLOSE);
+
+       ret = sendto(f->fd, &msg, sizeof(msg), MSG_WAITALL, to,
+                       sizeof(nd->addr));
+       if (ret < 0)
+               td_verror(td, errno, "sendto udp link close");
+}
+
+static int fio_netio_close_file(struct thread_data *td, struct fio_file *f)
+{
+       struct netio_data *nd = td->io_ops->data;
+
+       /*
+        * If this is an UDP connection, notify the receiver that we are
+        * closing down the link
+        */
+       if (nd->net_protocol == IPPROTO_UDP)
+               fio_netio_udp_close(td, f);
+
+       return generic_close_file(td, f);
+}
+
 static int fio_netio_setup_connect(struct thread_data *td, const char *host,
                                   unsigned short port)
 {
@@ -555,7 +633,7 @@ static struct ioengine_ops ioengine_splice = {
        .open_file      = fio_netio_open_file,
        .close_file     = generic_close_file,
        .flags          = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
-                         FIO_SIGQUIT,
+                         FIO_SIGTERM | FIO_PIPEIO,
 };
 #endif
 
@@ -568,9 +646,9 @@ static struct ioengine_ops ioengine_rw = {
        .init           = fio_netio_init,
        .cleanup        = fio_netio_cleanup,
        .open_file      = fio_netio_open_file,
-       .close_file     = generic_close_file,
+       .close_file     = fio_netio_close_file,
        .flags          = FIO_SYNCIO | FIO_DISKLESSIO | FIO_UNIDIR |
-                         FIO_SIGQUIT,
+                         FIO_SIGTERM | FIO_PIPEIO,
 };
 
 static void fio_init fio_netio_register(void)