Fix socket error handling.

1. Grab the socket lock for the whole conversation.

We need to keep the lock until the whole conversation is over.
Otherwise we may have concurrency issues where communication gets intermixed
and errors in one thread can cause a thread to hang.

Here is what we observed:

thread 1: grabs lock and send a request.
thread 2: grabs lock and sends a request
server: thread 2 request causes a fatal error and the server close the
connection
thread 2: grabs the lock and waits for a reply.
thread 2: gets the error and returns to caller with it (connection is closed).
thread 1: grabs the lock (which reopens the closed channel) and reads ...
... forever as the server has already killed all the previous state.

2. Fail immediately on short reads for the initial 4 byte length header.

If the first 4 bytes do not come at once don't bother retrying. In 99.9% of the
cases what we are witnessing here is a fatal error from the proxy that closed
the socket. Reopening the scket cannot accomplish anything as the request sent
down the channel is tied to the specific socket, so once the socket is closed
there is no hope to ever get back a reply.

Signed-off-by: Simo Sorce <simo@redhat.com>
Reviewed-by: Günther Deschner <gdeschner@redhat.com>

1 files changed, 30 insertions, 44 deletions

diff --git a/proxy/src/client/gpm_common.c b/proxy/src/client/gpm_common.c
index 8ac1a2a..df1f5a1 100644
--- a/proxy/src/client/gpm_common.c
+++ b/proxy/src/client/gpm_common.c
@@ -161,6 +161,7 @@ static int gpm_release_sock(struct gpm_ctx *gpmctx)
     return pthread_mutex_unlock(&gpmctx->lock);
 }
 
+/* must be called after the lock has been grabbed */
 static int gpm_send_buffer(struct gpm_ctx *gpmctx,
                            char *buffer, uint32_t length)
 {
@@ -174,8 +175,6 @@ static int gpm_send_buffer(struct gpm_ctx *gpmctx,
         return EINVAL;
     }
 
-    gpm_grab_sock(gpmctx);
-
     size = length | FRAGMENT_BIT;
     size = htonl(size);
 
@@ -225,46 +224,29 @@ done:
         /* on errors we can only close the fd and return */
         gpm_close_socket(gpmctx);
     }
-    gpm_release_sock(gpmctx);
     return ret;
 }
 
+/* must be called after the lock has been grabbed */
 static int gpm_recv_buffer(struct gpm_ctx *gpmctx,
                            char *buffer, uint32_t *length)
 {
     uint32_t size;
     size_t rn;
     size_t pos;
-    bool retry;
     int ret;
 
-    gpm_grab_sock(gpmctx);
-
-    retry = false;
+    ret = 0;
     do {
-        ret = 0;
-        do {
-            rn = read(gpmctx->fd, &size, sizeof(uint32_t));
-            if (rn == -1) {
-                ret = errno;
-            }
-        } while (ret == EINTR);
-        if (rn != 4) {
-            /* reopen and retry once */
-            if (retry == false) {
-                gpm_close_socket(gpmctx);
-                ret = gpm_open_socket(gpmctx);
-                if (ret == 0) {
-                    retry = true;
-                    continue;
-                }
-            } else {
-                ret = EIO;
-            }
-            goto done;
+        rn = read(gpmctx->fd, &size, sizeof(uint32_t));
+        if (rn == -1) {
+            ret = errno;
         }
-        retry = false;
-    } while (retry);
+    } while (ret == EINTR);
+    if (rn != 4) {
+        ret = EIO;
+        goto done;
+    }
 
     *length = ntohl(size);
     *length &= ~FRAGMENT_BIT;
@@ -298,29 +280,22 @@ done:
         /* on errors we can only close the fd and return */
         gpm_close_socket(gpmctx);
     }
-    gpm_release_sock(gpmctx);
     return ret;
 }
 
-static int gpm_next_xid(struct gpm_ctx *gpmctx, uint32_t *xid)
+/* must be called after the lock has been grabbed */
+static uint32_t gpm_next_xid(struct gpm_ctx *gpmctx)
 {
-    int ret;
-
-    ret = gpm_grab_sock(gpmctx);
-    if (ret) {
-        goto done;
-    }
+    uint32_t xid;
 
     if (gpmctx->next_xid < 0) {
-        *xid = 0;
         gpmctx->next_xid = 1;
+        xid = 0;
     } else {
-        *xid = gpmctx->next_xid++;
+        xid = gpmctx->next_xid++;
     }
 
-done:
-    gpm_release_sock(gpmctx);
-    return ret;
+    return xid;
 }
 
 static struct gpm_ctx *gpm_get_ctx(void)
@@ -435,6 +410,7 @@ int gpm_make_call(int proc, union gp_rpc_arg *arg, union gp_rpc_res *res)
     uint32_t length;
     uint32_t xid;
     bool xdrok;
+    bool sockgrab = false;
     int ret;
 
     xdrmem_create(&xdr_call_ctx, buffer, MAX_RPC_SIZE, XDR_ENCODE);
@@ -458,11 +434,14 @@ int gpm_make_call(int proc, union gp_rpc_arg *arg, union gp_rpc_res *res)
         return EINVAL;
     }
 
-    ret = gpm_next_xid(gpmctx, &xid);
+    /* grab the lock for the whole conversation */
+    ret = gpm_grab_sock(gpmctx);
     if (ret) {
         goto done;
     }
-    msg.xid = xid;
+    sockgrab = true;
+
+    msg.xid = xid = gpm_next_xid(gpmctx);
 
     /* encode header */
     xdrok = xdr_gp_rpc_msg(&xdr_call_ctx, &msg);
@@ -490,6 +469,10 @@ int gpm_make_call(int proc, union gp_rpc_arg *arg, union gp_rpc_res *res)
         goto done;
     }
 
+    /* release the lock */
+    gpm_release_sock(gpmctx);
+    sockgrab = false;
+
     /* decode header */
     xdrok = xdr_gp_rpc_msg(&xdr_reply_ctx, &msg);
     if (!xdrok) {
@@ -512,6 +495,9 @@ int gpm_make_call(int proc, union gp_rpc_arg *arg, union gp_rpc_res *res)
     }
 
 done:
+    if (sockgrab) {
+        gpm_release_sock(gpmctx);
+    }
     xdr_free((xdrproc_t)xdr_gp_rpc_msg, (char *)&msg);
     xdr_destroy(&xdr_call_ctx);
     xdr_destroy(&xdr_reply_ctx);