branch: HEAD

[fs/lustre-release.git] / lustre / lov / lov_lock.c

diff --git a/lustre/lov/lov_lock.c b/lustre/lov/lov_lock.c
index 905b6cc..79a6942 100644 (file)
--- a/lustre/lov/lov_lock.c
+++ b/lustre/lov/lov_lock.c
@@ -144,7 +144,7 @@ static struct cl_lock *lov_sublock_alloc(const struct lu_env *env,
         LASSERT(idx < lck->lls_nr);
         ENTRY;
 
         LASSERT(idx < lck->lls_nr);
         ENTRY;
 

-        OBD_SLAB_ALLOC_PTR(link, lov_lock_link_kmem);
+        OBD_SLAB_ALLOC_PTR_GFP(link, lov_lock_link_kmem, CFS_ALLOC_IO);

         if (link != NULL) {
                 struct lov_sublock_env *subenv;
                 struct lov_lock_sub  *lls;
         if (link != NULL) {
                 struct lov_sublock_env *subenv;
                 struct lov_lock_sub  *lls;


@@ -330,6 +330,8 @@ static int lov_lock_sub_init(const struct lu_env *env,
                         descr->cld_start = cl_index(descr->cld_obj, start);
                         descr->cld_end   = cl_index(descr->cld_obj, end);
                         descr->cld_mode  = parent->cll_descr.cld_mode;
                         descr->cld_start = cl_index(descr->cld_obj, start);
                         descr->cld_end   = cl_index(descr->cld_obj, end);
                         descr->cld_mode  = parent->cll_descr.cld_mode;

+                        descr->cld_gid   = parent->cll_descr.cld_gid;
+                        /* XXX has no effect */

                         lck->lls_sub[nr].sub_got = *descr;
                         lck->lls_sub[nr].sub_stripe = stripe;
                         nr++;
                         lck->lls_sub[nr].sub_got = *descr;
                         lck->lls_sub[nr].sub_stripe = stripe;
                         nr++;


@@ -401,6 +403,7 @@ static int lov_sublock_release(const struct lu_env *env, struct lov_lock *lck,
                  * while sub-lock is being paged out.
                  */
                 dying = (sublock->cll_descr.cld_mode == CLM_PHANTOM ||
                  * while sub-lock is being paged out.
                  */
                 dying = (sublock->cll_descr.cld_mode == CLM_PHANTOM ||

+                         sublock->cll_descr.cld_mode == CLM_GROUP ||

                          (sublock->cll_flags & (CLF_CANCELPEND|CLF_DOOMED))) &&
                         sublock->cll_holds == 1;
                 if (dying)
                          (sublock->cll_flags & (CLF_CANCELPEND|CLF_DOOMED))) &&
                         sublock->cll_holds == 1;
                 if (dying)


@@ -789,19 +792,49 @@ static int lock_lock_multi_match()
 }
 #endif
 
 }
 #endif
 

-static int lov_is_same_stripe(struct lov_object *lov, int stripe,
-                              const struct cl_lock_descr *descr)
+/**
+ * Check if the extent region \a descr is covered by \a child against the
+ * specific \a stripe.
+ */
+static int lov_lock_stripe_is_matching(const struct lu_env *env,
+                                       struct lov_object *lov, int stripe,
+                                       const struct cl_lock_descr *child,
+                                       const struct cl_lock_descr *descr)

 {
         struct lov_stripe_md *lsm = lov_r0(lov)->lo_lsm;
         obd_off start;
         obd_off end;
 {
         struct lov_stripe_md *lsm = lov_r0(lov)->lo_lsm;
         obd_off start;
         obd_off end;

+        int result;

 
 

+        if (lov_r0(lov)->lo_nr == 1)
+                return cl_lock_ext_match(child, descr);
+
+        /*
+         * For a multi-stripes object:
+         * - make sure the descr only covers child's stripe, and
+         * - check if extent is matching.
+         */

         start = cl_offset(&lov->lo_cl, descr->cld_start);
         end   = cl_offset(&lov->lo_cl, descr->cld_end + 1) - 1;
         start = cl_offset(&lov->lo_cl, descr->cld_start);
         end   = cl_offset(&lov->lo_cl, descr->cld_end + 1) - 1;

-        return
-                end - start <= lsm->lsm_stripe_size &&
-                stripe == lov_stripe_number(lsm, start) &&
-                stripe == lov_stripe_number(lsm, end);
+        result = end - start <= lsm->lsm_stripe_size &&
+                 stripe == lov_stripe_number(lsm, start) &&
+                 stripe == lov_stripe_number(lsm, end);
+        if (result) {
+                struct cl_lock_descr *subd = &lov_env_info(env)->lti_ldescr;
+                obd_off sub_start;
+                obd_off sub_end;
+
+                subd->cld_obj  = NULL;   /* don't need sub object at all */
+                subd->cld_mode = descr->cld_mode;
+                subd->cld_gid  = descr->cld_gid;
+                result = lov_stripe_intersects(lsm, stripe, start, end,
+                                               &sub_start, &sub_end);
+                LASSERT(result);
+                subd->cld_start = cl_index(child->cld_obj, sub_start);
+                subd->cld_end   = cl_index(child->cld_obj, sub_end);
+                result = cl_lock_ext_match(child, subd);
+        }
+        return result;

 }
 
 /**
 }
 
 /**


@@ -827,20 +860,17 @@ static int lov_lock_fits_into(const struct lu_env *env,
 
         ENTRY;
 
 
         ENTRY;
 

-        if (lov->lls_nr == 1) {
+        if (need->cld_mode == CLM_GROUP)

                 /*
                 /*

-                 * If a lock is on a single stripe, it's enough to check that
-                 * @need lock matches actually granted stripe lock, and...
+                 * always allow to match group lock.

                  */
                  */

-                result = cl_lock_ext_match(&lov->lls_sub[0].sub_got, need);
-                if (result && lov_r0(obj)->lo_nr > 1)
-                        /*
-                         * ... @need is on the same stripe, if multiple
-                         * stripes are possible at all for this object.
-                         */
-                        result = lov_is_same_stripe(cl2lov(slice->cls_obj),
-                                                    lov->lls_sub[0].sub_stripe,
-                                                    need);
+                result = cl_lock_ext_match(&lov->lls_orig, need);
+        else if (lov->lls_nr == 1) {
+                struct cl_lock_descr *got = &lov->lls_sub[0].sub_got;
+                result = lov_lock_stripe_is_matching(env,
+                                                     cl2lov(slice->cls_obj),
+                                                     lov->lls_sub[0].sub_stripe,
+                                                     got, need);

         } else if (io->ci_type != CIT_TRUNC && io->ci_type != CIT_MISC &&
                    !cl_io_is_append(io) && need->cld_mode != CLM_PHANTOM)
                 /*
         } else if (io->ci_type != CIT_TRUNC && io->ci_type != CIT_MISC &&
                    !cl_io_is_append(io) && need->cld_mode != CLM_PHANTOM)
                 /*


@@ -1003,7 +1033,7 @@ int lov_lock_init_raid0(const struct lu_env *env, struct cl_object *obj,
         int result;
 
         ENTRY;
         int result;
 
         ENTRY;

-        OBD_SLAB_ALLOC_PTR(lck, lov_lock_kmem);
+        OBD_SLAB_ALLOC_PTR_GFP(lck, lov_lock_kmem, CFS_ALLOC_IO);

         if (lck != NULL) {
                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_lock_ops);
                 result = lov_lock_sub_init(env, lck, io);
         if (lck != NULL) {
                 cl_lock_slice_add(lock, &lck->lls_cl, obj, &lov_lock_ops);
                 result = lov_lock_sub_init(env, lck, io);


@@ -1018,7 +1048,7 @@ static struct cl_lock_closure *lov_closure_get(const struct lu_env *env,
         struct cl_lock_closure *closure;
 
         closure = &lov_env_info(env)->lti_closure;
         struct cl_lock_closure *closure;
 
         closure = &lov_env_info(env)->lti_closure;

-        LINVRNT(list_empty(&closure->clc_list));
+        LASSERT(list_empty(&closure->clc_list));

         cl_lock_closure_init(env, closure, parent, 1);
         return closure;
 }
         cl_lock_closure_init(env, closure, parent, 1);
         return closure;
 }