LU-1592 ldlm: protect obd_export:exp_imp_reverse's change

* Protect obd_export::exp_imp_reverse from reconnect and destroy race.
* Add an assertion in class_import_put() to catch race in the first
  place.

Signed-off-by: Bobi Jam <bobijam@whamcloud.com>
Change-Id: If0abf6717456931c567d8d41c1d20fe49452e959
Reviewed-on: http://review.whamcloud.com/3684
Tested-by: Hudson
Reviewed-by: Fan Yong <yong.fan@whamcloud.com>
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: wangdi <di.wang@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/lustre_export.h b/lustre/include/lustre_export.h
index 66a0c6c..8dec5f2 100644 (file)
--- a/lustre/include/lustre_export.h
+++ b/lustre/include/lustre_export.h
@@ -202,7 +202,10 @@ struct obd_export {
         cfs_list_t                exp_obd_chain_timed;
         /** Obd device of this export */
         struct obd_device        *exp_obd;
         cfs_list_t                exp_obd_chain_timed;
         /** Obd device of this export */
         struct obd_device        *exp_obd;

-        /** "reverse" import to send requests (e.g. from ldlm) back to client */
+       /**
+        * "reverse" import to send requests (e.g. from ldlm) back to client
+        * exp_lock protect its change
+        */

         struct obd_import        *exp_imp_reverse;
         struct nid_stat          *exp_nid_stats;
         struct lprocfs_stats     *exp_md_stats;
         struct obd_import        *exp_imp_reverse;
         struct nid_stat          *exp_nid_stats;
         struct lprocfs_stats     *exp_md_stats;


@@ -226,7 +229,10 @@ struct obd_export {
         cfs_time_t                exp_last_request_time;
         /** On replay all requests waiting for replay are linked here */
         cfs_list_t                exp_req_replay_queue;
         cfs_time_t                exp_last_request_time;
         /** On replay all requests waiting for replay are linked here */
         cfs_list_t                exp_req_replay_queue;

-        /** protects exp_flags and exp_outstanding_replies */
+       /**
+        * protects exp_flags, exp_outstanding_replies and the change
+        * of exp_imp_reverse
+        */

         cfs_spinlock_t            exp_lock;
         /** Compatibility flags for this export */
         __u64                     exp_connect_flags;
         cfs_spinlock_t            exp_lock;
         /** Compatibility flags for this export */
         __u64                     exp_connect_flags;

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index c8c0697..f79160f 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -734,6 +734,7 @@ int target_handle_connect(struct ptlrpc_request *req)
        struct obd_device *target = NULL, *targref = NULL;
         struct obd_export *export = NULL;
         struct obd_import *revimp;
        struct obd_device *target = NULL, *targref = NULL;
         struct obd_export *export = NULL;
         struct obd_import *revimp;

+       struct obd_import *tmp_imp = NULL;

         struct lustre_handle conn;
         struct lustre_handle *tmp;
         struct obd_uuid tgtuuid;
         struct lustre_handle conn;
         struct lustre_handle *tmp;
         struct obd_uuid tgtuuid;


@@ -1184,23 +1185,25 @@ dont_check_exports:
         tmp = req_capsule_client_get(&req->rq_pill, &RMF_CONN);
         conn = *tmp;
 
         tmp = req_capsule_client_get(&req->rq_pill, &RMF_CONN);
         conn = *tmp;
 

-        if (export->exp_imp_reverse != NULL) {
-                /* destroyed import can be still referenced in ctxt */
-                obd_set_info_async(req->rq_svc_thread->t_env, export,
-                                   sizeof(KEY_REVIMP_UPD), KEY_REVIMP_UPD,
-                                   0, NULL, NULL);
-
-                client_destroy_import(export->exp_imp_reverse);
-        }
+       /* for the rest part, we return -ENOTCONN in case of errors
+        * in order to let client initialize connection again.
+        */
+       revimp = class_new_import(target);
+       if (revimp == NULL) {
+               CERROR("fail to alloc new reverse import.\n");
+               GOTO(out, rc = -ENOTCONN);
+       }

 
 

-        /* for the rest part, we return -ENOTCONN in case of errors
-         * in order to let client initialize connection again.
-         */
-        revimp = export->exp_imp_reverse = class_new_import(target);
-        if (!revimp) {
-                CERROR("fail to alloc new reverse import.\n");
-                GOTO(out, rc = -ENOTCONN);
-        }
+       cfs_spin_lock(&export->exp_lock);
+       if (export->exp_imp_reverse != NULL) {
+               /* destroyed import can be still referenced in ctxt */
+               obd_set_info_async(req->rq_svc_thread->t_env, export,
+                                  sizeof(KEY_REVIMP_UPD), KEY_REVIMP_UPD,
+                                  0, NULL, NULL);
+               tmp_imp = export->exp_imp_reverse;
+       }
+       export->exp_imp_reverse = revimp;
+       cfs_spin_unlock(&export->exp_lock);

 
         revimp->imp_connection = ptlrpc_connection_addref(export->exp_connection);
         revimp->imp_client = &export->exp_obd->obd_ldlm_client;
 
         revimp->imp_connection = ptlrpc_connection_addref(export->exp_connection);
         revimp->imp_client = &export->exp_obd->obd_ldlm_client;


@@ -1224,15 +1227,20 @@ dont_check_exports:
         else
                 revimp->imp_msghdr_flags &= ~MSGHDR_CKSUM_INCOMPAT18;
 
         else
                 revimp->imp_msghdr_flags &= ~MSGHDR_CKSUM_INCOMPAT18;
 

-        rc = sptlrpc_import_sec_adapt(revimp, req->rq_svc_ctx, &req->rq_flvr);
-        if (rc) {
-                CERROR("Failed to get sec for reverse import: %d\n", rc);
-                export->exp_imp_reverse = NULL;
-                class_destroy_import(revimp);
-        }
+       rc = sptlrpc_import_sec_adapt(revimp, req->rq_svc_ctx, &req->rq_flvr);
+       if (rc) {
+               CERROR("Failed to get sec for reverse import: %d\n", rc);
+               cfs_spin_lock(&export->exp_lock);
+               export->exp_imp_reverse = NULL;
+               cfs_spin_unlock(&export->exp_lock);
+               class_destroy_import(revimp);
+       }
+
+       class_import_put(revimp);

 
 

-        class_import_put(revimp);

 out:
 out:

+       if (tmp_imp != NULL)
+               client_destroy_import(tmp_imp);

         if (export) {
                 cfs_spin_lock(&export->exp_lock);
                 export->exp_connecting = 0;
         if (export) {
                 cfs_spin_lock(&export->exp_lock);
                 export->exp_connecting = 0;


@@ -1269,15 +1277,22 @@ int target_handle_disconnect(struct ptlrpc_request *req)
 
 void target_destroy_export(struct obd_export *exp)
 {
 
 void target_destroy_export(struct obd_export *exp)
 {

-        /* exports created from last_rcvd data, and "fake"
-           exports created by lctl don't have an import */
-        if (exp->exp_imp_reverse != NULL)
-                client_destroy_import(exp->exp_imp_reverse);
+       struct obd_import       *imp = NULL;
+       /* exports created from last_rcvd data, and "fake"
+          exports created by lctl don't have an import */
+       cfs_spin_lock(&exp->exp_lock);
+       if (exp->exp_imp_reverse != NULL) {
+               imp = exp->exp_imp_reverse;
+               exp->exp_imp_reverse = NULL;
+       }
+       cfs_spin_unlock(&exp->exp_lock);
+       if (imp != NULL)
+               client_destroy_import(imp);

 
 

-        LASSERT_ATOMIC_ZERO(&exp->exp_locks_count);
-        LASSERT_ATOMIC_ZERO(&exp->exp_rpc_count);
-        LASSERT_ATOMIC_ZERO(&exp->exp_cb_count);
-        LASSERT_ATOMIC_ZERO(&exp->exp_replay_count);
+       LASSERT_ATOMIC_ZERO(&exp->exp_locks_count);
+       LASSERT_ATOMIC_ZERO(&exp->exp_rpc_count);
+       LASSERT_ATOMIC_ZERO(&exp->exp_cb_count);
+       LASSERT_ATOMIC_ZERO(&exp->exp_replay_count);

 }
 
 /*
 }
 
 /*

diff --git a/lustre/obdclass/genops.c b/lustre/obdclass/genops.c
index 57e612a..8a141ab 100644 (file)
--- a/lustre/obdclass/genops.c
+++ b/lustre/obdclass/genops.c
@@ -974,7 +974,9 @@ void class_import_put(struct obd_import *imp)
                 obd_zombie_import_add(imp);
         }
 
                 obd_zombie_import_add(imp);
         }
 

-        EXIT;
+       /* catch possible import put race */
+       LASSERT_ATOMIC_GE_LT(&imp->imp_refcount, 0, LI_POISON);
+       EXIT;

 }
 EXPORT_SYMBOL(class_import_put);
 
 }
 EXPORT_SYMBOL(class_import_put);