LU-1428 ldlm: fix a race in ldlm_lock_destroy_internal

ldlm_lock::l_exp_hash should be protected by internal lock of
cfs_hash, but we called cfs_hlist_unhashed(lock::l_exp_hash)
w/o holding cfs_hash lock in ldlm_lock_destroy_internal,
which means if someone called ldlm_lock_cancel on a lock while
export::exp_lock_hash is in progress of rehashing (thread context of
cfs_workitem), there could be tiny window between deleting this lock
from bucket[A] and re-adding it to bucket[B] of l_exp_hash, and
cfs_hlist_unhashed(lock::l_exp_hash) will return 1 in this window,
then we destroyed a lock but left it on l_exp_hash forever because
lock::l_destroyed has been set to 1 and ldlm_lock_destroy_internal()
wouldn't be able to remove the lock from l_exp_hash even it's called
infinite times in ldlm_cancel_locks_for_export_cb().

This patch also added some debug information to
ldlm_cancel_locks_for_export_cb in case this patch can't fix this
problem.

Signed-off-by: Liang Zhen <liang@whamcloud.com>
Change-Id: Ia0932658b3f085a55535e36bee4fb833e74fa242
Reviewed-on: http://review.whamcloud.com/3028
Tested-by: Hudson
Reviewed-by: Fan Yong <yong.fan@whamcloud.com>
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Lai Siyao <laisiyao@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>

diff --git a/libcfs/libcfs/hash.c b/libcfs/libcfs/hash.c
index 8b8161e..d3df598 100644 (file)
--- a/libcfs/libcfs/hash.c
+++ b/libcfs/libcfs/hash.c
@@ -1306,10 +1306,15 @@ cfs_hash_del(cfs_hash_t *hs, const void *key, cfs_hlist_node_t *hnode)
         cfs_hash_lock(hs, 0);
         cfs_hash_dual_bd_get_and_lock(hs, key, bds, 1);
 
-        if (bds[1].bd_bucket == NULL && hnode != NULL)
-                cfs_hash_bd_del_locked(hs, &bds[0], hnode);
-        else
-                hnode = cfs_hash_dual_bd_finddel_locked(hs, bds, key, hnode);
+       /* NB: do nothing if @hnode is not in hash table */
+       if (hnode == NULL || !cfs_hlist_unhashed(hnode)) {
+               if (bds[1].bd_bucket == NULL && hnode != NULL) {
+                       cfs_hash_bd_del_locked(hs, &bds[0], hnode);
+               } else {
+                       hnode = cfs_hash_dual_bd_finddel_locked(hs, bds,
+                                                               key, hnode);
+               }
+       }
 
         if (hnode != NULL) {
                 obj  = cfs_hash_object(hs, hnode);

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 550cc1c..09126de 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -335,10 +335,12 @@ int ldlm_lock_destroy_internal(struct ldlm_lock *lock)
         }
         lock->l_destroyed = 1;
 
-        if (lock->l_export && lock->l_export->exp_lock_hash &&
-            !cfs_hlist_unhashed(&lock->l_exp_hash))
-                cfs_hash_del(lock->l_export->exp_lock_hash,
-                             &lock->l_remote_handle, &lock->l_exp_hash);
+       if (lock->l_export && lock->l_export->exp_lock_hash) {
+               /* NB: it's safe to call cfs_hash_del() even lock isn't
+                * in exp_lock_hash. */
+               cfs_hash_del(lock->l_export->exp_lock_hash,
+                            &lock->l_remote_handle, &lock->l_exp_hash);
+       }
 
         ldlm_lock_remove_from_lru(lock);
         class_handle_unhash(&lock->l_handle);
@@ -1818,11 +1820,17 @@ int ldlm_lock_set_data(struct lustre_handle *lockh, void *data)
 }
 EXPORT_SYMBOL(ldlm_lock_set_data);
 
+struct export_cl_data {
+       struct obd_export       *ecl_exp;
+       int                     ecl_loop;
+};
+
 int ldlm_cancel_locks_for_export_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd,
                                     cfs_hlist_node_t *hnode, void *data)
 
 {
-        struct obd_export    *exp  = data;
+       struct export_cl_data   *ecl = (struct export_cl_data *)data;
+       struct obd_export       *exp  = ecl->ecl_exp;
         struct ldlm_lock     *lock = cfs_hash_object(hs, hnode);
         struct ldlm_resource *res;
 
@@ -1835,13 +1843,28 @@ int ldlm_cancel_locks_for_export_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd,
         ldlm_reprocess_all(res);
         ldlm_resource_putref(res);
         LDLM_LOCK_RELEASE(lock);
-        return 0;
+
+       ecl->ecl_loop++;
+       if ((ecl->ecl_loop & -ecl->ecl_loop) == ecl->ecl_loop) {
+               CDEBUG(D_INFO,
+                      "Cancel lock %p for export %p (loop %d), still have "
+                      "%d locks left on hash table.\n",
+                      lock, exp, ecl->ecl_loop,
+                      cfs_atomic_read(&hs->hs_count));
+       }
+
+       return 0;
 }
 
 void ldlm_cancel_locks_for_export(struct obd_export *exp)
 {
-        cfs_hash_for_each_empty(exp->exp_lock_hash,
-                                ldlm_cancel_locks_for_export_cb, exp);
+       struct export_cl_data   ecl = {
+               .ecl_exp        = exp,
+               .ecl_loop       = 0,
+       };
+
+       cfs_hash_for_each_empty(exp->exp_lock_hash,
+                               ldlm_cancel_locks_for_export_cb, &ecl);
 }
 
 /**

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 0c4a181..4aad818 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -2218,10 +2218,13 @@ int ldlm_revoke_lock_cb(cfs_hash_t *hs, cfs_hash_bd_t *bd,
         LASSERT(!lock->l_blocking_lock);
 
         lock->l_flags |= LDLM_FL_AST_SENT;
-        if (lock->l_export && lock->l_export->exp_lock_hash &&
-            !cfs_hlist_unhashed(&lock->l_exp_hash))
-                cfs_hash_del(lock->l_export->exp_lock_hash,
-                             &lock->l_remote_handle, &lock->l_exp_hash);
+        if (lock->l_export && lock->l_export->exp_lock_hash) {
+               /* NB: it's safe to call cfs_hash_del() even lock isn't
+                * in exp_lock_hash. */
+               cfs_hash_del(lock->l_export->exp_lock_hash,
+                            &lock->l_remote_handle, &lock->l_exp_hash);
+       }
+
         cfs_list_add_tail(&lock->l_rk_ast, rpc_list);
         LDLM_LOCK_GET(lock);