LU-68 Fix a race between lock cancel and write

[fs/lustre-release.git] / lustre / osc / osc_lock.c

diff --git a/lustre/osc/osc_lock.c b/lustre/osc/osc_lock.c
index 42de3de..0b03ea7 100644 (file)
--- a/lustre/osc/osc_lock.c
+++ b/lustre/osc/osc_lock.c
@@ -155,18 +155,19 @@ static void osc_lock_detach(const struct lu_env *env, struct osc_lock *olck)
         if (dlmlock->l_granted_mode == dlmlock->l_req_mode) {
                 struct cl_object *obj = olck->ols_cl.cls_obj;
                 struct cl_attr *attr  = &osc_env_info(env)->oti_attr;
-                __u64 old_kms = cl2osc(obj)->oo_oinfo->loi_kms;
+                __u64 old_kms;
 
+                cl_object_attr_lock(obj);
+                /* Must get the value under the lock to avoid possible races. */
+                old_kms = cl2osc(obj)->oo_oinfo->loi_kms;
                 /* Update the kms. Need to loop all granted locks.
                  * Not a problem for the client */
                 attr->cat_kms = ldlm_extent_shift_kms(dlmlock, old_kms);
-                unlock_res_and_lock(dlmlock);
 
-                cl_object_attr_lock(obj);
                 cl_object_attr_set(env, obj, attr, CAT_KMS);
                 cl_object_attr_unlock(obj);
-        } else
-                unlock_res_and_lock(dlmlock);
+        }
+        unlock_res_and_lock(dlmlock);
 
         /* release a reference taken in osc_lock_upcall0(). */
         LASSERT(olck->ols_has_ref);
@@ -1104,12 +1105,26 @@ static int osc_lock_enqueue_wait(const struct lu_env *env,
         cfs_spin_unlock(&hdr->coh_lock_guard);
 
         if (conflict) {
-                CDEBUG(D_DLMTRACE, "lock %p is confliced with %p, will wait\n",
-                       lock, conflict);
-                lu_ref_add(&conflict->cll_reference, "cancel-wait", lock);
-                LASSERT(lock->cll_conflict == NULL);
-                lock->cll_conflict = conflict;
-                rc = CLO_WAIT;
+                if (lock->cll_descr.cld_mode == CLM_GROUP) {
+                        /* we want a group lock but a previous lock request
+                         * conflicts, we do not wait but return 0 so the
+                         * request is send to the server
+                         */
+                        CDEBUG(D_DLMTRACE, "group lock %p is conflicted "
+                                           "with %p, no wait, send to server\n",
+                               lock, conflict);
+                        cl_lock_put(env, conflict);
+                        rc = 0;
+                } else {
+                        CDEBUG(D_DLMTRACE, "lock %p is conflicted with %p, "
+                                           "will wait\n",
+                               lock, conflict);
+                        LASSERT(lock->cll_conflict == NULL);
+                        lu_ref_add(&conflict->cll_reference, "cancel-wait",
+                                   lock);
+                        lock->cll_conflict = conflict;
+                        rc = CLO_WAIT;
+                }
         }
         RETURN(rc);
 }
@@ -1134,11 +1149,6 @@ static int osc_lock_enqueue(const struct lu_env *env,
 {
         struct osc_lock          *ols     = cl2osc_lock(slice);
         struct cl_lock           *lock    = ols->ols_cl.cls_lock;
-        struct osc_object        *obj     = cl2osc(slice->cls_obj);
-        struct osc_thread_info   *info    = osc_env_info(env);
-        struct ldlm_res_id       *resname = &info->oti_resname;
-        ldlm_policy_data_t       *policy  = &info->oti_policy;
-        struct ldlm_enqueue_info *einfo   = &ols->ols_einfo;
         int result;
         ENTRY;
 
@@ -1146,18 +1156,22 @@ static int osc_lock_enqueue(const struct lu_env *env,
         LASSERT(lock->cll_state == CLS_QUEUING);
         LASSERT(ols->ols_state == OLS_NEW);
 
-        osc_lock_build_res(env, obj, resname);
-        osc_lock_build_policy(env, lock, policy);
         ols->ols_flags = osc_enq2ldlm_flags(enqflags);
         if (ols->ols_flags & LDLM_FL_HAS_INTENT)
                 ols->ols_glimpse = 1;
-        if (!(enqflags & CEF_MUST))
+        if (!osc_lock_is_lockless(ols) && !(enqflags & CEF_MUST))
                 /* try to convert this lock to a lockless lock */
                 osc_lock_to_lockless(env, ols, (enqflags & CEF_NEVER));
 
         result = osc_lock_enqueue_wait(env, ols);
         if (result == 0) {
                 if (!osc_lock_is_lockless(ols)) {
+                        struct osc_object        *obj = cl2osc(slice->cls_obj);
+                        struct osc_thread_info   *info = osc_env_info(env);
+                        struct ldlm_res_id       *resname = &info->oti_resname;
+                        ldlm_policy_data_t       *policy = &info->oti_policy;
+                        struct ldlm_enqueue_info *einfo = &ols->ols_einfo;
+
                         if (ols->ols_locklessable)
                                 ols->ols_flags |= LDLM_FL_DENY_ON_CONTENTION;
 
@@ -1171,6 +1185,8 @@ static int osc_lock_enqueue(const struct lu_env *env,
                          * ldlm_lock_match(LDLM_FL_LVB_READY) waits for
                          * LDLM_CP_CALLBACK.
                          */
+                        osc_lock_build_res(env, obj, resname);
+                        osc_lock_build_policy(env, lock, policy);
                         result = osc_enqueue_base(osc_export(obj), resname,
                                           &ols->ols_flags, policy,
                                           &ols->ols_lvb,
@@ -1501,14 +1517,6 @@ static const struct cl_lock_operations osc_lock_ops = {
         .clo_fits_into = osc_lock_fits_into,
 };
 
-static int osc_lock_lockless_enqueue(const struct lu_env *env,
-                                     const struct cl_lock_slice *slice,
-                                     struct cl_io *unused, __u32 enqflags)
-{
-        LBUG();
-        return 0;
-}
-
 static int osc_lock_lockless_unuse(const struct lu_env *env,
                                    const struct cl_lock_slice *slice)
 {
@@ -1584,7 +1592,7 @@ static int osc_lock_lockless_fits_into(const struct lu_env *env,
 
 static const struct cl_lock_operations osc_lock_lockless_ops = {
         .clo_fini      = osc_lock_fini,
-        .clo_enqueue   = osc_lock_lockless_enqueue,
+        .clo_enqueue   = osc_lock_enqueue,
         .clo_wait      = osc_lock_lockless_wait,
         .clo_unuse     = osc_lock_lockless_unuse,
         .clo_state     = osc_lock_lockless_state,