LU-12296 llite: improve ll_dom_lock_cancel

ll_dom_lock_cancel() should zero kms attribute similar to
mdc_ldlm_blocking_ast0().

In order to avoid code duplication between mdc_ldlm_blocking_ast0()
and ll_dom_lock_cancel() - add cl_object_operations method to be able
to reach mdc's blocking ast from llite level.

Test illustrating the issue is added.

Cray-bug-id: LUS-7118
Signed-off-by: Vladimir Saveliev <c17830@cray.com>
Change-Id: I2b100ead6d420dbf561bc61be973d64dad317214
Reviewed-on: https://review.whamcloud.com/34858
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Mike Pershin <mpershin@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index 9f9d2e5..b13d1d0 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -420,6 +420,13 @@ struct cl_object_operations {
        void (*coo_req_attr_set)(const struct lu_env *env,
                                 struct cl_object *obj,
                                 struct cl_req_attr *attr);
        void (*coo_req_attr_set)(const struct lu_env *env,
                                 struct cl_object *obj,
                                 struct cl_req_attr *attr);

+       /**
+        * Flush \a obj data corresponding to \a lock. Used for DoM
+        * locks in llite's cancelling blocking ast callback.
+        */
+       int (*coo_object_flush)(const struct lu_env *env,
+                               struct cl_object *obj,
+                               struct ldlm_lock *lock);

 };
 
 /**
 };
 
 /**


@@ -2104,6 +2111,9 @@ int cl_object_fiemap(const struct lu_env *env, struct cl_object *obj,
 int cl_object_layout_get(const struct lu_env *env, struct cl_object *obj,
                         struct cl_layout *cl);
 loff_t cl_object_maxbytes(struct cl_object *obj);
 int cl_object_layout_get(const struct lu_env *env, struct cl_object *obj,
                         struct cl_layout *cl);
 loff_t cl_object_maxbytes(struct cl_object *obj);

+int cl_object_flush(const struct lu_env *env, struct cl_object *obj,
+                   struct ldlm_lock *lock);
+

 
 /**
  * Returns true, iff \a o0 and \a o1 are slices of the same object.
 
 /**
  * Returns true, iff \a o0 and \a o1 are slices of the same object.

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index 8b519ff..03911dc 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -184,15 +184,12 @@ int ll_test_inode_by_fid(struct inode *inode, void *opaque)
        return lu_fid_eq(&ll_i2info(inode)->lli_fid, opaque);
 }
 
        return lu_fid_eq(&ll_i2info(inode)->lli_fid, opaque);
 }
 

-int ll_dom_lock_cancel(struct inode *inode, struct ldlm_lock *lock)
+static int ll_dom_lock_cancel(struct inode *inode, struct ldlm_lock *lock)

 {
        struct lu_env *env;
        struct ll_inode_info *lli = ll_i2info(inode);
 {
        struct lu_env *env;
        struct ll_inode_info *lli = ll_i2info(inode);

-       struct cl_layout clt = { .cl_layout_gen = 0, };
-       int rc;

        __u16 refcheck;
        __u16 refcheck;

-
-
+       int rc;

        ENTRY;
 
        if (!lli->lli_clob) {
        ENTRY;
 
        if (!lli->lli_clob) {


@@ -206,28 +203,14 @@ int ll_dom_lock_cancel(struct inode *inode, struct ldlm_lock *lock)
        if (IS_ERR(env))
                RETURN(PTR_ERR(env));
 
        if (IS_ERR(env))
                RETURN(PTR_ERR(env));
 

-       rc = cl_object_layout_get(env, lli->lli_clob, &clt);
-       if (rc) {
-               CDEBUG(D_INODE, "Cannot get layout for "DFID"\n",
-                      PFID(ll_inode2fid(inode)));
-               rc = -ENODATA;
-       } else if (clt.cl_size == 0 || clt.cl_dom_comp_size == 0) {
-               CDEBUG(D_INODE, "DOM lock without DOM layout for "DFID"\n",
-                      PFID(ll_inode2fid(inode)));
-       } else {
-               enum cl_fsync_mode mode;
-               loff_t end = clt.cl_dom_comp_size - 1;
+       /* reach MDC layer to flush data under  the DoM ldlm lock */
+       rc = cl_object_flush(env, lli->lli_clob, lock);

 
 

-               mode = ldlm_is_discard_data(lock) ?
-                                       CL_FSYNC_DISCARD : CL_FSYNC_LOCAL;
-               rc = cl_sync_file_range(inode, 0, end, mode, 1);
-               truncate_inode_pages_range(inode->i_mapping, 0, end);
-       }

        cl_env_put(env, &refcheck);
        RETURN(rc);
 }
 
        cl_env_put(env, &refcheck);
        RETURN(rc);
 }
 

