b24336 ldlm_resource::lr_lvb_data is protected by wrong lock

- ldlm_resource::lr_lvb_data should always be protected by lr_lvb_sem
- cleanup some unnecessary lock dance

i=vitaly.fertman
i=green

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index d788767..c9f4536 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -471,8 +471,7 @@ void ldlm_lock2handle(const struct ldlm_lock *lock, struct lustre_handle *lockh)
 struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
                                      int flags)
 {
 struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
                                      int flags)
 {

-        struct ldlm_namespace *ns;
-        struct ldlm_lock *lock, *retval = NULL;
+        struct ldlm_lock *lock;

         ENTRY;
 
         LASSERT(handle);
         ENTRY;
 
         LASSERT(handle);


@@ -481,36 +480,36 @@ struct ldlm_lock *__ldlm_handle2lock(const struct lustre_handle *handle,
         if (lock == NULL)
                 RETURN(NULL);
 
         if (lock == NULL)
                 RETURN(NULL);
 

-        LASSERT(lock->l_resource != NULL);
-        ns = ldlm_lock_to_ns(lock);
-        LASSERT(ns != NULL);
+        /* It's unlikely but possible that someone marked the lock as
+         * destroyed after we did handle2object on it */
+        if (flags == 0 && !lock->l_destroyed) {
+                lu_ref_add(&lock->l_reference, "handle", cfs_current());
+                RETURN(lock);
+        }

 
 

-        lu_ref_add_atomic(&lock->l_reference, "handle", cfs_current());

         lock_res_and_lock(lock);
 
         lock_res_and_lock(lock);
 

-        /* It's unlikely but possible that someone marked the lock as
-         * destroyed after we did handle2object on it */
-        if (lock->l_destroyed) {
+        LASSERT(lock->l_resource != NULL);
+
+        lu_ref_add_atomic(&lock->l_reference, "handle", cfs_current());
+        if (unlikely(lock->l_destroyed)) {

                 unlock_res_and_lock(lock);
                 CDEBUG(D_INFO, "lock already destroyed: lock %p\n", lock);
                 LDLM_LOCK_PUT(lock);
                 unlock_res_and_lock(lock);
                 CDEBUG(D_INFO, "lock already destroyed: lock %p\n", lock);
                 LDLM_LOCK_PUT(lock);

-                GOTO(out, retval);
+                RETURN(NULL);

         }
 
         if (flags && (lock->l_flags & flags)) {
                 unlock_res_and_lock(lock);
                 LDLM_LOCK_PUT(lock);
         }
 
         if (flags && (lock->l_flags & flags)) {
                 unlock_res_and_lock(lock);
                 LDLM_LOCK_PUT(lock);

-                GOTO(out, retval);
+                RETURN(NULL);

         }
 
         if (flags)
                 lock->l_flags |= flags;
 
         unlock_res_and_lock(lock);
         }
 
         if (flags)
                 lock->l_flags |= flags;
 
         unlock_res_and_lock(lock);

-        retval = lock;
-        EXIT;
- out:
-        return retval;
+        RETURN(lock);

 }
 
 void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc)
 }
 
 void ldlm_lock2desc(struct ldlm_lock *lock, struct ldlm_lock_desc *desc)

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index a3f96fc..b750b1c 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -851,10 +851,10 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         if (lock->l_resource->lr_lvb_len) {
                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
 
         if (lock->l_resource->lr_lvb_len) {
                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
 

-                lock_res_and_lock(lock);
+                cfs_down(&lock->l_resource->lr_lvb_sem);

                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                        lock->l_resource->lr_lvb_len);
                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                        lock->l_resource->lr_lvb_len);

-                unlock_res_and_lock(lock);
+                cfs_up(&lock->l_resource->lr_lvb_sem);

         }
 
         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",
         }
 
         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",


@@ -1137,13 +1137,11 @@ existing_lock:
                  * local_lock_enqueue by the policy function. */
                 cookie = req;
         } else {
                  * local_lock_enqueue by the policy function. */
                 cookie = req;
         } else {

-                lock_res_and_lock(lock);

                 if (lock->l_resource->lr_lvb_len) {
                         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
                                              RCL_SERVER,
                                              lock->l_resource->lr_lvb_len);
                 }
                 if (lock->l_resource->lr_lvb_len) {
                         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
                                              RCL_SERVER,
                                              lock->l_resource->lr_lvb_len);
                 }

-                unlock_res_and_lock(lock);

 
                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
                         GOTO(out, rc = -ENOMEM);
 
                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE_EXTENT_ERR))
                         GOTO(out, rc = -ENOMEM);


@@ -1244,7 +1242,6 @@ existing_lock:
                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
                            "(err=%d, rc=%d)", err, rc);
 
                 LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
                            "(err=%d, rc=%d)", err, rc);
 

-                lock_res_and_lock(lock);

                 if (rc == 0) {
                         if (lock->l_resource->lr_lvb_len > 0) {
                                 void *lvb;
                 if (rc == 0) {
                         if (lock->l_resource->lr_lvb_len > 0) {
                                 void *lvb;


@@ -1254,14 +1251,17 @@ existing_lock:
                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
                                          req, lock);
 
                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
                                          req, lock);
 

+                                cfs_down(&lock->l_resource->lr_lvb_sem);

                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                                        lock->l_resource->lr_lvb_len);
                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                                        lock->l_resource->lr_lvb_len);

+                                cfs_up(&lock->l_resource->lr_lvb_sem);

                         }
                 } else {
                         }
                 } else {

+                        lock_res_and_lock(lock);

                         ldlm_resource_unlink_lock(lock);
                         ldlm_lock_destroy_nolock(lock);
                         ldlm_resource_unlink_lock(lock);
                         ldlm_lock_destroy_nolock(lock);

+                        unlock_res_and_lock(lock);

                 }
                 }

-                unlock_res_and_lock(lock);

 
                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
                         ldlm_reprocess_all(lock->l_resource);
 
                 if (!err && dlm_req->lock_desc.l_resource.lr_type != LDLM_FLOCK)
                         ldlm_reprocess_all(lock->l_resource);