LU-11549 mdd: set LUSTRE_ORPHAN_FL for non-dirs

mdd_mark_orphan_object() sets LUSTRE_ORPHAN_FL only for
directories, which is not correct, causing the important bit of
orphan object state not transferring across OSP link and
allowing a distributed link operation to succeed
for an orphan source object , causing a dangling reference
on one mdt and an unconnected inode on another mdt.
mdd_open_sanity_check() conditions had to be relaxed in
case of replay.

Signed-off-by: Alexander Zarochentsev <c17826@cray.com>
Change-Id: If0d868b3de4d68406e1a3b371827f354566d3e42
Reviewed-on: https://review.whamcloud.com/35776
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Andrew Perepechko <c17827@cray.com>
Reviewed-by: Lai Siyao <lai.siyao@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/md_object.h b/lustre/include/md_object.h
index 763e54b..72e7719 100644 (file)
--- a/lustre/include/md_object.h
+++ b/lustre/include/md_object.h
@@ -239,8 +239,8 @@ struct md_object_operations {
                             enum changelog_rec_flags clf_flags,
                             struct md_device *m, const struct lu_fid *fid);
 
                             enum changelog_rec_flags clf_flags,
                             struct md_device *m, const struct lu_fid *fid);
 

-       int (*moo_open)(const struct lu_env *env,
-                       struct md_object *obj, u64 open_flags);
+       int (*moo_open)(const struct lu_env *env, struct md_object *obj,
+                       u64 open_flags, struct md_op_spec*);

 
        int (*moo_close)(const struct lu_env *env, struct md_object *obj,
                         struct md_attr *ma, u64 open_flags);
 
        int (*moo_close)(const struct lu_env *env, struct md_object *obj,
                         struct md_attr *ma, u64 open_flags);


