LU-10961 ldlm: don't cancel DoM locks before replay

Weigh a DOM locks before lock replay like that is done
for OSC EXTENT locks and don't cancel locks with data.

Add DoM replay tests for file creation and write cases.

Signed-off-by: Mikhail Pershin <mpershin@whamcloud.com>
Change-Id: If706835455c2b25c01cfcde033fccc4f3c67ee24
Reviewed-on: https://review.whamcloud.com/32791
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Patrick Farrell <paf@cray.com>
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/lustre_osc.h b/lustre/include/lustre_osc.h
index ebc0168..3fb2d33 100644 (file)
--- a/lustre/include/lustre_osc.h
+++ b/lustre/include/lustre_osc.h
@@ -722,6 +722,7 @@ void osc_lock_cancel(const struct lu_env *env,
                     const struct cl_lock_slice *slice);
 void osc_lock_fini(const struct lu_env *env, struct cl_lock_slice *slice);
 int osc_ldlm_glimpse_ast(struct ldlm_lock *dlmlock, void *data);
+unsigned long osc_ldlm_weigh_ast(struct ldlm_lock *dlmlock);
 
 /*****************************************************************************
  *

diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c
index 2ef7061..67c6291 100644 (file)
--- a/lustre/mdc/mdc_request.c
+++ b/lustre/mdc/mdc_request.c
@@ -2507,6 +2507,12 @@ static int mdc_cancel_weight(struct ldlm_lock *lock)
        if (lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_OPEN)
                RETURN(0);
 
+       /* Special case for DoM locks, cancel only unused and granted locks */
+       if (ldlm_has_dom(lock) &&
+           (lock->l_granted_mode != lock->l_req_mode ||
+            osc_ldlm_weigh_ast(lock) != 0))
+               RETURN(0);
+
        RETURN(1);
 }
 

