Adding async_disconnecting flag to avoid multiple async rpc disconn

* async_disconnecting is used to check if waiting is needed.

* No need to call session_put() in rm_remote_free().
  Because at this moment, maybe some other requests are still in progress.

Signed-off-by: Hua Huang <[email protected]>

Author: Hua Huang

rm/rm.c

@@ -1385,7 +1385,6 @@ static void rm_remote_free(struct m0_ref *ref)
 {
 	struct m0_rm_remote *rem =
 	             container_of(ref, struct m0_rm_remote, rem_refcnt);
-	struct m0_rpc_session *sess = rem->rem_session;
 
 	/*
 	 * Free only those remotes who connected to us asking for
@@ -1399,13 +1398,6 @@ static void rm_remote_free(struct m0_ref *ref)
 	m0_remotes_tlist_del(rem);
 	m0_rm_remote_fini(rem);
 	m0_free(rem);
-
-	/*
-	 * Note: it seems the session can be NULL only in UTs here.
-	 * (In particular, at rm-ut:fom-funcs.)
-	 */
-	if (sess != NULL)
-		m0_rpc_service_reverse_session_put(sess, true);
 }
 
 M0_INTERNAL void m0_rm_remote_init(struct m0_rm_remote   *rem,

rpc/link.c

@@ -615,6 +615,7 @@ M0_INTERNAL int m0_rpc_link_init(struct m0_rpc_link *rlink,
 
 	rlink->rlk_connected = false;
 	rlink->rlk_rc        = 0;
+	rlink->rlk_async_disconnecting = false;
 
 	rc = m0_net_end_point_create(&net_ep, &mach->rm_tm, ep);
 	if (rc == 0) {

rpc/link.h

@@ -53,6 +53,7 @@ struct m0_rpc_link {
 	struct m0_mutex        rlk_wait_mutex;
 	m0_time_t              rlk_timeout;
 	bool                   rlk_connected;
+	bool                   rlk_async_disconnecting;
 };
 
 enum m0_rpc_link_states {

rpc/service.c

@@ -237,7 +237,9 @@ m0_rpc_service_reverse_session_put(struct m0_rpc_session *sess, bool disc)
 	rlk  = container_of(sess, struct m0_rpc_link, rlk_sess);
 	revc = container_of(rlk, struct m0_reverse_connection, rcf_rlink);
 
-	if (revc->rcf_rlink.rlk_connected) {
+	if (revc->rcf_rlink.rlk_connected &&
+	    !revc->rcf_rlink.rlk_async_disconnecting) {
+		revc->rcf_rlink.rlk_async_disconnecting = true;
 		M0_PRE(!m0_clink_is_armed(&revc->rcf_disc_wait));
 		m0_clink_fini(&revc->rcf_disc_wait);
 		m0_clink_init(&revc->rcf_disc_wait,
@@ -271,7 +273,8 @@ m0_rpc_service_reverse_sessions_cleanup(struct m0_reqh_service *service)
 						   false);
 	} m0_tlist_endfor;
 	m0_tl_teardown(rev_conn, &svc->rps_rev_conns, revc) {
-		if (m0_clink_is_armed(&revc->rcf_disc_wait)) {
+		if (m0_clink_is_armed(&revc->rcf_disc_wait) &&
+		    revc->rcf_rlink.rlk_async_disconnecting) {
 			m0_chan_wait(&revc->rcf_disc_wait);
 			m0_clink_fini(&revc->rcf_disc_wait);
 		}