rpc: Cleanup rpc object in TRANSPORT_CLEANUP event

rpc_transport object should be alive as long as the rpc_clnt object is
alive. To ensure this, on rpc_clnt's last unref, we cleanup the
corresponding rpc_transport object and complete the rpc_clnt cleanup
later, in a bottom-up fashion.

Introduced rpc_clnt_is_disabled, to allow higher layers to differentiate
between the 'final'[1] disconnect triggered from upper layers, and a
normal 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 going 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/5107
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

4 files changed, 43 insertions, 26 deletions

diff --git a/rpc/rpc-lib/src/rpc-clnt.c b/rpc/rpc-lib/src/rpc-clnt.c
index c8bb8e29855..b1d004aa869 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,21 @@ 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);
+}
+
+static void
+rpc_clnt_destroy (struct rpc_clnt *rpc)
+{
+        if (!rpc)
+                return;
 
-        rpc_clnt_reconnect_cleanup (&rpc->conn);
         saved_frames_destroy (rpc->conn.saved_frames);
         pthread_mutex_destroy (&rpc->lock);
         pthread_mutex_destroy (&rpc->conn.lock);
@@ -1579,13 +1583,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 1f74e55d73e..584963ad036 100644
--- a/rpc/rpc-lib/src/rpc-clnt.h
+++ b/rpc/rpc-lib/src/rpc-clnt.h
@@ -240,4 +240,7 @@ int rpcclnt_cbk_program_register (struct rpc_clnt *svc,
 void
 rpc_clnt_disable (struct rpc_clnt *rpc);
 
+char
+rpc_clnt_is_disabled (struct rpc_clnt *rpc);
+
 #endif /* !_RPC_CLNT_H */
diff --git a/rpc/rpc-lib/src/rpc-transport.c b/rpc/rpc-lib/src/rpc-transport.c
index 55ff0a3e865..89f3b3e8a0a 100644
--- a/rpc/rpc-lib/src/rpc-transport.c
+++ b/rpc/rpc-lib/src/rpc-transport.c
@@ -477,6 +477,8 @@ rpc_transport_unref (rpc_transport_t *this)
                 if (this->mydata)
                         this->notify (this, this->mydata, RPC_TRANSPORT_CLEANUP,
                                       NULL);
+                this->mydata = NULL;
+                this->notify = NULL;
                 rpc_transport_destroy (this);
 	}
 
@@ -520,18 +522,6 @@ out:
 }
 
 
-inline int
-rpc_transport_unregister_notify (rpc_transport_t *trans)
-{
-        GF_VALIDATE_OR_GOTO ("rpc-transport", trans, out);
-
-        trans->notify = NULL;
-        trans->mydata = NULL;
-
-out:
-        return 0;
-}
-
 
 //give negative values to skip setting that value
 //this function asserts if both the values are negative.
diff --git a/rpc/rpc-lib/src/rpc-transport.h b/rpc/rpc-lib/src/rpc-transport.h
index 9e78b5a040c..a8744d61810 100644
--- a/rpc/rpc-lib/src/rpc-transport.h
+++ b/rpc/rpc-lib/src/rpc-transport.h
@@ -273,9 +273,6 @@ int
 rpc_transport_register_notify (rpc_transport_t *trans, rpc_transport_notify_t,
                                void *mydata);
 
-int
-rpc_transport_unregister_notify (rpc_transport_t *trans);
-
 int32_t
 rpc_transport_get_peername (rpc_transport_t *this, char *hostname, int hostlen);