rstream: Set rsocket nonblocking if set to async operation

If asynchronous use is specified (use of poll/select), set the
rsocket to nonblocking.  This matches the common usage case for
asynchronous sockets.

When asynchronous support is enabled, the nonblocking/blocking
test option determines whether the poll/select call will block,
or if rstream will spin on the calls.

This provides more flexibility with how the rsocket is used.
Specifically, MPI often uses nonblocking sockets, but spins on
poll/select.  However, many apps will use nonblocking sockets,
but wait on poll/select.

Signed-off-by: Sean Hefty <sean.hefty@intel.com>

diff --git a/examples/rstream.c b/examples/rstream.c
index 8d5a22dc4874f9b3f69132fd4a3d94bad7e63df9..2b25ef56c23295476d54273f9c8fef08b783f10a 100644 (file)
--- a/examples/rstream.c
+++ b/examples/rstream.c
@@ -57,6 +57,7 @@ static int use_rs = 1;
 static int use_async;
 static int verify;
 static int flags = MSG_DONTWAIT;
+static int poll_timeout = 0;
 static int custom;
 static int iterations = 1;
 static int transfer_size = 1000;
@@ -190,6 +191,17 @@ static int verify_buf(void *buf, int size)
        return 0;
 }
 
+static int do_poll(struct pollfd *fds)
+{
+       int ret;
+
+       do {
+               ret = rs_poll(fds, 1, poll_timeout);
+       } while (!ret);
+
+       return ret == 1 ? 0 : ret;
+}
+
 static int send_xfer(int rs, int size)
 {
        struct pollfd fds;
@@ -205,8 +217,8 @@ static int send_xfer(int rs, int size)
 
        for (offset = 0; offset < size; ) {
                if (use_async) {
-                       ret = rs_poll(&fds, 1, -1);
-                       if (ret != 1)
+                       ret = do_poll(&fds);
+                       if (ret)
                                return ret;
                }
 
@@ -234,8 +246,8 @@ static int recv_xfer(int rs, int size)
 
        for (offset = 0; offset < size; ) {
                if (use_async) {
-                       ret = rs_poll(&fds, 1, -1);
-                       if (ret != 1)
+                       ret = do_poll(&fds);
+                       if (ret)
                                return ret;
                }
 
@@ -323,8 +335,8 @@ static void set_options(int rs)
                        break;
        }
 
-       rs_setsockopt(rs, IPPROTO_TCP, TCP_NODELAY,
-                     (void *) &no_delay, sizeof(no_delay));
+       val = 1;
+       rs_setsockopt(rs, IPPROTO_TCP, TCP_NODELAY, (void *) &val, sizeof(val));
 
        if (flags & MSG_DONTWAIT) {
                rs_fcntl(rs, F_SETFL, O_NONBLOCK);
@@ -377,8 +389,8 @@ static int server_connect(void)
                        fds.fd = lrs;
                        fds.events = POLLIN;
 
-                       ret = rs_poll(&fds, 1, -1);
-                       if (ret != 1) {
+                       ret = do_poll(&fds);
+                       if (ret) {
                                perror("rpoll");
                                goto close;
                        }
@@ -421,21 +433,24 @@ static int client_connect(void)
        ret = rs_connect(rs, res->ai_addr, res->ai_addrlen);
        if (ret && (errno != EINPROGRESS)) {
                perror("rconnect");
-               rs_close(rs);
-               rs = ret;
+               goto err;
        }
 
        if (errno == EINPROGRESS) {
                fds.fd = rs;
                fds.events = POLLOUT;
-               do {
-                       ret = rs_poll(&fds, 1, -1);
-               } while (!ret);
+               ret = do_poll(&fds);
+               if (ret)
+                       goto err;
        }
 
 free:
        freeaddrinfo(res);
        return rs;
+err:
+       freeaddrinfo(res);
+       rs_close(rs);
+       return ret;
 }
 
 static int run(void)
@@ -562,6 +577,9 @@ int main(int argc, char **argv)
                }
        }
 
+       if (!(flags & MSG_DONTWAIT))
+               poll_timeout = -1;
+
        ret = run();
        return ret;
 }