@@ -512,10 +512,10 @@ static inline int mo_swap_layouts(const struct lu_env *env,
 }
 
 static inline int mo_open(const struct lu_env *env, struct md_object *m,
 }
 
 static inline int mo_open(const struct lu_env *env, struct md_object *m,

-                         u64 open_flags)
+                         u64 open_flags, struct md_op_spec *spec)

 {
        LASSERT(m->mo_ops->moo_open);
 {
        LASSERT(m->mo_ops->moo_open);

-       return m->mo_ops->moo_open(env, m, open_flags);
+       return m->mo_ops->moo_open(env, m, open_flags, spec);

 }
 
 static inline int mo_close(const struct lu_env *env, struct md_object *m,
 }
 
 static inline int mo_close(const struct lu_env *env, struct md_object *m,

diff --git a/lustre/mdd/mdd_dir.c b/lustre/mdd/mdd_dir.c
index 17613d6..7e71f4c 100644 (file)
--- a/lustre/mdd/mdd_dir.c
+++ b/lustre/mdd/mdd_dir.c
@@ -1456,9 +1456,6 @@ static int mdd_mark_orphan_object(const struct lu_env *env,
        struct lu_attr *attr = MDD_ENV_VAR(env, la_for_start);
        int rc;
 
        struct lu_attr *attr = MDD_ENV_VAR(env, la_for_start);
        int rc;
 

-       if (!S_ISDIR(mdd_object_type(obj)))
-               return 0;
-

        attr->la_valid = LA_FLAGS;
        attr->la_flags = LUSTRE_ORPHAN_FL;
 
        attr->la_valid = LA_FLAGS;
        attr->la_flags = LUSTRE_ORPHAN_FL;
 

diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index a67cedc..c993376 100644 (file)
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -2999,7 +2999,8 @@ static int accmode(const struct lu_env *env, const struct lu_attr *la,
 
 static int mdd_open_sanity_check(const struct lu_env *env,
                                 struct mdd_object *obj,
 
 static int mdd_open_sanity_check(const struct lu_env *env,
                                 struct mdd_object *obj,

-                                const struct lu_attr *attr, u64 open_flags)
+                                const struct lu_attr *attr, u64 open_flags,
+                                int is_replay)

 {
        int mode, rc;
        ENTRY;
 {
        int mode, rc;
        ENTRY;


@@ -3007,7 +3008,7 @@ static int mdd_open_sanity_check(const struct lu_env *env,
        /* EEXIST check, also opening of *open* orphans is allowed so we can
         * open-by-handle unlinked files
         */
        /* EEXIST check, also opening of *open* orphans is allowed so we can
         * open-by-handle unlinked files
         */

-       if (mdd_is_dead_obj(obj) &&
+       if (mdd_is_dead_obj(obj) && !is_replay &&

            likely(!(mdd_is_orphan_obj(obj) && obj->mod_count > 0)))
                RETURN(-ENOENT);
 
            likely(!(mdd_is_orphan_obj(obj) && obj->mod_count > 0)))
                RETURN(-ENOENT);
 


@@ -3042,7 +3043,7 @@ static int mdd_open_sanity_check(const struct lu_env *env,
 }
 
 static int mdd_open(const struct lu_env *env, struct md_object *obj,
 }
 
 static int mdd_open(const struct lu_env *env, struct md_object *obj,

-                   u64 open_flags)
+                   u64 open_flags, struct md_op_spec *spec)

 {
        struct mdd_object *mdd_obj = md2mdd_obj(obj);
        struct md_device *md_dev = lu2md_dev(mdd2lu_dev(mdo2mdd(obj)));
 {
        struct mdd_object *mdd_obj = md2mdd_obj(obj);
        struct md_device *md_dev = lu2md_dev(mdd2lu_dev(mdo2mdd(obj)));


@@ -3052,6 +3053,7 @@ static int mdd_open(const struct lu_env *env, struct md_object *obj,
        struct mdd_device *mdd = mdo2mdd(obj);
        enum changelog_rec_type type = CL_OPEN;
        int rc = 0;
        struct mdd_device *mdd = mdo2mdd(obj);
        enum changelog_rec_type type = CL_OPEN;
        int rc = 0;

+       ENTRY;

 
        mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
 
 
        mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
 


@@ -3059,7 +3061,8 @@ static int mdd_open(const struct lu_env *env, struct md_object *obj,
        if (rc != 0)
                GOTO(out, rc);
 
        if (rc != 0)
                GOTO(out, rc);
 

-       rc = mdd_open_sanity_check(env, mdd_obj, attr, open_flags);
+       rc = mdd_open_sanity_check(env, mdd_obj, attr, open_flags,
+                                  spec->no_create);

        if ((rc == -EACCES) && (mdd->mdd_cl.mc_mask & (1 << CL_DN_OPEN)))
                type = CL_DN_OPEN;
        else if (rc != 0)
        if ((rc == -EACCES) && (mdd->mdd_cl.mc_mask & (1 << CL_DN_OPEN)))
                type = CL_DN_OPEN;
        else if (rc != 0)

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index 8ef5b2c..bcb7c2f 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -394,7 +394,8 @@ static int mdt_mfd_open(struct mdt_thread_info *info, struct mdt_object *p,
                RETURN(rc);
 
        rc = mo_open(info->mti_env, mdt_object_child(o),
                RETURN(rc);
 
        rc = mo_open(info->mti_env, mdt_object_child(o),

-                    created ? open_flags | MDS_OPEN_CREATED : open_flags);
+                    created ? open_flags | MDS_OPEN_CREATED : open_flags,
+                    &info->mti_spec);

        if (rc != 0) {
                /* If we allow the client to chgrp (CFS_SETGRP_PERM), but the
                 * client does not know which suppgid should be sent to the MDS,
        if (rc != 0) {
                /* If we allow the client to chgrp (CFS_SETGRP_PERM), but the
                 * client does not know which suppgid should be sent to the MDS,


@@ -1702,7 +1703,7 @@ static struct mdt_object *mdt_orphan_open(struct mdt_thread_info *info,
                GOTO(out, rc);
        }
 
                GOTO(out, rc);
        }
 

-       rc = mo_open(env, mdt_object_child(obj), MDS_OPEN_CREATED);
+       rc = mo_open(env, mdt_object_child(obj), MDS_OPEN_CREATED, spec);

        if (rc < 0)
                CERROR("%s: cannot open volatile file "DFID", orphan "
                       "file will be left in PENDING directory until "
        if (rc < 0)
                CERROR("%s: cannot open volatile file "DFID", orphan "
                       "file will be left in PENDING directory until "