-void ll_lock_cancel_bits(struct ldlm_lock *lock, __u64 to_cancel)
+static void ll_lock_cancel_bits(struct ldlm_lock *lock, __u64 to_cancel)

 {
        struct inode *inode = ll_inode_from_resource_lock(lock);
        struct ll_inode_info *lli;
 {
        struct inode *inode = ll_inode_from_resource_lock(lock);
        struct ll_inode_info *lli;

diff --git a/lustre/lov/lov_object.c b/lustre/lov/lov_object.c
index ae535c0..cc041c0 100644 (file)
--- a/lustre/lov/lov_object.c
+++ b/lustre/lov/lov_object.c
@@ -76,6 +76,8 @@ struct lov_layout_operations {
                             struct cl_object *obj, struct cl_io *io);
         int  (*llo_getattr)(const struct lu_env *env, struct cl_object *obj,
                             struct cl_attr *attr);
                             struct cl_object *obj, struct cl_io *io);
         int  (*llo_getattr)(const struct lu_env *env, struct cl_object *obj,
                             struct cl_attr *attr);

+       int  (*llo_flush)(const struct lu_env *env, struct cl_object *obj,
+                         struct ldlm_lock *lock);

 };
 
 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov);
 };
 
 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov);


@@ -1035,6 +1037,22 @@ static int lov_attr_get_composite(const struct lu_env *env,
        RETURN(0);
 }
 
        RETURN(0);
 }
 

