rsocket: Fix hang in rrecv/rsend after disconnecting

If a user calls rrecv() after a blocking rsocket has been disconnected,
it will hang.  This problem and the cause was reported by Sirdhar Samudrala
<samudrala@us.ibm.com>.  It can be reproduced by running netserver -f -D
using the rs-preload library.  A similar issue exists with rsend().

Fix this by not blocking on a CQ unless we're connected.

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

diff --git a/src/rsocket.c b/src/rsocket.c
index 8cbd425268a207d50ae70c13ac3266a74ac3cbb6..1fcc4af4d1a859901e456d5ad1e77bf13f3def7d 100644 (file)
--- a/src/rsocket.c
+++ b/src/rsocket.c
@@ -937,6 +937,11 @@ static int rs_can_send(struct rsocket *rs)
               (rs->target_sgl[rs->target_sge].length != 0);
 }
 
+static int rs_conn_can_send(struct rsocket *rs)
+{
+       return rs_can_send(rs) || (rs->state != rs_connected);
+}
+
 static int rs_can_send_ctrl(struct rsocket *rs)
 {
        return rs->ctrl_avail;
@@ -947,6 +952,11 @@ static int rs_have_rdata(struct rsocket *rs)
        return (rs->rmsg_head != rs->rmsg_tail);
 }
 
+static int rs_conn_have_rdata(struct rsocket *rs)
+{
+       return rs_have_rdata(rs) || (rs->state != rs_connected);
+}
+
 static int rs_all_sends_done(struct rsocket *rs)
 {
        return (rs->sqe_avail + rs->ctrl_avail) == rs->sq_size;
@@ -1007,7 +1017,7 @@ ssize_t rrecv(int socket, void *buf, size_t len, int flags)
        }
        fastlock_acquire(&rs->rlock);
        if (!rs_have_rdata(rs)) {
-               ret = rs_process_cq(rs, rs_nonblocking(rs, flags), rs_have_rdata);
+               ret = rs_process_cq(rs, rs_nonblocking(rs, flags), rs_conn_have_rdata);
                if (ret && errno != ECONNRESET)
                        goto out;
        }
@@ -1111,9 +1121,14 @@ ssize_t rsend(int socket, const void *buf, size_t len, int flags)
        fastlock_acquire(&rs->slock);
        for (left = len; left; left -= xfer_size, buf += xfer_size) {
                if (!rs_can_send(rs)) {
-                       ret = rs_process_cq(rs, rs_nonblocking(rs, flags), rs_can_send);
+                       ret = rs_process_cq(rs, rs_nonblocking(rs, flags),
+                                           rs_conn_can_send);
                        if (ret)
                                break;
+                       if (rs->state != rs_connected) {
+                               ret = ERR(ECONNRESET);
+                               break;
+                       }
                }
 
                if (olen < left) {
@@ -1208,9 +1223,14 @@ static ssize_t rsendv(int socket, const struct iovec *iov, int iovcnt, int flags
        fastlock_acquire(&rs->slock);
        for (left = len; left; left -= xfer_size) {
                if (!rs_can_send(rs)) {
-                       ret = rs_process_cq(rs, rs_nonblocking(rs, flags), rs_can_send);
+                       ret = rs_process_cq(rs, rs_nonblocking(rs, flags),
+                                           rs_conn_can_send);
                        if (ret)
                                break;
+                       if (rs->state != rs_connected) {
+                               ret = ERR(ECONNRESET);
+                               break;
+                       }
                }
 
                if (olen < left) {