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 9ce58d593b0d80f56dbb337c40e51f13f8752bf3..c11179701833b0b18b2f5b42b9614948e3c6b2f9 100644 (file)
--- a/src/rsocket.c
+++ b/src/rsocket.c
@@ -954,6 +954,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;
@@ -964,6 +969,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;
@@ -1024,7 +1034,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;
        }
@@ -1128,9 +1138,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) {
@@ -1237,9 +1252,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) {