rpc: Cleanup rpc object in TRANSPORT_CLEANUP event

        Backport of http://review.gluster.org/5107 (upstream)

This is to ensure that unref of rpc_clnt object doesn't race with the
unref of the corresponding rpc_transport object.

rpc_transport has ref_count 2, in normal scheme of things. One held by
the socket layer and the other held by rpc layer. This inequality in
ref_count between rpc_clnt and rpc_transport could lead to concurrent
destruction of the objects and possibly lead to a crash.  To avoid this,
we defer the clean up of rpc_clnt obj to TRANSPORT_CLEANUP event. ie,
once rpc_transport's ref_count goes to zero.

Introduced rpc_clnt_disabled, to allow higher layers to differentiate
between the 'final'[1] disconnect, triggered from upper layers, and disconnect
seen as a consequence of transport disconnect. This differentiation
helps in cleaning up resources, at higher layers, in a race-free manner.

[1] - 'final' here means that the rpc and the associated connection, is not
to be used anymore. eg - glusterd_brick_disconnect on volume-stop.

Change-Id: I2ecf891a36e3b02cd9eacca964e659525d1bbc6e
BUG: 962619
Signed-off-by: Krishnan Parthasarathi <kparthas@redhat.com>
Reviewed-on: http://review.gluster.org/5213
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

2 files changed, 42 insertions, 11 deletions

diff --git a/rpc/rpc-lib/src/rpc-clnt.c b/rpc/rpc-lib/src/rpc-clnt.c
index e6c681df8..73ae8a622 100644
--- a/rpc/rpc-lib/src/rpc-clnt.c
+++ b/rpc/rpc-lib/src/rpc-clnt.c
@@ -819,6 +819,9 @@ out:
         return;
 }
 
+static void
+rpc_clnt_destroy (struct rpc_clnt *rpc);
+
 int
 rpc_clnt_notify (rpc_transport_t *trans, void *mydata,
                  rpc_transport_event_t event, void *data, ...)
@@ -864,9 +867,7 @@ rpc_clnt_notify (rpc_transport_t *trans, void *mydata,
         }
 
         case RPC_TRANSPORT_CLEANUP:
-                /* this event should not be received on a client for, a
-                 * transport is only disconnected, but never destroyed.
-                 */
+                rpc_clnt_destroy (clnt);
                 ret = 0;
                 break;
 
@@ -1541,18 +1542,22 @@ rpc_clnt_ref (struct rpc_clnt *rpc)
 
 
 static void
-rpc_clnt_destroy (struct rpc_clnt *rpc)
+rpc_clnt_trigger_destroy (struct rpc_clnt *rpc)
 {
         if (!rpc)
                 return;
 
-        if (rpc->conn.trans) {
-                rpc_transport_unregister_notify (rpc->conn.trans);
-                rpc_transport_disconnect (rpc->conn.trans);
-                rpc_transport_unref (rpc->conn.trans);
-        }
+        rpc_clnt_disable (rpc);
+        rpc_transport_unref (rpc->conn.trans);
+}
 
-        rpc_clnt_reconnect_cleanup (&rpc->conn);
+static void
+rpc_clnt_destroy (struct rpc_clnt *rpc)
+{
+        if (!rpc)
+                return;
+
+        rpc_transport_unregister_notify (rpc->conn.trans);
         saved_frames_destroy (rpc->conn.saved_frames);
         pthread_mutex_destroy (&rpc->lock);
         pthread_mutex_destroy (&rpc->conn.lock);
@@ -1579,13 +1584,36 @@ rpc_clnt_unref (struct rpc_clnt *rpc)
         }
         pthread_mutex_unlock (&rpc->lock);
         if (!count) {
-                rpc_clnt_destroy (rpc);
+                rpc_clnt_trigger_destroy (rpc);
                 return NULL;
         }
         return rpc;
 }
 
 
+char
+rpc_clnt_is_disabled (struct rpc_clnt *rpc)
+{
+
+        rpc_clnt_connection_t *conn = NULL;
+        char                   disabled = 0;
+
+        if (!rpc) {
+                goto out;
+        }
+
+        conn = &rpc->conn;
+
+        pthread_mutex_lock (&conn->lock);
+        {
+                disabled = rpc->disabled;
+        }
+        pthread_mutex_unlock (&conn->lock);
+
+out:
+        return disabled;
+}
+
 void
 rpc_clnt_disable (struct rpc_clnt *rpc)
 {
diff --git a/rpc/rpc-lib/src/rpc-clnt.h b/rpc/rpc-lib/src/rpc-clnt.h
index 0da165559..263d5f731 100644
--- a/rpc/rpc-lib/src/rpc-clnt.h
+++ b/rpc/rpc-lib/src/rpc-clnt.h
@@ -244,4 +244,7 @@ rpc_clnt_transport_unix_options_build (dict_t **options, char *filepath,
 void
 rpc_clnt_disable (struct rpc_clnt *rpc);
 
+char
+rpc_clnt_is_disabled (struct rpc_clnt *rpc);
+
 #endif /* !_RPC_CLNT_H */