diff --git a/lustre/osc/osc_lock.c b/lustre/osc/osc_lock.c
index c80296a..b2042a4 100644 (file)
--- a/lustre/osc/osc_lock.c
+++ b/lustre/osc/osc_lock.c
@@ -611,9 +611,8 @@ static int weigh_cb(const struct lu_env *env, struct cl_io *io,
 {
        struct cl_page *page = ops->ops_cl.cpl_page;
 
-       if (cl_page_is_vmlocked(env, page)
-           || PageDirty(page->cp_vmpage) || PageWriteback(page->cp_vmpage)
-          )
+       if (cl_page_is_vmlocked(env, page) || PageDirty(page->cp_vmpage) ||
+           PageWriteback(page->cp_vmpage))
                return CLP_GANG_ABORT;
 
        *(pgoff_t *)cbdata = osc_index(ops) + 1;
@@ -622,12 +621,13 @@ static int weigh_cb(const struct lu_env *env, struct cl_io *io,
 
 static unsigned long osc_lock_weight(const struct lu_env *env,
                                     struct osc_object *oscobj,
-                                    struct ldlm_extent *extent)
+                                    loff_t start, loff_t end)
 {
-       struct cl_io     *io = osc_env_thread_io(env);
+       struct cl_io *io = osc_env_thread_io(env);
        struct cl_object *obj = cl_object_top(&oscobj->oo_cl);
-       pgoff_t          page_index;
-       int              result;
+       pgoff_t page_index;
+       int result;
+
        ENTRY;
 
        io->ci_obj = obj;
@@ -636,11 +636,10 @@ static unsigned long osc_lock_weight(const struct lu_env *env,
        if (result != 0)
                RETURN(result);
 
-       page_index = cl_index(obj, extent->start);
+       page_index = cl_index(obj, start);
        do {
                result = osc_page_gang_lookup(env, io, oscobj,
-                                             page_index,
-                                             cl_index(obj, extent->end),
+                                             page_index, cl_index(obj, end),
                                              weigh_cb, (void *)&page_index);
                if (result == CLP_GANG_ABORT)
                        break;
@@ -657,12 +656,13 @@ static unsigned long osc_lock_weight(const struct lu_env *env,
  */
 unsigned long osc_ldlm_weigh_ast(struct ldlm_lock *dlmlock)
 {
-       struct lu_env           *env;
-       struct osc_object       *obj;
-       struct osc_lock         *oscl;
-       unsigned long            weight;
-       bool                    found = false;
-       __u16                   refcheck;
+       struct lu_env *env;
+       struct osc_object *obj;
+       struct osc_lock *oscl;
+       unsigned long weight;
+       bool found = false;
+       __u16 refcheck;
+
        ENTRY;
 
        might_sleep();
@@ -678,7 +678,8 @@ unsigned long osc_ldlm_weigh_ast(struct ldlm_lock *dlmlock)
                /* Mostly because lack of memory, do not eliminate this lock */
                RETURN(1);
 
-       LASSERT(dlmlock->l_resource->lr_type == LDLM_EXTENT);
+       LASSERT(dlmlock->l_resource->lr_type == LDLM_EXTENT ||
+               ldlm_has_dom(dlmlock));
        lock_res_and_lock(dlmlock);
        obj = dlmlock->l_ast_data;
        if (obj)
@@ -702,7 +703,13 @@ unsigned long osc_ldlm_weigh_ast(struct ldlm_lock *dlmlock)
                GOTO(out, weight = 1);
        }
 
-       weight = osc_lock_weight(env, obj, &dlmlock->l_policy_data.l_extent);
+       if (ldlm_has_dom(dlmlock))
+               weight = osc_lock_weight(env, obj, 0, OBD_OBJECT_EOF);
+       else
+               weight = osc_lock_weight(env, obj,
+                                        dlmlock->l_policy_data.l_extent.start,
+                                        dlmlock->l_policy_data.l_extent.end);
+
        EXIT;
 
 out:
@@ -712,6 +719,7 @@ out:
        cl_env_put(env, &refcheck);
        return weight;
 }
+EXPORT_SYMBOL(osc_ldlm_weigh_ast);
 
 static void osc_lock_build_einfo(const struct lu_env *env,
                                 const struct cl_lock *lock,

diff --git a/lustre/tests/replay-single.sh b/lustre/tests/replay-single.sh
index b2a8e2d..2b93f2d 100755 (executable)
--- a/lustre/tests/replay-single.sh
+++ b/lustre/tests/replay-single.sh
@@ -4699,6 +4699,65 @@ test_121() {
 }
 run_test 121 "lock replay timed out and race"
 
+test_130a() {
+       [ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.10.90) ] &&
+               skip "Do not support Data-on-MDT before 2.11"
+
+       replay_barrier $SINGLEMDS
+       $LFS setstripe -E 1M -L mdt -E EOF -c 2 $DIR/$tfile
+       fail $SINGLEMDS
+
+       [ $($LFS getstripe -L $DIR/$tfile) == "mdt" ] ||
+               error "Fail to replay DoM file creation"
+}
+run_test 130a "DoM file create (setstripe) replay"
+
+test_130b() {
+       [ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.10.90) ] &&
+               skip "Do not support Data-on-MDT before 2.11"
+
+       mkdir $DIR/$tdir
+       $LFS setstripe -E 1M -L mdt -E EOF -c 2 $DIR/$tdir
+       replay_barrier $SINGLEMDS
+       touch $DIR/$tdir/$tfile
+       fail $SINGLEMDS
+
+       [ $($LFS getstripe -L $DIR/$tdir/$tfile) == "mdt" ] ||
+               error "Fail to replay DoM file creation"
+}
+run_test 130b "DoM file create (inherited) replay"
+
+test_131a() {
+       [ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.10.90) ] &&
+               skip "Do not support Data-on-MDT before 2.11"
+
+       $LFS setstripe -E 1M -L mdt -E EOF -c 2 $DIR/$tfile
+       replay_barrier $SINGLEMDS
+       echo "dom_data" | dd of=$DIR/$tfile bs=1 count=8
+       # lock is not canceled and will be replayed
+       fail $SINGLEMDS
+
+       [ $(cat $DIR/$tfile) == "dom_data" ] ||
+               error "Wrong file content after failover"
+}
+run_test 131a "DoM file write lock replay"
+
+test_131b() {
+       [ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.10.90) ] &&
+               skip "Do not support Data-on-MDT before 2.11"
+
+       $LFS setstripe -E 1M -L mdt -E EOF -c 2 $DIR/$tfile
+       replay_barrier $SINGLEMDS
+       echo "dom_data" | dd of=$DIR/$tfile bs=1 count=8
+       cancel_lru_locks mdc
+
+       fail $SINGLEMDS
+
+       [ $(cat $DIR/$tfile) == "dom_data" ] ||
+               error "Wrong file content after failover"
+}
+run_test 131b "DoM file write replay"
+
 complete $SECONDS
 check_and_cleanup_lustre
 exit_status