+static int lov_flush_composite(const struct lu_env *env,
+                              struct cl_object *obj,
+                              struct ldlm_lock *lock)
+{
+       struct lov_object *lov = cl2lov(obj);
+       struct lovsub_object *lovsub;
+
+       ENTRY;
+
+       if (!lsme_is_dom(lov->lo_lsm->lsm_entries[0]))
+               RETURN(-EINVAL);
+
+       lovsub = lov->u.composite.lo_entries[0].lle_dom.lo_dom;
+       RETURN(cl_object_flush(env, lovsub2cl(lovsub), lock));
+}
+

 const static struct lov_layout_operations lov_dispatch[] = {
        [LLT_EMPTY] = {
                .llo_init      = lov_init_empty,
 const static struct lov_layout_operations lov_dispatch[] = {
        [LLT_EMPTY] = {
                .llo_init      = lov_init_empty,


@@ -1065,6 +1083,7 @@ const static struct lov_layout_operations lov_dispatch[] = {
                .llo_lock_init = lov_lock_init_composite,
                .llo_io_init   = lov_io_init_composite,
                .llo_getattr   = lov_attr_get_composite,
                .llo_lock_init = lov_lock_init_composite,
                .llo_io_init   = lov_io_init_composite,
                .llo_getattr   = lov_attr_get_composite,

+               .llo_flush     = lov_flush_composite,

        },
        [LLT_FOREIGN] = {
                .llo_init      = lov_init_foreign,
        },
        [LLT_FOREIGN] = {
                .llo_init      = lov_init_foreign,


@@ -2078,6 +2097,12 @@ static loff_t lov_object_maxbytes(struct cl_object *obj)
        return maxbytes;
 }
 
        return maxbytes;
 }
 

+static int lov_object_flush(const struct lu_env *env, struct cl_object *obj,
+                           struct ldlm_lock *lock)
+{
+       return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_flush, env, obj, lock);
+}
+

 static const struct cl_object_operations lov_ops = {
        .coo_page_init    = lov_page_init,
        .coo_lock_init    = lov_lock_init,
 static const struct cl_object_operations lov_ops = {
        .coo_page_init    = lov_page_init,
        .coo_lock_init    = lov_lock_init,


@@ -2089,6 +2114,7 @@ static const struct cl_object_operations lov_ops = {
        .coo_layout_get   = lov_object_layout_get,
        .coo_maxbytes     = lov_object_maxbytes,
        .coo_fiemap       = lov_object_fiemap,
        .coo_layout_get   = lov_object_layout_get,
        .coo_maxbytes     = lov_object_maxbytes,
        .coo_fiemap       = lov_object_fiemap,

+       .coo_object_flush = lov_object_flush

 };
 
 static const struct lu_object_operations lov_lu_obj_ops = {
 };
 
 static const struct lu_object_operations lov_lu_obj_ops = {

diff --git a/lustre/mdc/mdc_dev.c b/lustre/mdc/mdc_dev.c
index f148f46..15e5f51 100644 (file)
--- a/lustre/mdc/mdc_dev.c
+++ b/lustre/mdc/mdc_dev.c
@@ -294,7 +294,7 @@ void mdc_lock_lockless_cancel(const struct lu_env *env,
  */
 static int mdc_dlm_blocking_ast0(const struct lu_env *env,
                                 struct ldlm_lock *dlmlock,
  */
 static int mdc_dlm_blocking_ast0(const struct lu_env *env,
                                 struct ldlm_lock *dlmlock,

-                                void *data, int flag)
+                                int flag)

 {
        struct cl_object *obj = NULL;
        int result = 0;
 {
        struct cl_object *obj = NULL;
        int result = 0;


@@ -386,7 +386,7 @@ int mdc_ldlm_blocking_ast(struct ldlm_lock *dlmlock,
                        break;
                }
 
                        break;
                }
 

-               rc = mdc_dlm_blocking_ast0(env, dlmlock, data, flag);
+               rc = mdc_dlm_blocking_ast0(env, dlmlock, flag);

                cl_env_put(env, &refcheck);
                break;
        }
                cl_env_put(env, &refcheck);
                break;
        }


@@ -1414,6 +1414,12 @@ int mdc_object_prune(const struct lu_env *env, struct cl_object *obj)
        return 0;
 }
 
        return 0;
 }
 

+static int mdc_object_flush(const struct lu_env *env, struct cl_object *obj,
+                           struct ldlm_lock *lock)
+{
+       RETURN(mdc_dlm_blocking_ast0(env, lock, LDLM_CB_CANCELING));
+}
+

 static const struct cl_object_operations mdc_ops = {
        .coo_page_init = osc_page_init,
        .coo_lock_init = mdc_lock_init,
 static const struct cl_object_operations mdc_ops = {
        .coo_page_init = osc_page_init,
        .coo_lock_init = mdc_lock_init,


@@ -1423,6 +1429,7 @@ static const struct cl_object_operations mdc_ops = {
        .coo_glimpse = osc_object_glimpse,
        .coo_req_attr_set = mdc_req_attr_set,
        .coo_prune = mdc_object_prune,
        .coo_glimpse = osc_object_glimpse,
        .coo_req_attr_set = mdc_req_attr_set,
        .coo_prune = mdc_object_prune,

+       .coo_object_flush = mdc_object_flush

 };
 
 static const struct osc_object_operations mdc_object_ops = {
 };
 
 static const struct osc_object_operations mdc_object_ops = {

diff --git a/lustre/obdclass/cl_object.c b/lustre/obdclass/cl_object.c
index 2914017..5aa59de 100644 (file)
--- a/lustre/obdclass/cl_object.c
+++ b/lustre/obdclass/cl_object.c
@@ -422,6 +422,24 @@ loff_t cl_object_maxbytes(struct cl_object *obj)
 }
 EXPORT_SYMBOL(cl_object_maxbytes);
 
 }
 EXPORT_SYMBOL(cl_object_maxbytes);
 

+int cl_object_flush(const struct lu_env *env, struct cl_object *obj,
+                        struct ldlm_lock *lock)
+{
+       struct lu_object_header *top = obj->co_lu.lo_header;
+       int rc = 0;
+       ENTRY;
+
+       list_for_each_entry(obj, &top->loh_layers, co_lu.lo_linkage) {
+               if (obj->co_ops->coo_object_flush) {
+                       rc = obj->co_ops->coo_object_flush(env, obj, lock);
+                       if (rc)
+                               break;
+               }
+       }
+       RETURN(rc);
+}
+EXPORT_SYMBOL(cl_object_flush);
+

 /**
  * Helper function removing all object locks, and marking object for
  * deletion. All object pages must have been deleted at this point.
 /**
  * Helper function removing all object locks, and marking object for
  * deletion. All object pages must have been deleted at this point.

diff --git a/lustre/tests/sanity-dom.sh b/lustre/tests/sanity-dom.sh
index 2ebccc3..41b29eb 100644 (file)
--- a/lustre/tests/sanity-dom.sh
+++ b/lustre/tests/sanity-dom.sh
@@ -106,6 +106,20 @@ test_4() {
 }
 run_test 4 "DoM: glimpse doesn't produce duplicated locks"
 
 }
 run_test 4 "DoM: glimpse doesn't produce duplicated locks"
 

+test_6() {
+       $MULTIOP $DIR1/$tfile Oz40960w100_z200w100c &
+       MULTIPID=$!
+
+       # let MULTIPID to create the file
+       sleep 1
+       $MULTIOP $DIR2/$tfile oO_RDWR:Tw100c
+       kill -USR1 $MULTIPID
+       wait
+       $MULTIOP $DIR2/$tfile oO_RDWR:z400w100c
+       $CHECKSTAT -s 500 $DIR2/$tfile || error "wrong size"
+}
+run_test 6 "Race two writes, check file size"
+

 test_fsx() {
        local file1=$DIR1/$tfile
        local file2=$DIR2/$tfile
 test_fsx() {
        local file1=$DIR1/$tfile
        local file2=$DIR2/$tfile