LU-5791 lfsck: use bottom device to locate object

For the LFSCK modification, if only updates single object, or the
objects to be updated reside on the same server, in spite of local
or remote, then try to locate the object(s) against the bottom (OSD
or OSP) device; otherwise, there will be some update(s) on the local
server, and others on remote server, then either locate the object(s)
against LOD device or use two transaction for the modification.

Similarly, the transaction handle will be created on the proper device
corresponding to the object(s).

This patch also fixes some memory leak issues caused by using wrong
device for remote modification, one of the reason for LU-6138.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: I09a60bed3bd49a193d57214c4252904cb4546ab2
Reviewed-on: http://review.whamcloud.com/13392
Tested-by: Jenkins
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: wangdi <di.wang@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/lfsck/lfsck_bookmark.c b/lustre/lfsck/lfsck_bookmark.c
index 0d1f6d0..15a0afd 100644 (file)
--- a/lustre/lfsck/lfsck_bookmark.c
+++ b/lustre/lfsck/lfsck_bookmark.c
@@ -100,6 +100,7 @@ int lfsck_bookmark_store(const struct lu_env *env, struct lfsck_instance *lfsck)
 {
        struct thandle    *handle;
        struct dt_object  *obj    = lfsck->li_bookmark_obj;
 {
        struct thandle    *handle;
        struct dt_object  *obj    = lfsck->li_bookmark_obj;

+       struct dt_device  *dev    = lfsck_obj2dev(obj);

        loff_t             pos    = 0;
        int                len    = sizeof(struct lfsck_bookmark);
        int                rc;
        loff_t             pos    = 0;
        int                len    = sizeof(struct lfsck_bookmark);
        int                rc;


@@ -107,7 +108,7 @@ int lfsck_bookmark_store(const struct lu_env *env, struct lfsck_instance *lfsck)
 
        lfsck_bookmark_cpu_to_le(&lfsck->li_bookmark_disk,
                                 &lfsck->li_bookmark_ram);
 
        lfsck_bookmark_cpu_to_le(&lfsck->li_bookmark_disk,
                                 &lfsck->li_bookmark_ram);

-       handle = dt_trans_create(env, lfsck->li_bottom);
+       handle = dt_trans_create(env, dev);

        if (IS_ERR(handle))
                GOTO(log, rc = PTR_ERR(handle));
 
        if (IS_ERR(handle))
                GOTO(log, rc = PTR_ERR(handle));
 


@@ -118,7 +119,7 @@ int lfsck_bookmark_store(const struct lu_env *env, struct lfsck_instance *lfsck)
        if (rc != 0)
                GOTO(out, rc);
 
        if (rc != 0)
                GOTO(out, rc);
 

-       rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(out, rc);
 
        if (rc != 0)
                GOTO(out, rc);
 


@@ -129,7 +130,7 @@ int lfsck_bookmark_store(const struct lu_env *env, struct lfsck_instance *lfsck)
        GOTO(out, rc);
 
 out:
        GOTO(out, rc);
 
 out:

-       dt_trans_stop(env, lfsck->li_bottom, handle);
+       dt_trans_stop(env, dev, handle);

 
 log:
        if (rc != 0)
 
 log:
        if (rc != 0)


@@ -167,7 +168,7 @@ int lfsck_bookmark_setup(const struct lu_env *env,
                RETURN(PTR_ERR(root));
 
        if (unlikely(!dt_try_as_dir(env, root))) {
                RETURN(PTR_ERR(root));
 
        if (unlikely(!dt_try_as_dir(env, root))) {

-               lu_object_put(env, &root->do_lu);
+               lfsck_object_put(env, root);

 
                RETURN(-ENOTDIR);
        }
 
                RETURN(-ENOTDIR);
        }


@@ -175,7 +176,7 @@ int lfsck_bookmark_setup(const struct lu_env *env,
        obj = local_file_find_or_create(env, lfsck->li_los, root,
                                        LFSCK_BOOKMARK,
                                        S_IFREG | S_IRUGO | S_IWUSR);
        obj = local_file_find_or_create(env, lfsck->li_los, root,
                                        LFSCK_BOOKMARK,
                                        S_IFREG | S_IRUGO | S_IWUSR);

-       lu_object_put(env, &root->do_lu);
+       lfsck_object_put(env, root);

        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 

diff --git a/lustre/lfsck/lfsck_engine.c b/lustre/lfsck/lfsck_engine.c
index 8eb6f02..0ed0b35 100644 (file)
--- a/lustre/lfsck/lfsck_engine.c
+++ b/lustre/lfsck/lfsck_engine.c
@@ -98,7 +98,7 @@ static int lfsck_update_lma(const struct lu_env *env,
 {
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
 {
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;

-       struct dt_device                *dt     = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct lustre_mdt_attrs         *lma    = &info->lti_lma;
        struct lu_buf                   *buf;
        struct thandle                  *th;
        struct lustre_mdt_attrs         *lma    = &info->lti_lma;
        struct lu_buf                   *buf;
        struct thandle                  *th;


@@ -129,7 +129,7 @@ static int lfsck_update_lma(const struct lu_env *env,
        }
        lustre_lma_swab(lma);
 
        }
        lustre_lma_swab(lma);
 

-       th = dt_trans_create(env, dt);
+       th = dt_trans_create(env, dev);

        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 


@@ -138,7 +138,7 @@ static int lfsck_update_lma(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dt, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -147,7 +147,7 @@ static int lfsck_update_lma(const struct lu_env *env,
        GOTO(stop, rc);
 
 stop:
        GOTO(stop, rc);
 
 stop:

-       dt_trans_stop(env, dt, th);
+       dt_trans_stop(env, dev, th);

        return rc;
 }
 
        return rc;
 }
 


@@ -194,9 +194,8 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
        struct lfsck_thread_info *info    = lfsck_env_info(env);
        struct lu_fid            *fid     = &info->lti_fid;
        struct lu_seq_range      *range   = &info->lti_range;
        struct lfsck_thread_info *info    = lfsck_env_info(env);
        struct lu_fid            *fid     = &info->lti_fid;
        struct lu_seq_range      *range   = &info->lti_range;

-       struct dt_device         *dev     = lfsck->li_bottom;
-       struct seq_server_site   *ss      = lu_site2seq(dev->dd_lu_dev.ld_site);
-       __u32                     idx     = lfsck_dev_idx(dev);
+       struct seq_server_site   *ss      = lfsck_dev_site(lfsck);
+       __u32                     idx     = lfsck_dev_idx(lfsck);

        int                       depth   = 0;
        int                       rc      = 0;
 
        int                       depth   = 0;
        int                       rc      = 0;
 


@@ -237,7 +236,7 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
                        return 1;
 
                if (obj == NULL) {
                        return 1;
 
                if (obj == NULL) {

-                       obj = lfsck_object_find(env, lfsck, fid);
+                       obj = lfsck_object_find_bottom(env, lfsck, fid);

                        if (IS_ERR(obj))
                                return PTR_ERR(obj);
 
                        if (IS_ERR(obj))
                                return PTR_ERR(obj);
 


@@ -327,25 +326,14 @@ static int lfsck_load_stripe_lmv(const struct lu_env *env,
                        RETURN(-ENOMEM);
                }
 
                        RETURN(-ENOMEM);
                }
 

-               /* Find the object against the bottom device. */
-               obj = lfsck_object_find_by_dev(env, lfsck->li_bottom,
-                                              lfsck_dto2fid(obj));
-               if (IS_ERR(obj)) {
-                       OBD_FREE_LARGE(lslr, sizeof(*lslr) * stripes);
-                       OBD_FREE_PTR(llmv);
-
-                       RETURN(PTR_ERR(obj));
-               }
-

                llmv->ll_stripes_allocated = stripes;
                llmv->ll_hash_type = LMV_HASH_TYPE_UNKNOWN;
                llmv->ll_lslr = lslr;
                llmv->ll_stripes_allocated = stripes;
                llmv->ll_hash_type = LMV_HASH_TYPE_UNKNOWN;
                llmv->ll_lslr = lslr;

-               lfsck->li_obj_dir = obj;

        } else {
                llmv->ll_lmv_slave = 1;
        } else {
                llmv->ll_lmv_slave = 1;

-               lfsck->li_obj_dir = lfsck_object_get(obj);

        }
 
        }
 

+       lfsck->li_obj_dir = lfsck_object_get(obj);

        llmv->ll_lmv = *lmv;
        atomic_set(&llmv->ll_ref, 1);
        lfsck->li_lmv = llmv;
        llmv->ll_lmv = *lmv;
        atomic_set(&llmv->ll_ref, 1);
        lfsck->li_lmv = llmv;


@@ -530,7 +518,7 @@ static int lfsck_prep(const struct lu_env *env, struct lfsck_instance *lfsck,
                GOTO(out, rc = 0);
 
        /* Find the directory for namespace-based traverse. */
                GOTO(out, rc = 0);
 
        /* Find the directory for namespace-based traverse. */

-       obj = lfsck_object_find(env, lfsck, &pos->lp_dir_parent);
+       obj = lfsck_object_find_bottom(env, lfsck, &pos->lp_dir_parent);

        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 


@@ -865,9 +853,8 @@ static int lfsck_master_oit_engine(const struct lu_env *env,
        struct lu_fid            *fid   = &info->lti_fid;
        struct lfsck_bookmark    *bk    = &lfsck->li_bookmark_ram;
        struct ptlrpc_thread     *thread = &lfsck->li_thread;
        struct lu_fid            *fid   = &info->lti_fid;
        struct lfsck_bookmark    *bk    = &lfsck->li_bookmark_ram;
        struct ptlrpc_thread     *thread = &lfsck->li_thread;

-       struct dt_device         *dev   = lfsck->li_bottom;
-       struct seq_server_site   *ss    = lu_site2seq(dev->dd_lu_dev.ld_site);
-       __u32                    idx    = lfsck_dev_idx(dev);
+       struct seq_server_site   *ss    = lfsck_dev_site(lfsck);
+       __u32                    idx    = lfsck_dev_idx(lfsck);

        int                      rc;
        ENTRY;
 
        int                      rc;
        ENTRY;
 


@@ -980,7 +967,7 @@ static int lfsck_master_oit_engine(const struct lu_env *env,
                        }
                }
 
                        }
                }
 

-               target = lfsck_object_find(env, lfsck, fid);
+               target = lfsck_object_find_bottom(env, lfsck, fid);

                if (IS_ERR(target)) {
                        CDEBUG(D_LFSCK, "%s: OIT scan failed at find target "
                               DFID", cookie "LPU64": rc = %d\n",
                if (IS_ERR(target)) {
                        CDEBUG(D_LFSCK, "%s: OIT scan failed at find target "
                               DFID", cookie "LPU64": rc = %d\n",


@@ -1195,7 +1182,6 @@ static int lfsck_assistant_query_others(const struct lu_env *env,
 
        lad->lad_touch_gen++;
        memset(lr, 0, sizeof(*lr));
 
        lad->lad_touch_gen++;
        memset(lr, 0, sizeof(*lr));

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);

        lr->lr_event = LE_QUERY;
        lr->lr_active = com->lc_type;
        laia->laia_com = com;
        lr->lr_event = LE_QUERY;
        lr->lr_active = com->lc_type;
        laia->laia_com = com;


@@ -1312,7 +1298,7 @@ static int lfsck_assistant_notify_others(const struct lu_env *env,
        if (set == NULL)
                RETURN(-ENOMEM);
 
        if (set == NULL)
                RETURN(-ENOMEM);
 

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_index = lfsck_dev_idx(lfsck);

        lr->lr_active = com->lc_type;
        laia->laia_com = com;
        laia->laia_lr = lr;
        lr->lr_active = com->lc_type;
        laia->laia_com = com;
        laia->laia_lr = lr;

diff --git a/lustre/lfsck/lfsck_internal.h b/lustre/lfsck/lfsck_internal.h
index 6094721..f21a7a7 100644 (file)
--- a/lustre/lfsck/lfsck_internal.h
+++ b/lustre/lfsck/lfsck_internal.h
@@ -947,15 +947,14 @@ int lfsck_namespace_repair_dirent(const struct lu_env *env,
                                  struct dt_object *child,
                                  const char *name, const char *name2,
                                  __u16 type, bool update, bool dec);
                                  struct dt_object *child,
                                  const char *name, const char *name2,
                                  __u16 type, bool update, bool dec);

-int lfsck_verify_linkea(const struct lu_env *env, struct dt_device *dev,
-                       struct dt_object *obj, const struct lu_name *cname,
-                       const struct lu_fid *pfid);
+int lfsck_verify_linkea(const struct lu_env *env, struct dt_object *obj,
+                       const struct lu_name *cname, const struct lu_fid *pfid);

 int lfsck_links_get_first(const struct lu_env *env, struct dt_object *obj,
                          char *name, struct lu_fid *pfid);
 int lfsck_update_name_entry(const struct lu_env *env,
                            struct lfsck_instance *lfsck,
 int lfsck_links_get_first(const struct lu_env *env, struct dt_object *obj,
                          char *name, struct lu_fid *pfid);
 int lfsck_update_name_entry(const struct lu_env *env,
                            struct lfsck_instance *lfsck,

-                           struct dt_object *parent, const char *name,
-                           const struct lu_fid *pfid, __u32 type);
+                           struct dt_object *dir, const char *name,
+                           const struct lu_fid *fid, __u32 type);

 int lfsck_namespace_setup(const struct lu_env *env,
                          struct lfsck_instance *lfsck);
 
 int lfsck_namespace_setup(const struct lu_env *env,
                          struct lfsck_instance *lfsck);
 


@@ -1016,7 +1015,7 @@ static inline bool name_is_dot_or_dotdot(const char *name, int namelen)
               (namelen == 1 || (namelen == 2 && name[1] == '.'));
 }
 
               (namelen == 1 || (namelen == 2 && name[1] == '.'));
 }
 

-static inline struct dt_device *lfsck_obj2dt_dev(struct dt_object *obj)
+static inline struct dt_device *lfsck_obj2dev(struct dt_object *obj)

 {
        return container_of0(obj->do_lu.lo_dev, struct dt_device, dd_lu_dev);
 }
 {
        return container_of0(obj->do_lu.lo_dev, struct dt_device, dd_lu_dev);
 }


@@ -1163,9 +1162,15 @@ static inline void lfsck_object_put(const struct lu_env *env,
        lu_object_put(env, &obj->do_lu);
 }
 
        lu_object_put(env, &obj->do_lu);
 }
 

-static inline u32 lfsck_dev_idx(struct dt_device *dev)
+static inline struct seq_server_site
+*lfsck_dev_site(struct lfsck_instance *lfsck)

 {
 {

-       return dev->dd_lu_dev.ld_site->ld_seq_site->ss_node_id;
+       return lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
+}
+
+static inline u32 lfsck_dev_idx(struct lfsck_instance *lfsck)
+{
+       return lfsck_dev_site(lfsck)->ss_node_id;

 }
 
 static inline struct dt_object *
 }
 
 static inline struct dt_object *


@@ -1185,13 +1190,6 @@ lfsck_object_find_by_dev(const struct lu_env *env, struct dt_device *dev,
        return lu2dt(lu_object_find_slice(env, dt2lu_dev(dev), fid, NULL));
 }
 
        return lu2dt(lu_object_find_slice(env, dt2lu_dev(dev), fid, NULL));
 }
 

-static inline struct dt_object *lfsck_object_find(const struct lu_env *env,
-                                                 struct lfsck_instance *lfsck,
-                                                 const struct lu_fid *fid)
-{
-       return lfsck_object_find_by_dev(env, lfsck->li_next, fid);
-}
-

 static inline struct dt_device *
 lfsck_find_dev_by_fid(const struct lu_env *env, struct lfsck_instance *lfsck,
                      const struct lu_fid *fid)
 static inline struct dt_device *
 lfsck_find_dev_by_fid(const struct lu_env *env, struct lfsck_instance *lfsck,
                      const struct lu_fid *fid)


@@ -1199,11 +1197,14 @@ lfsck_find_dev_by_fid(const struct lu_env *env, struct lfsck_instance *lfsck,
        struct dt_device *dev;
        int               idx;
 
        struct dt_device *dev;
        int               idx;
 

+       if (!lfsck->li_master)
+               return lfsck->li_bottom;
+

        idx = lfsck_find_mdt_idx_by_fid(env, lfsck, fid);
        if (idx < 0)
                return ERR_PTR(idx);
 
        idx = lfsck_find_mdt_idx_by_fid(env, lfsck, fid);
        if (idx < 0)
                return ERR_PTR(idx);
 

-       if (idx == lfsck_dev_idx(lfsck->li_bottom)) {
+       if (idx == lfsck_dev_idx(lfsck)) {

                dev = lfsck->li_bottom;
        } else {
                struct lfsck_tgt_desc *ltd;
                dev = lfsck->li_bottom;
        } else {
                struct lfsck_tgt_desc *ltd;


@@ -1245,6 +1246,23 @@ lfsck_object_find_bottom_nowait(const struct lu_env *env,
        return lfsck_object_find_by_dev_nowait(env, dev, fid);
 }
 
        return lfsck_object_find_by_dev_nowait(env, dev, fid);
 }
 

+static inline struct dt_object *
+lfsck_object_locate(struct dt_device *dev, struct dt_object *obj)
+{
+       if (lfsck_obj2dev(obj) == dev) {
+               return obj;
+       } else {
+               struct lu_object *lo;
+
+               lo = lu_object_locate(obj->do_lu.lo_header,
+                                     dev->dd_lu_dev.ld_type);
+               if (unlikely(lo == NULL))
+                       return ERR_PTR(-ENOENT);
+
+               return lu2dt(lo);
+       }
+}
+

 static inline struct lfsck_tgt_desc *lfsck_tgt_get(struct lfsck_tgt_descs *ltds,
                                                   __u32 index)
 {
 static inline struct lfsck_tgt_desc *lfsck_tgt_get(struct lfsck_tgt_descs *ltds,
                                                   __u32 index)
 {


@@ -1281,11 +1299,11 @@ static inline void lfsck_component_put(const struct lu_env *env,
                for (i = 0, lsto = &com->lc_sub_trace_objs[0];
                     i < LFSCK_STF_COUNT; i++, lsto++) {
                        if (lsto->lsto_obj != NULL)
                for (i = 0, lsto = &com->lc_sub_trace_objs[0];
                     i < LFSCK_STF_COUNT; i++, lsto++) {
                        if (lsto->lsto_obj != NULL)

-                               lu_object_put(env, &lsto->lsto_obj->do_lu);
+                               lfsck_object_put(env, lsto->lsto_obj);

                }
 
                if (com->lc_obj != NULL)
                }
 
                if (com->lc_obj != NULL)

-                       lu_object_put_nocache(env, &com->lc_obj->do_lu);
+                       lfsck_object_put(env, com->lc_obj);

                if (com->lc_file_ram != NULL)
                        OBD_FREE(com->lc_file_ram, com->lc_file_size);
                if (com->lc_file_disk != NULL)
                if (com->lc_file_ram != NULL)
                        OBD_FREE(com->lc_file_ram, com->lc_file_size);
                if (com->lc_file_disk != NULL)

diff --git a/lustre/lfsck/lfsck_layout.c b/lustre/lfsck/lfsck_layout.c
index 632da13..70d6932 100644 (file)
--- a/lustre/lfsck/lfsck_layout.c
+++ b/lustre/lfsck/lfsck_layout.c
@@ -92,7 +92,6 @@ struct lfsck_layout_object {
        struct lu_attr           llo_attr;
        atomic_t                 llo_ref;
        __u64                    llo_cookie;
        struct lu_attr           llo_attr;
        atomic_t                 llo_ref;
        __u64                    llo_cookie;

-       __u16                    llo_gen;

 };
 
 struct lfsck_layout_req {
 };
 
 struct lfsck_layout_req {


@@ -111,7 +110,7 @@ struct lfsck_layout_slave_async_args {
 
 static struct lfsck_layout_object *
 lfsck_layout_object_init(const struct lu_env *env, struct dt_object *obj,
 
 static struct lfsck_layout_object *
 lfsck_layout_object_init(const struct lu_env *env, struct dt_object *obj,

-                        __u64 cookie, __u16 gen)
+                        __u64 cookie)

 {
        struct lfsck_layout_object *llo;
        int                         rc;
 {
        struct lfsck_layout_object *llo;
        int                         rc;


@@ -128,9 +127,6 @@ lfsck_layout_object_init(const struct lu_env *env, struct dt_object *obj,
        }
 
        llo->llo_cookie = cookie;
        }
 
        llo->llo_cookie = cookie;

-       /* The gen can be used to check whether some others have changed the
-        * file layout after LFSCK pre-fetching but before real verification. */
-       llo->llo_gen = gen;

        atomic_set(&llo->llo_ref, 1);
 
        return llo;
        atomic_set(&llo->llo_ref, 1);
 
        return llo;


@@ -256,7 +252,7 @@ static void lfsck_layout_assistant_req_fini(const struct lu_env *env,
        struct lfsck_layout_req *llr =
                        container_of0(lar, struct lfsck_layout_req, llr_lar);
 
        struct lfsck_layout_req *llr =
                        container_of0(lar, struct lfsck_layout_req, llr_lar);
 

-       lu_object_put(env, &llr->llr_child->do_lu);
+       lfsck_object_put(env, llr->llr_child);

        lfsck_layout_object_put(env, llr->llr_parent);
        OBD_FREE_PTR(llr);
 }
        lfsck_layout_object_put(env, llr->llr_parent);
        OBD_FREE_PTR(llr);
 }


@@ -610,7 +606,7 @@ static int lfsck_rbtree_setup(const struct lu_env *env,
        struct dt_object                *obj;
 
        fid->f_seq = FID_SEQ_LAYOUT_RBTREE;
        struct dt_object                *obj;
 
        fid->f_seq = FID_SEQ_LAYOUT_RBTREE;

-       fid->f_oid = lfsck_dev_idx(dev);
+       fid->f_oid = lfsck_dev_idx(lfsck);

        fid->f_ver = 0;
        obj = dt_locate(env, dev, fid);
        if (IS_ERR(obj))
        fid->f_ver = 0;
        obj = dt_locate(env, dev, fid);
        if (IS_ERR(obj))


@@ -658,7 +654,7 @@ static void lfsck_rbtree_cleanup(const struct lu_env *env,
        }
 
        if (llsd->llsd_rb_obj != NULL) {
        }
 
        if (llsd->llsd_rb_obj != NULL) {

-               lu_object_put(env, &llsd->llsd_rb_obj->do_lu);
+               lfsck_object_put(env, llsd->llsd_rb_obj);

                llsd->llsd_rb_obj = NULL;
        }
 
                llsd->llsd_rb_obj = NULL;
        }
 


@@ -940,7 +936,7 @@ static int lfsck_layout_store(const struct lu_env *env,
        struct lfsck_layout     *lo_ram = com->lc_file_ram;
        struct lfsck_layout     *lo     = com->lc_file_disk;
        struct thandle          *th;
        struct lfsck_layout     *lo_ram = com->lc_file_ram;
        struct lfsck_layout     *lo     = com->lc_file_disk;
        struct thandle          *th;

-       struct dt_device        *dev    = lfsck->li_bottom;
+       struct dt_device        *dev    = lfsck_obj2dev(obj);

        cfs_bitmap_t            *bitmap = NULL;
        loff_t                   pos;
        ssize_t                  size   = com->lc_file_size;
        cfs_bitmap_t            *bitmap = NULL;
        loff_t                   pos;
        ssize_t                  size   = com->lc_file_size;


@@ -1022,10 +1018,11 @@ static int lfsck_layout_init(const struct lu_env *env,
        return rc;
 }
 
        return rc;
 }
 

-static int fid_is_for_ostobj(const struct lu_env *env, struct dt_device *dt,
+static int fid_is_for_ostobj(const struct lu_env *env,
+                            struct lfsck_instance *lfsck,

                             struct dt_object *obj, const struct lu_fid *fid)
 {
                             struct dt_object *obj, const struct lu_fid *fid)
 {

-       struct seq_server_site  *ss     = lu_site2seq(dt->dd_lu_dev.ld_site);
+       struct seq_server_site  *ss     = lfsck_dev_site(lfsck);

        struct lu_seq_range     *range  = &lfsck_env_info(env)->lti_range;
        struct lustre_mdt_attrs *lma;
        int                      rc;
        struct lu_seq_range     *range  = &lfsck_env_info(env)->lti_range;
        struct lustre_mdt_attrs *lma;
        int                      rc;


@@ -1094,7 +1091,7 @@ lfsck_layout_lastid_create(const struct lu_env *env,
        struct lu_attr           *la     = &info->lti_la;
        struct dt_object_format  *dof    = &info->lti_dof;
        struct lfsck_bookmark    *bk     = &lfsck->li_bookmark_ram;
        struct lu_attr           *la     = &info->lti_la;
        struct dt_object_format  *dof    = &info->lti_dof;
        struct lfsck_bookmark    *bk     = &lfsck->li_bookmark_ram;

-       struct dt_device         *dt     = lfsck->li_bottom;
+       struct dt_device         *dt     = lfsck_obj2dev(obj);

        struct thandle           *th;
        __u64                     lastid = 0;
        loff_t                    pos    = 0;
        struct thandle           *th;
        __u64                     lastid = 0;
        loff_t                    pos    = 0;


@@ -1107,6 +1104,7 @@ lfsck_layout_lastid_create(const struct lu_env *env,
        memset(la, 0, sizeof(*la));
        la->la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
        la->la_valid = LA_MODE | LA_UID | LA_GID;
        memset(la, 0, sizeof(*la));
        la->la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
        la->la_valid = LA_MODE | LA_UID | LA_GID;

+       memset(dof, 0, sizeof(*dof));

        dof->dof_type = dt_mode_to_dft(S_IFREG);
 
        th = dt_trans_create(env, dt);
        dof->dof_type = dt_mode_to_dft(S_IFREG);
 
        th = dt_trans_create(env, dt);


@@ -1281,7 +1279,7 @@ lfsck_layout_lastid_load(const struct lu_env *env,
        int                      rc;
        ENTRY;
 
        int                      rc;
        ENTRY;
 

-       lu_last_id_fid(fid, lls->lls_seq, lfsck_dev_idx(lfsck->li_bottom));
+       lu_last_id_fid(fid, lls->lls_seq, lfsck_dev_idx(lfsck));

        obj = dt_locate(env, lfsck->li_bottom, fid);
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
        obj = dt_locate(env, lfsck->li_bottom, fid);
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));


@@ -1500,6 +1498,7 @@ static int lfsck_layout_refill_lovea(const struct lu_env *env,
        int                      rc;
        __u32                    magic;
        __u16                    count;
        int                      rc;
        __u32                    magic;
        __u16                    count;

+       ENTRY;

 
        magic = le32_to_cpu(lmm->lmm_magic);
        count = le16_to_cpu(lmm->lmm_stripe_count);
 
        magic = le32_to_cpu(lmm->lmm_magic);
        count = le16_to_cpu(lmm->lmm_stripe_count);


@@ -1534,7 +1533,7 @@ static int lfsck_layout_refill_lovea(const struct lu_env *env,
        if (rc == 0)
                rc = 1;
 
        if (rc == 0)
                rc = 1;
 

-       return rc;
+       RETURN(rc);

 }
 
 /**
 }
 
 /**


@@ -1621,6 +1620,47 @@ static int lfsck_layout_extend_lovea(const struct lu_env *env,
        RETURN(rc);
 }
 
        RETURN(rc);
 }
 

+static int __lfsck_layout_update_pfid(const struct lu_env *env,
+                                     struct dt_object *child,
+                                     const struct lu_fid *pfid, __u32 offset)
+{
+       struct dt_device        *dev    = lfsck_obj2dev(child);
+       struct filter_fid       *ff     = &lfsck_env_info(env)->lti_new_pfid;
+       struct thandle          *handle;
+       struct lu_buf            buf    = { NULL };
+       int                      rc;
+
+       ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
+       ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
+       /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
+        * MDT-object's FID::f_ver, instead it is the OST-object index in its
+        * parent MDT-object's layout EA. */
+       ff->ff_parent.f_stripe_idx = cpu_to_le32(offset);
+       lfsck_buf_init(&buf, ff, sizeof(struct filter_fid));
+
+       handle = dt_trans_create(env, dev);
+       if (IS_ERR(handle))
+               RETURN(PTR_ERR(handle));
+
+       rc = dt_declare_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
+       if (rc != 0)
+               GOTO(stop, rc);
+
+       rc = dt_trans_start_local(env, dev, handle);
+       if (rc != 0)
+               GOTO(stop, rc);
+
+       rc = dt_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle,
+                         BYPASS_CAPA);
+
+       GOTO(stop, rc);
+
+stop:
+       dt_trans_stop(env, dev, handle);
+
+       return rc;
+}
+

 /**
  * \retval      +1: repaired
  * \retval       0: did nothing
 /**
  * \retval      +1: repaired
  * \retval       0: did nothing


@@ -1632,11 +1672,7 @@ static int lfsck_layout_update_pfid(const struct lu_env *env,
                                    struct lu_fid *cfid,
                                    struct dt_device *cdev, __u32 ea_off)
 {
                                    struct lu_fid *cfid,
                                    struct dt_device *cdev, __u32 ea_off)
 {

-       struct filter_fid       *pfid   = &lfsck_env_info(env)->lti_new_pfid;

        struct dt_object        *child;
        struct dt_object        *child;

-       struct thandle          *handle;
-       const struct lu_fid     *tfid   = lu_object_fid(&parent->do_lu);
-       struct lu_buf           *buf;

        int                      rc     = 0;
        ENTRY;
 
        int                      rc     = 0;
        ENTRY;
 


@@ -1644,38 +1680,11 @@ static int lfsck_layout_update_pfid(const struct lu_env *env,
        if (IS_ERR(child))
                RETURN(PTR_ERR(child));
 
        if (IS_ERR(child))
                RETURN(PTR_ERR(child));
 

-       handle = dt_trans_create(env, cdev);
-       if (IS_ERR(handle))
-               GOTO(out, rc = PTR_ERR(handle));
-
-       pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
-       pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
-       /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
-        * MDT-object's FID::f_ver, instead it is the OST-object index in its
-        * parent MDT-object's layout EA. */
-       pfid->ff_parent.f_stripe_idx = cpu_to_le32(ea_off);
-       buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
-
-       rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
-       if (rc != 0)
-               GOTO(stop, rc);
-
-       rc = dt_trans_start(env, cdev, handle);
-       if (rc != 0)
-               GOTO(stop, rc);
-
-       rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle,
-                         BYPASS_CAPA);
-
-       GOTO(stop, rc = (rc == 0 ? 1 : rc));
-
-stop:
-       dt_trans_stop(env, cdev, handle);
-
-out:
-       lu_object_put(env, &child->do_lu);
+       rc = __lfsck_layout_update_pfid(env, child,
+                                       lu_object_fid(&parent->do_lu), ea_off);
+       lfsck_object_put(env, child);

 
 

-       return rc;
+       RETURN(rc == 0 ? 1 : rc);

 }
 
 /**
 }
 
 /**


@@ -1750,11 +1759,11 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lu_fid                   *pfid   = &rec->lor_fid;
        struct lu_fid                   *tfid   = &info->lti_fid3;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lu_fid                   *pfid   = &rec->lor_fid;
        struct lu_fid                   *tfid   = &info->lti_fid3;

-       struct dt_device                *next   = lfsck->li_next;
+       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_object                *lpf    = lfsck->li_lpf_obj;

        struct dt_object                *pobj   = NULL;
        struct dt_object                *cobj   = NULL;
        struct thandle                  *th     = NULL;
        struct dt_object                *pobj   = NULL;
        struct dt_object                *cobj   = NULL;
        struct thandle                  *th     = NULL;

-       struct lu_buf                    pbuf   = { NULL };

        struct lu_buf                   *ea_buf = &info->lti_big_buf;
        struct lu_buf                    lov_buf;
        struct lustre_handle             lh     = { 0 };
        struct lu_buf                   *ea_buf = &info->lti_big_buf;
        struct lu_buf                    lov_buf;
        struct lustre_handle             lh     = { 0 };


@@ -1766,31 +1775,33 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
        int                              rc     = 0;
        ENTRY;
 
        int                              rc     = 0;
        ENTRY;
 

-       if (unlikely(lfsck->li_lpf_obj == NULL))
+       if (unlikely(lpf == NULL))

                GOTO(log, rc = -ENXIO);
 
                GOTO(log, rc = -ENXIO);
 

-       if (fid_is_zero(pfid)) {
-               struct filter_fid *ff = &info->lti_new_pfid;
+       /* We use two separated transactions to repair the inconsistency.
+        *
+        * 1) create the MDT-object locally.
+        * 2) update the OST-object's PFID EA if necessary.
+        *
+        * If 1) succeed, but 2) failed, then the OST-object's PFID EA will be
+        * updated when the layout LFSCK run next time.
+        *
+        * If 1) failed, but 2) succeed, then such MDT-object will be re-created
+        * when the layout LFSCK run next time. */

 
 

+       if (fid_is_zero(pfid)) {

                rc = lfsck_fid_alloc(env, lfsck, pfid, false);
                if (rc != 0)
                rc = lfsck_fid_alloc(env, lfsck, pfid, false);
                if (rc != 0)

-                       RETURN(rc);
+                       GOTO(log, rc);

 
 

-               ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
-               ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
-               /* Currently, the filter_fid::ff_parent::f_ver is not the
-                * real parent MDT-object's FID::f_ver, instead it is the
-                * OST-object index in its parent MDT-object's layout EA. */
-               ff->ff_parent.f_stripe_idx = cpu_to_le32(ea_off);
-               lfsck_buf_init(&pbuf, ff, sizeof(struct filter_fid));

                cobj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, cfid);
                if (IS_ERR(cobj))
                        GOTO(log, rc = PTR_ERR(cobj));
        }
 
                cobj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, cfid);
                if (IS_ERR(cobj))
                        GOTO(log, rc = PTR_ERR(cobj));
        }
 

-       pobj = lfsck_object_find_by_dev(env, lfsck->li_bottom, pfid);
+       pobj = lfsck_object_find_by_dev(env, dev, pfid);

        if (IS_ERR(pobj))
        if (IS_ERR(pobj))

-               GOTO(put, rc = PTR_ERR(pobj));
+               GOTO(log, rc = PTR_ERR(pobj));

 
        LASSERT(infix != NULL);
        LASSERT(type != NULL);
 
        LASSERT(infix != NULL);
        LASSERT(type != NULL);


@@ -1801,18 +1812,18 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
                rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
                               (const struct dt_key *)name, BYPASS_CAPA);
                if (rc != 0 && rc != -ENOENT)
                rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
                               (const struct dt_key *)name, BYPASS_CAPA);
                if (rc != 0 && rc != -ENOENT)

-                       GOTO(put, rc);
+                       GOTO(log, rc);

        } while (rc == 0);
 
        rc = linkea_data_new(&ldata,
                             &lfsck_env_info(env)->lti_linkea_buf);
        if (rc != 0)
        } while (rc == 0);
 
        rc = linkea_data_new(&ldata,
                             &lfsck_env_info(env)->lti_linkea_buf);
        if (rc != 0)

-               GOTO(put, rc);
+               GOTO(log, rc);

 
        pname = lfsck_name_get_const(env, name, strlen(name));
        rc = linkea_add_buf(&ldata, pname, lfsck_dto2fid(lfsck->li_lpf_obj));
        if (rc != 0)
 
        pname = lfsck_name_get_const(env, name, strlen(name));
        rc = linkea_add_buf(&ldata, pname, lfsck_dto2fid(lfsck->li_lpf_obj));
        if (rc != 0)

-               GOTO(put, rc);
+               GOTO(log, rc);

 
        memset(la, 0, sizeof(*la));
        la->la_uid = rec->lor_uid;
 
        memset(la, 0, sizeof(*la));
        la->la_uid = rec->lor_uid;


@@ -1822,12 +1833,15 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(S_IFREG);
 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(S_IFREG);

+       /* Because the dof->dof_reg.striped = 0, the LOD will not create
+        * the stripe(s). The LFSCK will specify the LOV EA via
+        * lfsck_layout_extend_lovea(). */

 
        size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
        if (ea_buf->lb_len < size) {
                lu_buf_realloc(ea_buf, size);
                if (ea_buf->lb_buf == NULL)
 
        size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
        if (ea_buf->lb_len < size) {
                lu_buf_realloc(ea_buf, size);
                if (ea_buf->lb_buf == NULL)

-                       GOTO(put, rc = -ENOMEM);
+                       GOTO(log, rc = -ENOMEM);

        }
 
        /* Hold update lock on the .lustre/lost+found/MDTxxxx/.
        }
 
        /* Hold update lock on the .lustre/lost+found/MDTxxxx/.


@@ -1836,48 +1850,34 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
         *      because creating MDT-object for orphan OST-object is rare, we
         *      do not much care about the performance. It can be improved in
         *      the future when needed. */
         *      because creating MDT-object for orphan OST-object is rare, we
         *      do not much care about the performance. It can be improved in
         *      the future when needed. */

-       rc = lfsck_ibits_lock(env, lfsck, lfsck->li_lpf_obj, &lh,
+       rc = lfsck_ibits_lock(env, lfsck, lpf, &lh,

                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)

-               GOTO(put, rc);
+               GOTO(log, rc);

 
 

-       th = dt_trans_create(env, next);
+       /* The 1st transaction. */
+       th = dt_trans_create(env, dev);

        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 

-       /* 1a. Update OST-object's parent information remotely.
-        *
-        * If other subsequent modifications failed, then next LFSCK scanning
-        * will process the OST-object as orphan again with known parent FID. */
-       if (cobj != NULL) {
-               rc = dt_declare_xattr_set(env, cobj, &pbuf, XATTR_NAME_FID,
-                                         0, th);
-               if (rc != 0)
-                       GOTO(stop, rc);
-       }
-
-       /* 2a. Create the MDT-object locally. */

        rc = dt_declare_create(env, pobj, la, NULL, dof, th);
        if (rc != 0)
                GOTO(stop, rc);
 
        rc = dt_declare_create(env, pobj, la, NULL, dof, th);
        if (rc != 0)
                GOTO(stop, rc);
 

-       /* 3a. Add layout EA for the MDT-object. */

        lfsck_buf_init(&lov_buf, ea_buf->lb_buf, size);
        rc = dt_declare_xattr_set(env, pobj, &lov_buf, XATTR_NAME_LOV,
        lfsck_buf_init(&lov_buf, ea_buf->lb_buf, size);
        rc = dt_declare_xattr_set(env, pobj, &lov_buf, XATTR_NAME_LOV,

-                                 LU_XATTR_CREATE, th);
+                                 LU_XATTR_REPLACE, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       /* 4a. Insert the MDT-object to .lustre/lost+found/MDTxxxx/ */

        dtrec->rec_fid = pfid;
        dtrec->rec_type = S_IFREG;
        dtrec->rec_fid = pfid;
        dtrec->rec_type = S_IFREG;

-       rc = dt_declare_insert(env, lfsck->li_lpf_obj,
+       rc = dt_declare_insert(env, lpf,

                               (const struct dt_rec *)dtrec,
                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 
                               (const struct dt_rec *)dtrec,
                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 

-       /* 5a. insert linkEA for parent. */

        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);
        rc = dt_declare_xattr_set(env, pobj, &linkea_buf,
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);
        rc = dt_declare_xattr_set(env, pobj, &linkea_buf,


@@ -1885,55 +1885,49 @@ static int lfsck_layout_recreate_parent(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, next, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       /* 1b. Update OST-object's parent information remotely. */
-       if (cobj != NULL) {
-               rc = dt_xattr_set(env, cobj, &pbuf, XATTR_NAME_FID, 0, th,
-                                 BYPASS_CAPA);
-               if (rc != 0)
-                       GOTO(stop, rc);
-       }
-

        dt_write_lock(env, pobj, 0);
        dt_write_lock(env, pobj, 0);

-       /* 2b. Create the MDT-object locally. */

        rc = dt_create(env, pobj, la, NULL, dof, th);
        if (rc == 0)
        rc = dt_create(env, pobj, la, NULL, dof, th);
        if (rc == 0)

-               /* 3b. Add layout EA for the MDT-object. */

                rc = lfsck_layout_extend_lovea(env, lfsck, th, pobj, cfid,
                rc = lfsck_layout_extend_lovea(env, lfsck, th, pobj, cfid,

-                                              &lov_buf, LU_XATTR_CREATE,
-                                              ltd->ltd_index, ea_off, false);
+                               &lov_buf, 0, ltd->ltd_index, ea_off, true);

        dt_write_unlock(env, pobj);
        if (rc < 0)
                GOTO(stop, rc);
 
        dt_write_unlock(env, pobj);
        if (rc < 0)
                GOTO(stop, rc);
 

-       /* 4b. Insert the MDT-object to .lustre/lost+found/MDTxxxx/ */
-       rc = dt_insert(env, lfsck->li_lpf_obj, (const struct dt_rec *)dtrec,
+       rc = dt_insert(env, lpf, (const struct dt_rec *)dtrec,

                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(stop, rc);
 
                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(stop, rc);
 

-       /* 5b. insert linkEA for parent. */

        rc = dt_xattr_set(env, pobj, &linkea_buf,
                          XATTR_NAME_LINK, 0, th, BYPASS_CAPA);
        rc = dt_xattr_set(env, pobj, &linkea_buf,
                          XATTR_NAME_LINK, 0, th, BYPASS_CAPA);

+       if (rc == 0 && cobj != NULL) {
+               dt_trans_stop(env, dev, th);
+               th = NULL;
+
+               /* The 2nd transaction. */
+               rc = __lfsck_layout_update_pfid(env, cobj, pfid, ea_off);
+       }

 
        GOTO(stop, rc);
 
 stop:
 
        GOTO(stop, rc);
 
 stop:

-       dt_trans_stop(env, next, th);
+       if (th != NULL)
+               dt_trans_stop(env, dev, th);

 
 unlock:
        lfsck_ibits_unlock(&lh, LCK_EX);
 
 
 unlock:
        lfsck_ibits_unlock(&lh, LCK_EX);
 

-put:
+log:

        if (cobj != NULL && !IS_ERR(cobj))
        if (cobj != NULL && !IS_ERR(cobj))

-               lu_object_put(env, &cobj->do_lu);
+               lfsck_object_put(env, cobj);

        if (pobj != NULL && !IS_ERR(pobj))
        if (pobj != NULL && !IS_ERR(pobj))

-               lu_object_put(env, &pobj->do_lu);
+               lfsck_object_put(env, pobj);

 
 

-log:

        if (rc < 0)
                CDEBUG(D_LFSCK, "%s layout LFSCK assistant failed to "
                       "recreate the lost MDT-object: parent "DFID
        if (rc < 0)
                CDEBUG(D_LFSCK, "%s layout LFSCK assistant failed to "
                       "recreate the lost MDT-object: parent "DFID


@@ -2098,7 +2092,7 @@ unlock:
        ldlm_lock_decref(&lh, LCK_EX);
 
 put:
        ldlm_lock_decref(&lh, LCK_EX);
 
 put:

-       lu_object_put(env, &obj->do_lu);
+       lfsck_object_put(env, obj);

 
        return rc;
 }
 
        return rc;
 }


@@ -2130,7 +2124,7 @@ static int lfsck_layout_conflict_create(const struct lu_env *env,
        struct lu_fid            *cfid2         = &info->lti_fid2;
        struct ost_id            *oi            = &info->lti_oi;
        struct lov_mds_md_v1     *lmm           = ea_buf->lb_buf;
        struct lu_fid            *cfid2         = &info->lti_fid2;
        struct ost_id            *oi            = &info->lti_oi;
        struct lov_mds_md_v1     *lmm           = ea_buf->lb_buf;

-       struct dt_device         *dev           = com->lc_lfsck->li_bottom;
+       struct dt_device         *dev           = lfsck_obj2dev(parent);

        struct thandle           *th            = NULL;
        struct lustre_handle      lh            = { 0 };
        __u32                     ost_idx2      = le32_to_cpu(slot->l_ost_idx);
        struct thandle           *th            = NULL;
        struct lustre_handle      lh            = { 0 };
        __u32                     ost_idx2      = le32_to_cpu(slot->l_ost_idx);


@@ -2228,7 +2222,7 @@ static int lfsck_layout_recreate_lovea(const struct lu_env *env,
        struct lu_fid            *fid           = &info->lti_fid2;
        struct ost_id            *oi            = &info->lti_oi;
        struct lfsck_instance    *lfsck         = com->lc_lfsck;
        struct lu_fid            *fid           = &info->lti_fid2;
        struct ost_id            *oi            = &info->lti_oi;
        struct lfsck_instance    *lfsck         = com->lc_lfsck;

-       struct dt_device         *dt            = lfsck->li_bottom;
+       struct dt_device         *dt            = lfsck_obj2dev(parent);

        struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
        struct thandle            *handle       = NULL;
        size_t                    lovea_size;
        struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
        struct thandle            *handle       = NULL;
        size_t                    lovea_size;


@@ -2522,7 +2516,7 @@ static int lfsck_layout_scan_orphan_one(const struct lu_env *env,
                GOTO(put, rc = -EXDEV);
 
        if (dt_object_exists(parent) == 0) {
                GOTO(put, rc = -EXDEV);
 
        if (dt_object_exists(parent) == 0) {

-               lu_object_put(env, &parent->do_lu);
+               lfsck_object_put(env, parent);

                rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
                                                  "", "R", ea_off);
                GOTO(out, rc);
                rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
                                                  "", "R", ea_off);
                GOTO(out, rc);


@@ -2538,7 +2532,7 @@ static int lfsck_layout_scan_orphan_one(const struct lu_env *env,
 
 put:
        if (rc <= 0)
 
 put:
        if (rc <= 0)

-               lu_object_put(env, &parent->do_lu);
+               lfsck_object_put(env, parent);

        else
                /* The layout EA is changed, need to be reloaded next time. */
                lu_object_put_nocache(env, &parent->do_lu);
        else
                /* The layout EA is changed, need to be reloaded next time. */
                lu_object_put_nocache(env, &parent->do_lu);


@@ -2653,7 +2647,7 @@ fini:
        iops->put(env, di);
        iops->fini(env, di);
 put:
        iops->put(env, di);
        iops->fini(env, di);
 put:

-       lu_object_put(env, &obj->do_lu);
+       lfsck_object_put(env, obj);

 
 log:
        CDEBUG(D_LFSCK, "%s: layout LFSCK assistant finished the orphan "
 
 log:
        CDEBUG(D_LFSCK, "%s: layout LFSCK assistant finished the orphan "


@@ -2680,10 +2674,10 @@ static int lfsck_layout_repair_dangling(const struct lu_env *env,
 {
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct filter_fid               *pfid   = &info->lti_new_pfid;
 {
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct filter_fid               *pfid   = &info->lti_new_pfid;

-       struct dt_allocation_hint       *hint   = &info->lti_hint;
+       struct dt_object_format         *dof    = &info->lti_dof;

        struct lu_attr                  *cla    = &info->lti_la2;
        struct dt_object                *child  = llr->llr_child;
        struct lu_attr                  *cla    = &info->lti_la2;
        struct dt_object                *child  = llr->llr_child;

-       struct dt_device                *dev    = lfsck_obj2dt_dev(child);
+       struct dt_device                *dev    = lfsck_obj2dev(child);

        const struct lu_fid             *tfid   = lu_object_fid(&parent->do_lu);
        struct thandle                  *handle;
        struct lu_buf                   *buf;
        const struct lu_fid             *tfid   = lu_object_fid(&parent->do_lu);
        struct thandle                  *handle;
        struct lu_buf                   *buf;


@@ -2706,6 +2700,7 @@ static int lfsck_layout_repair_dangling(const struct lu_env *env,
        cla->la_mode = S_IFREG | 0666;
        cla->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
                        LA_ATIME | LA_MTIME | LA_CTIME;
        cla->la_mode = S_IFREG | 0666;
        cla->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
                        LA_ATIME | LA_MTIME | LA_CTIME;

+       memset(dof, 0, sizeof(*dof));

 
        rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
                              MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
 
        rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
                              MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,


@@ -2713,12 +2708,6 @@ static int lfsck_layout_repair_dangling(const struct lu_env *env,
        if (rc != 0)
                GOTO(log, rc);
 
        if (rc != 0)
                GOTO(log, rc);
 

-       handle = dt_trans_create(env, dev);
-       if (IS_ERR(handle))
-               GOTO(unlock1, rc = PTR_ERR(handle));
-
-       hint->dah_parent = NULL;
-       hint->dah_mode = 0;

        pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
        pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
        /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
        pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
        pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
        /* Currently, the filter_fid::ff_parent::f_ver is not the real parent


@@ -2727,7 +2716,11 @@ static int lfsck_layout_repair_dangling(const struct lu_env *env,
        pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
        buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
 
        pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
        buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
 

-       rc = dt_declare_create(env, child, cla, hint, NULL, handle);
+       handle = dt_trans_create(env, dev);
+       if (IS_ERR(handle))
+               GOTO(unlock1, rc = PTR_ERR(handle));
+
+       rc = dt_declare_create(env, child, cla, NULL, dof, handle);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -2736,7 +2729,7 @@ static int lfsck_layout_repair_dangling(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, handle);
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -2744,7 +2737,7 @@ static int lfsck_layout_repair_dangling(const struct lu_env *env,
        if (unlikely(lfsck_is_dead_obj(parent)))
                GOTO(unlock2, rc = 1);
 
        if (unlikely(lfsck_is_dead_obj(parent)))
                GOTO(unlock2, rc = 1);
 

-       rc = dt_create(env, child, cla, hint, NULL, handle);
+       rc = dt_create(env, child, cla, NULL, dof, handle);

        if (rc != 0)
                GOTO(unlock2, rc);
 
        if (rc != 0)
                GOTO(unlock2, rc);
 


@@ -2788,7 +2781,7 @@ static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
        struct filter_fid               *pfid   = &info->lti_new_pfid;
        struct lu_attr                  *tla    = &info->lti_la3;
        struct dt_object                *child  = llr->llr_child;
        struct filter_fid               *pfid   = &info->lti_new_pfid;
        struct lu_attr                  *tla    = &info->lti_la3;
        struct dt_object                *child  = llr->llr_child;

-       struct dt_device                *dev    = lfsck_obj2dt_dev(child);
+       struct dt_device                *dev    = lfsck_obj2dev(child);

        const struct lu_fid             *tfid   = lu_object_fid(&parent->do_lu);
        struct thandle                  *handle;
        struct lu_buf                   *buf;
        const struct lu_fid             *tfid   = lu_object_fid(&parent->do_lu);
        struct thandle                  *handle;
        struct lu_buf                   *buf;


@@ -2802,10 +2795,6 @@ static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
        if (rc != 0)
                GOTO(log, rc);
 
        if (rc != 0)
                GOTO(log, rc);
 

-       handle = dt_trans_create(env, dev);
-       if (IS_ERR(handle))
-               GOTO(unlock1, rc = PTR_ERR(handle));
-

        pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
        pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
        /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
        pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
        pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
        /* Currently, the filter_fid::ff_parent::f_ver is not the real parent


@@ -2814,6 +2803,10 @@ static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
        pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
        buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
 
        pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
        buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
 

+       handle = dt_trans_create(env, dev);
+       if (IS_ERR(handle))
+               GOTO(unlock1, rc = PTR_ERR(handle));
+

        rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
        if (rc != 0)
                GOTO(stop, rc);
        rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
        if (rc != 0)
                GOTO(stop, rc);


@@ -2825,7 +2818,7 @@ static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, handle);
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -2881,117 +2874,156 @@ static int lfsck_layout_repair_multiple_references(const struct lu_env *env,
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct dt_allocation_hint       *hint   = &info->lti_hint;
        struct dt_object_format         *dof    = &info->lti_dof;
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct dt_allocation_hint       *hint   = &info->lti_hint;
        struct dt_object_format         *dof    = &info->lti_dof;

-       struct dt_device                *pdev   = com->lc_lfsck->li_next;

        struct ost_id                   *oi     = &info->lti_oi;
        struct ost_id                   *oi     = &info->lti_oi;

-       struct dt_device                *cdev   = lfsck_obj2dt_dev(llr->llr_child);
+       struct lfsck_instance           *lfsck  = com->lc_lfsck;
+       struct dt_device                *dev    = lfsck->li_bottom;
+       struct lu_device                *d      =
+                               &lfsck_obj2dev(llr->llr_child)->dd_lu_dev;
+       struct lu_object                *o;
+       struct lu_object                *n;

        struct dt_object                *child  = NULL;
        struct dt_object                *child  = NULL;

-       struct lu_device                *d      = &cdev->dd_lu_dev;
-       struct lu_object                *o      = NULL;
-       struct thandle                  *handle;
+       struct thandle                  *handle = NULL;

        struct lov_mds_md_v1            *lmm;
        struct lov_ost_data_v1          *objs;
        struct lov_mds_md_v1            *lmm;
        struct lov_ost_data_v1          *objs;

+       struct lu_fid                    tfid;

        struct lustre_handle             lh     = { 0 };
        struct lu_buf                    ea_buf;
        __u32                            magic;
        struct lustre_handle             lh     = { 0 };
        struct lu_buf                    ea_buf;
        __u32                            magic;

+       __u32                            index;

        int                              rc;
        ENTRY;
 
        int                              rc;
        ENTRY;
 

-       rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
-                             MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
-                             LCK_EX);
-       if (rc != 0)
-               GOTO(log, rc);
-
-       handle = dt_trans_create(env, pdev);
-       if (IS_ERR(handle))
-               GOTO(unlock1, rc = PTR_ERR(handle));
+       /* We use two separated transactions to repair the inconsistency.
+        *
+        * 1) create the child (OST-object).
+        * 2) update the parent LOV EA according to the child's FID.
+        *
+        * If 1) succeed, but 2) failed or aborted, then such OST-object will be
+        * handled as orphan when the layout LFSCK run next time.
+        *
+        * If 1) failed, but 2) succeed, then such OST-object will be re-created
+        * as dangling referened case when the layout LFSCK run next time. */

 
 

+       /* The 1st transaction. */

        o = lu_object_anon(env, d, NULL);
        if (IS_ERR(o))
        o = lu_object_anon(env, d, NULL);
        if (IS_ERR(o))

-               GOTO(stop, rc = PTR_ERR(o));
+               GOTO(log, rc = PTR_ERR(o));
+
+       n = lu_object_locate(o->lo_header, d->ld_type);
+       if (unlikely(n == NULL)) {
+               lu_object_put_nocache(env, o);

 
 

-       child = container_of(o, struct dt_object, do_lu);
-       o = lu_object_locate(o->lo_header, d->ld_type);
-       if (unlikely(o == NULL))
-               GOTO(stop, rc = -EINVAL);
+               GOTO(log, rc = -EINVAL);
+       }

 
 

-       child = container_of(o, struct dt_object, do_lu);
+       child = container_of(n, struct dt_object, do_lu);
+       memset(hint, 0, sizeof(*hint));

        la->la_valid = LA_UID | LA_GID;
        la->la_valid = LA_UID | LA_GID;

-       hint->dah_parent = NULL;
-       hint->dah_mode = 0;
-       dof->dof_type = DFT_REGULAR;
-       rc = dt_declare_create(env, child, la, NULL, NULL, handle);
+       memset(dof, 0, sizeof(*dof));
+
+       handle = dt_trans_create(env, dev);
+       if (IS_ERR(handle))
+               GOTO(log, rc = PTR_ERR(handle));
+
+       rc = dt_declare_create(env, child, la, hint, dof, handle);
+       if (rc != 0)
+               GOTO(stop, rc);
+
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

+       rc = dt_create(env, child, la, hint, dof, handle);
+       dt_trans_stop(env, dev, handle);
+       handle = NULL;
+       if (rc != 0)
+               GOTO(log, rc);
+
+       rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
+                             MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
+                             LCK_EX);
+       if (rc != 0)
+               GOTO(log, rc);
+
+       /* The 2nd transaction. */
+       handle = dt_trans_create(env, dev);
+       if (IS_ERR(handle))
+               GOTO(log, rc = PTR_ERR(handle));
+

        rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
                                  LU_XATTR_REPLACE, handle);
        if (rc != 0)
                GOTO(stop, rc);
 
        rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
                                  LU_XATTR_REPLACE, handle);
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, pdev, handle);
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(stop, rc);
 
        dt_write_lock(env, parent, 0);
        if (unlikely(lfsck_is_dead_obj(parent)))
        if (rc != 0)
                GOTO(stop, rc);
 
        dt_write_lock(env, parent, 0);
        if (unlikely(lfsck_is_dead_obj(parent)))

-               GOTO(unlock2, rc = 0);
+               GOTO(unlock, rc = 0);

 
        rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV, BYPASS_CAPA);
        if (unlikely(rc == 0 || rc == -ENODATA || rc == -ERANGE))
 
        rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV, BYPASS_CAPA);
        if (unlikely(rc == 0 || rc == -ENODATA || rc == -ERANGE))

-               GOTO(unlock2, rc = 0);
+               GOTO(unlock, rc = 0);

 
        lmm = buf->lb_buf;
 
        lmm = buf->lb_buf;

-       /* Someone change layout during the LFSCK, no need to repair then. */
-       if (le16_to_cpu(lmm->lmm_layout_gen) != llr->llr_parent->llo_gen)
-               GOTO(unlock2, rc = 0);
-
-       rc = dt_create(env, child, la, hint, dof, handle);
-       if (rc != 0)
-               GOTO(unlock2, rc);
-

        /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
         * been verified in lfsck_layout_verify_header() already. If some
         * new magic introduced in the future, then layout LFSCK needs to
         * be updated also. */
        magic = le32_to_cpu(lmm->lmm_magic);
        if (magic == LOV_MAGIC_V1) {
        /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
         * been verified in lfsck_layout_verify_header() already. If some
         * new magic introduced in the future, then layout LFSCK needs to
         * be updated also. */
        magic = le32_to_cpu(lmm->lmm_magic);
        if (magic == LOV_MAGIC_V1) {

-               objs = &lmm->lmm_objects[0];
+               objs = &lmm->lmm_objects[llr->llr_lov_idx];

        } else {
                LASSERT(magic == LOV_MAGIC_V3);
        } else {
                LASSERT(magic == LOV_MAGIC_V3);

-               objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
+               objs =
+               &((struct lov_mds_md_v3 *)lmm)->lmm_objects[llr->llr_lov_idx];

        }
 
        }
 

-       lmm->lmm_layout_gen = cpu_to_le16(llr->llr_parent->llo_gen + 1);
+       ostid_le_to_cpu(&objs->l_ost_oi, oi);
+       index = le32_to_cpu(objs->l_ost_idx);
+       rc = ostid_to_fid(&tfid, oi, index);
+       /* Someone changed layout during the LFSCK, no need to repair then. */
+       if (rc == 0 && !lu_fid_eq(&tfid, lu_object_fid(&llr->llr_child->do_lu)))
+               GOTO(unlock, rc = 0);
+
+       lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);

        fid_to_ostid(lu_object_fid(&child->do_lu), oi);
        fid_to_ostid(lu_object_fid(&child->do_lu), oi);

-       ostid_cpu_to_le(oi, &objs[llr->llr_lov_idx].l_ost_oi);
-       objs[llr->llr_lov_idx].l_ost_gen = cpu_to_le32(0);
-       objs[llr->llr_lov_idx].l_ost_idx = cpu_to_le32(llr->llr_ost_idx);
+       ostid_cpu_to_le(oi, &objs->l_ost_oi);
+       objs->l_ost_gen = cpu_to_le32(0);
+       objs->l_ost_idx = cpu_to_le32(llr->llr_ost_idx);

        lfsck_buf_init(&ea_buf, lmm,
                       lov_mds_md_size(le16_to_cpu(lmm->lmm_stripe_count),
                                       magic));
        rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV,
                          LU_XATTR_REPLACE, handle, BYPASS_CAPA);
 
        lfsck_buf_init(&ea_buf, lmm,
                       lov_mds_md_size(le16_to_cpu(lmm->lmm_stripe_count),
                                       magic));
        rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV,
                          LU_XATTR_REPLACE, handle, BYPASS_CAPA);
 

-       GOTO(unlock2, rc = (rc == 0 ? 1 : rc));
+       if (rc == 0) {
+               /* The @parent LOV EA has been updated, need to be re-loaded. */
+               set_bit(LU_OBJECT_HEARD_BANSHEE,
+                       &parent->do_lu.lo_header->loh_flags);
+               rc = 1;
+       }

 
 

-unlock2:
+       GOTO(unlock, rc);
+
+unlock:

        dt_write_unlock(env, parent);
 
 stop:
        dt_write_unlock(env, parent);
 
 stop:

-       if (child != NULL)
-               lu_object_put(env, &child->do_lu);
-
-       dt_trans_stop(env, pdev, handle);
+       if (handle != NULL)
+               dt_trans_stop(env, dev, handle);

 
 

-unlock1:
+log:

        lfsck_ibits_unlock(&lh, LCK_EX);
        lfsck_ibits_unlock(&lh, LCK_EX);

+       if (child != NULL)
+               lfsck_object_put(env, child);

 
 

-log:

        CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired multiple "
               "references for: parent "DFID", OST-index %u, stripe-index %u, "
               "owner %u/%u: rc = %d\n",
        CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired multiple "
               "references for: parent "DFID", OST-index %u, stripe-index %u, "
               "owner %u/%u: rc = %d\n",

-              lfsck_lfsck2name(com->lc_lfsck), PFID(lfsck_dto2fid(parent)),
+              lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(parent)),

               llr->llr_ost_idx, llr->llr_lov_idx, la->la_uid, la->la_gid, rc);
 
        return rc;
               llr->llr_ost_idx, llr->llr_lov_idx, la->la_uid, la->la_gid, rc);
 
        return rc;


@@ -3010,7 +3042,7 @@ static int lfsck_layout_repair_owner(const struct lu_env *env,
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct lu_attr                  *tla    = &info->lti_la3;
        struct dt_object                *child  = llr->llr_child;
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct lu_attr                  *tla    = &info->lti_la3;
        struct dt_object                *child  = llr->llr_child;

-       struct dt_device                *dev    = lfsck_obj2dt_dev(child);
+       struct dt_device                *dev    = lfsck_obj2dev(child);

        struct thandle                  *handle;
        int                              rc;
        ENTRY;
        struct thandle                  *handle;
        int                              rc;
        ENTRY;


@@ -3026,7 +3058,7 @@ static int lfsck_layout_repair_owner(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, handle);
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -3114,7 +3146,7 @@ static int lfsck_layout_check_parent(const struct lu_env *env,
                RETURN(LLIT_UNMATCHED_PAIR);
        }
 
                RETURN(LLIT_UNMATCHED_PAIR);
        }
 

-       tobj = lfsck_object_find(env, com->lc_lfsck, pfid);
+       tobj = lfsck_object_find_bottom(env, com->lc_lfsck, pfid);

        if (IS_ERR(tobj))
                RETURN(PTR_ERR(tobj));
 
        if (IS_ERR(tobj))
                RETURN(PTR_ERR(tobj));
 


@@ -3248,7 +3280,7 @@ static int lfsck_layout_assistant_handler_p1(const struct lu_env *env,
        int                                   rc;
        ENTRY;
 
        int                                   rc;
        ENTRY;
 

-       parent = lfsck_object_find(env, lfsck, &lar->lar_fid);
+       parent = lfsck_object_find_bottom(env, lfsck, &lar->lar_fid);

        if (IS_ERR(parent))
                RETURN(PTR_ERR(parent));
 
        if (IS_ERR(parent))
                RETURN(PTR_ERR(parent));
 


@@ -3375,7 +3407,7 @@ out:
        up_write(&com->lc_sem);
 
 put_parent:
        up_write(&com->lc_sem);
 
 put_parent:

-       lu_object_put(env, &parent->do_lu);
+       lfsck_object_put(env, parent);

 
        return rc;
 }
 
        return rc;
 }


@@ -3549,7 +3581,6 @@ lfsck_layout_slave_query_master(const struct lu_env *env,
                GOTO(log, rc = -ENOMEM);
 
        memset(lr, 0, sizeof(*lr));
                GOTO(log, rc = -ENOMEM);
 
        memset(lr, 0, sizeof(*lr));

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);

        lr->lr_event = LE_QUERY;
        lr->lr_active = LFSCK_TYPE_LAYOUT;
 
        lr->lr_event = LE_QUERY;
        lr->lr_active = LFSCK_TYPE_LAYOUT;
 


@@ -3630,7 +3661,7 @@ lfsck_layout_slave_notify_master(const struct lu_env *env,
        lr->lr_event = event;
        lr->lr_flags = LEF_FROM_OST;
        lr->lr_status = result;
        lr->lr_event = event;
        lr->lr_flags = LEF_FROM_OST;
        lr->lr_status = result;

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_index = lfsck_dev_idx(lfsck);

        lr->lr_active = LFSCK_TYPE_LAYOUT;
        lr->lr_flags2 = lo->ll_flags;
        llsd->llsd_touch_gen++;
        lr->lr_active = LFSCK_TYPE_LAYOUT;
        lr->lr_flags2 = lo->ll_flags;
        llsd->llsd_touch_gen++;


@@ -3700,7 +3731,7 @@ static int lfsck_layout_master_check_pairs(const struct lu_env *env,
        ENTRY;
 
        pfid->f_ver = 0;
        ENTRY;
 
        pfid->f_ver = 0;

-       obj = lfsck_object_find_by_dev(env, com->lc_lfsck->li_bottom, pfid);
+       obj = lfsck_object_find_bottom(env, com->lc_lfsck, pfid);

        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 
        if (IS_ERR(obj))
                RETURN(PTR_ERR(obj));
 


@@ -3750,7 +3781,7 @@ static int lfsck_layout_master_check_pairs(const struct lu_env *env,
 
 unlock:
        dt_read_unlock(env, obj);
 
 unlock:
        dt_read_unlock(env, obj);

-       lu_object_put(env, &obj->do_lu);
+       lfsck_object_put(env, obj);

 
        return rc;
 }
 
        return rc;
 }


@@ -3773,8 +3804,7 @@ static int lfsck_layout_slave_check_pairs(const struct lu_env *env,
 {
        struct lfsck_instance    *lfsck  = com->lc_lfsck;
        struct obd_device        *obd    = lfsck->li_obd;
 {
        struct lfsck_instance    *lfsck  = com->lc_lfsck;
        struct obd_device        *obd    = lfsck->li_obd;

-       struct seq_server_site   *ss     =
-                       lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
+       struct seq_server_site   *ss     = lfsck_dev_site(lfsck);

        struct obd_export        *exp    = NULL;
        struct ptlrpc_request    *req    = NULL;
        struct lfsck_request     *lr;
        struct obd_export        *exp    = NULL;
        struct ptlrpc_request    *req    = NULL;
        struct lfsck_request     *lr;


@@ -3838,48 +3868,27 @@ static int lfsck_layout_slave_repair_pfid(const struct lu_env *env,
                                          struct lfsck_component *com,
                                          struct lfsck_request *lr)
 {
                                          struct lfsck_component *com,
                                          struct lfsck_request *lr)
 {

-       struct lfsck_thread_info        *info   = lfsck_env_info(env);
-       struct filter_fid               *ff     = &info->lti_new_pfid;
-       struct lu_buf                   *buf;
-       struct dt_device                *dev    = com->lc_lfsck->li_bottom;
-       struct dt_object                *obj;
-       struct thandle                  *th     = NULL;
-       int                              rc     = 0;
+       struct dt_object        *obj;
+       int                      rc     = 0;

        ENTRY;
 
        ENTRY;
 

-       obj = lfsck_object_find_by_dev(env, dev, &lr->lr_fid);
+       obj = lfsck_object_find_bottom(env, com->lc_lfsck, &lr->lr_fid);

        if (IS_ERR(obj))
                GOTO(log, rc = PTR_ERR(obj));
 
        if (IS_ERR(obj))
                GOTO(log, rc = PTR_ERR(obj));
 

-       fid_cpu_to_le(&ff->ff_parent, &lr->lr_fid2);
-       buf = lfsck_buf_get(env, ff, sizeof(*ff));

        dt_write_lock(env, obj, 0);
        if (unlikely(dt_object_exists(obj) == 0 ||
                     lfsck_is_dead_obj(obj)))
                GOTO(unlock, rc = 0);
 
        dt_write_lock(env, obj, 0);
        if (unlikely(dt_object_exists(obj) == 0 ||
                     lfsck_is_dead_obj(obj)))
                GOTO(unlock, rc = 0);
 

-       th = dt_trans_create(env, dev);
-       if (IS_ERR(th))
-               GOTO(unlock, rc = PTR_ERR(th));
-
-       rc = dt_declare_xattr_set(env, obj, buf, XATTR_NAME_FID, 0, th);
-       if (rc != 0)
-               GOTO(stop, rc);
-
-       rc = dt_trans_start_local(env, dev, th);
-       if (rc != 0)
-               GOTO(stop, rc);
-
-       rc = dt_xattr_set(env, obj, buf, XATTR_NAME_FID, 0, th, BYPASS_CAPA);
+       rc = __lfsck_layout_update_pfid(env, obj, &lr->lr_fid2,
+                                       lr->lr_fid2.f_ver);

 
 

-       GOTO(stop, rc);
-
-stop:
-       dt_trans_stop(env, dev, th);
+       GOTO(unlock, rc);

 
 unlock:
        dt_write_unlock(env, obj);
 
 unlock:
        dt_write_unlock(env, obj);

-       lu_object_put(env, &obj->do_lu);
+       lfsck_object_put(env, obj);

 
 log:
        CDEBUG(D_LFSCK, "%s: layout LFSCK slave repaired pfid for "DFID
 
 log:
        CDEBUG(D_LFSCK, "%s: layout LFSCK slave repaired pfid for "DFID


@@ -4179,13 +4188,11 @@ static int lfsck_layout_scan_stripes(const struct lu_env *env,
        int                              i;
        __u32                            magic;
        __u16                            count;
        int                              i;
        __u32                            magic;
        __u16                            count;

-       __u16                            gen;

        ENTRY;
 
        lfsck_buf_init(&buf, &info->lti_old_pfid,
                       sizeof(struct filter_fid_old));
        count = le16_to_cpu(lmm->lmm_stripe_count);
        ENTRY;
 
        lfsck_buf_init(&buf, &info->lti_old_pfid,
                       sizeof(struct filter_fid_old));
        count = le16_to_cpu(lmm->lmm_stripe_count);

-       gen = le16_to_cpu(lmm->lmm_layout_gen);

        /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
         * been verified in lfsck_layout_verify_header() already. If some
         * new magic introduced in the future, then layout LFSCK needs to
        /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
         * been verified in lfsck_layout_verify_header() already. If some
         * new magic introduced in the future, then layout LFSCK needs to


@@ -4296,7 +4303,7 @@ static int lfsck_layout_scan_stripes(const struct lu_env *env,
 
                if (llo == NULL) {
                        llo = lfsck_layout_object_init(env, parent,
 
                if (llo == NULL) {
                        llo = lfsck_layout_object_init(env, parent,

-                               lfsck->li_pos_current.lp_oit_cookie, gen);
+                               lfsck->li_pos_current.lp_oit_cookie);

                        if (IS_ERR(llo)) {
                                rc = PTR_ERR(llo);
                                goto next;
                        if (IS_ERR(llo)) {
                                rc = PTR_ERR(llo);
                                goto next;


@@ -4337,7 +4344,7 @@ next:
                up_write(&com->lc_sem);
 
                if (cobj != NULL && !IS_ERR(cobj))
                up_write(&com->lc_sem);
 
                if (cobj != NULL && !IS_ERR(cobj))

-                       lu_object_put(env, &cobj->do_lu);
+                       lfsck_object_put(env, cobj);

 
                if (likely(tgt != NULL))
                        lfsck_tgt_put(tgt);
 
                if (likely(tgt != NULL))
                        lfsck_tgt_put(tgt);


@@ -4376,7 +4383,7 @@ static int lfsck_layout_master_exec_oit(const struct lu_env *env,
        struct thandle                  *handle = NULL;
        struct lu_buf                   *buf    = &info->lti_big_buf;
        struct lov_mds_md_v1            *lmm    = NULL;
        struct thandle                  *handle = NULL;
        struct lu_buf                   *buf    = &info->lti_big_buf;
        struct lov_mds_md_v1            *lmm    = NULL;

-       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct lustre_handle             lh     = { 0 };
        struct lu_buf                    ea_buf = { NULL };
        int                              rc     = 0;
        struct lustre_handle             lh     = { 0 };
        struct lu_buf                    ea_buf = { NULL };
        int                              rc     = 0;


@@ -4514,7 +4521,7 @@ static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
        LASSERT(llsd != NULL);
 
        if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY5) &&
        LASSERT(llsd != NULL);
 
        if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY5) &&

-           cfs_fail_val == lfsck_dev_idx(lfsck->li_bottom)) {
+           cfs_fail_val == lfsck_dev_idx(lfsck)) {

                struct l_wait_info       lwi = LWI_TIMEOUT(cfs_time_seconds(1),
                                                           NULL, NULL);
                struct ptlrpc_thread    *thread = &lfsck->li_thread;
                struct l_wait_info       lwi = LWI_TIMEOUT(cfs_time_seconds(1),
                                                           NULL, NULL);
                struct ptlrpc_thread    *thread = &lfsck->li_thread;


@@ -4530,7 +4537,7 @@ static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
        if (fid_is_idif(fid))
                seq = 0;
        else if (!fid_is_norm(fid) ||
        if (fid_is_idif(fid))
                seq = 0;
        else if (!fid_is_norm(fid) ||

-                !fid_is_for_ostobj(env, lfsck->li_next, obj, fid))
+                !fid_is_for_ostobj(env, lfsck, obj, fid))

                GOTO(unlock, rc = 0);
        else
                seq = fid_seq(fid);
                GOTO(unlock, rc = 0);
        else
                seq = fid_seq(fid);


@@ -5610,7 +5617,7 @@ int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
 
 out:
        if (root != NULL && !IS_ERR(root))
 
 out:
        if (root != NULL && !IS_ERR(root))

-               lu_object_put(env, &root->do_lu);
+               lfsck_object_put(env, root);

 
        if (rc != 0) {
                lfsck_component_cleanup(env, com);
 
        if (rc != 0) {
                lfsck_component_cleanup(env, com);


@@ -5648,7 +5655,7 @@ static int lfsck_fid_match_idx(const struct lu_env *env,
                return 0;
        }
 
                return 0;
        }
 

-       ss = lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
+       ss = lfsck_dev_site(lfsck);

        if (unlikely(ss == NULL))
                return -ENOTCONN;
 
        if (unlikely(ss == NULL))
                return -ENOTCONN;
 


@@ -5670,11 +5677,11 @@ static int lfsck_fid_match_idx(const struct lu_env *env,
 }
 
 static void lfsck_layout_destroy_orphan(const struct lu_env *env,
 }
 
 static void lfsck_layout_destroy_orphan(const struct lu_env *env,

-                                       struct dt_device *dev,

                                        struct dt_object *obj)
 {
                                        struct dt_object *obj)
 {

-       struct thandle *handle;
-       int             rc;
+       struct dt_device        *dev    = lfsck_obj2dev(obj);
+       struct thandle          *handle;
+       int                      rc;

        ENTRY;
 
        handle = dt_trans_create(env, dev);
        ENTRY;
 
        handle = dt_trans_create(env, dev);


@@ -5977,7 +5984,7 @@ again1:
        }
 
        key->f_oid = lrn->lrn_first_oid + pos;
        }
 
        key->f_oid = lrn->lrn_first_oid + pos;

-       obj = lfsck_object_find(env, lfsck, key);
+       obj = lfsck_object_find_bottom(env, lfsck, key);

        if (IS_ERR(obj)) {
                rc = PTR_ERR(obj);
                if (rc == -ENOENT) {
        if (IS_ERR(obj)) {
                rc = PTR_ERR(obj);
                if (rc == -ENOENT) {


@@ -6014,9 +6021,7 @@ again1:
                         * OST-object there. Destroy it now! */
                        if (unlikely(!(la->la_mode & S_ISUID))) {
                                dt_read_unlock(env, obj);
                         * OST-object there. Destroy it now! */
                        if (unlikely(!(la->la_mode & S_ISUID))) {
                                dt_read_unlock(env, obj);

-                               lfsck_layout_destroy_orphan(env,
-                                                           lfsck->li_bottom,
-                                                           obj);
+                               lfsck_layout_destroy_orphan(env, obj);

                                lfsck_object_put(env, obj);
                                pos++;
                                goto again1;
                                lfsck_object_put(env, obj);
                                pos++;
                                goto again1;

diff --git a/lustre/lfsck/lfsck_lib.c b/lustre/lfsck/lfsck_lib.c
index 95deeac..b64354d 100644 (file)
--- a/lustre/lfsck/lfsck_lib.c
+++ b/lustre/lfsck/lfsck_lib.c
@@ -442,8 +442,7 @@ int lfsck_find_mdt_idx_by_fid(const struct lu_env *env,
                              struct lfsck_instance *lfsck,
                              const struct lu_fid *fid)
 {
                              struct lfsck_instance *lfsck,
                              const struct lu_fid *fid)
 {

-       struct seq_server_site  *ss     =
-                       lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
+       struct seq_server_site  *ss     = lfsck_dev_site(lfsck);

        struct lu_seq_range     *range  = &lfsck_env_info(env)->lti_range;
        int                      rc;
 
        struct lu_seq_range     *range  = &lfsck_env_info(env)->lti_range;
        int                      rc;
 


@@ -479,7 +478,7 @@ static int lfsck_lpf_remove_name_entry(const struct lu_env *env,
                                       const char *name)
 {
        struct dt_object        *parent = lfsck->li_lpf_root_obj;
                                       const char *name)
 {
        struct dt_object        *parent = lfsck->li_lpf_root_obj;

-       struct dt_device        *dev    = lfsck->li_next;
+       struct dt_device        *dev    = lfsck_obj2dev(parent);

        struct thandle          *th;
        struct lustre_handle     lh     = { 0 };
        int                      rc;
        struct thandle          *th;
        struct lustre_handle     lh     = { 0 };
        int                      rc;


@@ -502,7 +501,7 @@ static int lfsck_lpf_remove_name_entry(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -538,7 +537,7 @@ static int lfsck_create_lpf_local(const struct lu_env *env,
 {
        struct dt_insert_rec    *rec    = &lfsck_env_info(env)->lti_dt_rec;
        struct dt_object        *parent = lfsck->li_lpf_root_obj;
 {
        struct dt_insert_rec    *rec    = &lfsck_env_info(env)->lti_dt_rec;
        struct dt_object        *parent = lfsck->li_lpf_root_obj;

-       struct dt_device        *dev    = lfsck->li_bottom;
+       struct dt_device        *dev    = lfsck_obj2dev(child);

        struct lfsck_bookmark   *bk     = &lfsck->li_bookmark_ram;
        struct dt_object        *bk_obj = lfsck->li_bookmark_obj;
        const struct lu_fid     *cfid   = lfsck_dto2fid(child);
        struct lfsck_bookmark   *bk     = &lfsck->li_bookmark_ram;
        struct dt_object        *bk_obj = lfsck->li_bookmark_obj;
        const struct lu_fid     *cfid   = lfsck_dto2fid(child);


@@ -727,7 +726,7 @@ static int lfsck_create_lpf_remote(const struct lu_env *env,
 
        /* Transaction I: locally */
 
 
        /* Transaction I: locally */
 

-       dev = lfsck->li_bottom;
+       dev = lfsck_obj2dev(child);

        th = dt_trans_create(env, dev);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
        th = dt_trans_create(env, dev);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));


@@ -809,11 +808,12 @@ static int lfsck_create_lpf_remote(const struct lu_env *env,
 
        /* Transaction II: remotely */
 
 
        /* Transaction II: remotely */
 

-       dev = lfsck->li_next;
+       dev = lfsck_obj2dev(parent);

        th = dt_trans_create(env, dev);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 
        th = dt_trans_create(env, dev);
        if (IS_ERR(th))
                RETURN(PTR_ERR(th));
 

+       th->th_sync = 1;

        /* 5a. insert name into parent dir */
        rec->rec_fid = cfid;
        rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
        /* 5a. insert name into parent dir */
        rec->rec_fid = cfid;
        rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,


@@ -826,7 +826,7 @@ static int lfsck_create_lpf_remote(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -848,7 +848,7 @@ unlock:
 stop:
        dt_trans_stop(env, dev, th);
 
 stop:
        dt_trans_stop(env, dev, th);
 

-       if (rc != 0 && dev == lfsck->li_next)
+       if (rc != 0 && dev == lfsck_obj2dev(parent))

                CDEBUG(D_LFSCK, "%s: partially created the object "DFID
                       "for orphans, but failed to insert the name %s "
                       "to the .lustre/lost+found/. Such inconsistency "
                CDEBUG(D_LFSCK, "%s: partially created the object "DFID
                       "for orphans, but failed to insert the name %s "
                       "to the .lustre/lost+found/. Such inconsistency "


@@ -884,7 +884,7 @@ static int lfsck_create_lpf(const struct lu_env *env,
        struct dt_object         *child = NULL;
        struct lustre_handle      lh    = { 0 };
        char                      name[8];
        struct dt_object         *child = NULL;
        struct lustre_handle      lh    = { 0 };
        char                      name[8];

-       int                       node  = lfsck_dev_idx(lfsck->li_bottom);
+       int                       node  = lfsck_dev_idx(lfsck);

        int                       rc    = 0;
        ENTRY;
 
        int                       rc    = 0;
        ENTRY;
 


@@ -920,7 +920,7 @@ static int lfsck_create_lpf(const struct lu_env *env,
                *cfid = bk->lb_lpf_fid;
        }
 
                *cfid = bk->lb_lpf_fid;
        }
 

-       child = lfsck_object_find_by_dev(env, lfsck->li_bottom, cfid);
+       child = lfsck_object_find_bottom(env, lfsck, cfid);

        if (IS_ERR(child))
                GOTO(unlock, rc = PTR_ERR(child));
 
        if (IS_ERR(child))
                GOTO(unlock, rc = PTR_ERR(child));
 


@@ -953,7 +953,7 @@ static int lfsck_create_lpf(const struct lu_env *env,
 unlock:
        lfsck_ibits_unlock(&lh, LCK_EX);
        if (rc != 0 && child != NULL && !IS_ERR(child))
 unlock:
        lfsck_ibits_unlock(&lh, LCK_EX);
        if (rc != 0 && child != NULL && !IS_ERR(child))

-               lu_object_put(env, &child->do_lu);
+               lfsck_object_put(env, child);

 
        return rc;
 }
 
        return rc;
 }


@@ -1129,8 +1129,7 @@ static int lfsck_verify_lpf_pairs(const struct lu_env *env,
                }
 
                cname = lfsck_name_get_const(env, name, strlen(name));
                }
 
                cname = lfsck_name_get_const(env, name, strlen(name));

-               rc = lfsck_verify_linkea(env, lfsck->li_bottom, child, cname,
-                                        &LU_LPF_FID);
+               rc = lfsck_verify_linkea(env, child, cname, &LU_LPF_FID);

                if (rc == 0)
                        rc = lfsck_update_lpf_entry(env, lfsck, parent, child,
                                                    name, type);
                if (rc == 0)
                        rc = lfsck_update_lpf_entry(env, lfsck, parent, child,
                                                    name, type);


@@ -1138,18 +1137,18 @@ static int lfsck_verify_lpf_pairs(const struct lu_env *env,
                GOTO(out_done, rc);
        }
 
                GOTO(out_done, rc);
        }
 

-       parent2 = lfsck_object_find_by_dev(env, lfsck->li_next, fid);
+       parent2 = lfsck_object_find_bottom(env, lfsck, fid);

        if (IS_ERR(parent2))
                GOTO(linkea, parent2);
 
        if (!dt_object_exists(parent2)) {
        if (IS_ERR(parent2))
                GOTO(linkea, parent2);
 
        if (!dt_object_exists(parent2)) {

-               lu_object_put(env, &parent2->do_lu);
+               lfsck_object_put(env, parent2);

 
                GOTO(linkea, parent2 = ERR_PTR(-ENOENT));
        }
 
        if (!dt_try_as_dir(env, parent2)) {
 
                GOTO(linkea, parent2 = ERR_PTR(-ENOENT));
        }
 
        if (!dt_try_as_dir(env, parent2)) {

-               lu_object_put(env, &parent2->do_lu);
+               lfsck_object_put(env, parent2);

 
                GOTO(linkea, parent2 = ERR_PTR(-ENOTDIR));
        }
 
                GOTO(linkea, parent2 = ERR_PTR(-ENOTDIR));
        }


@@ -1234,7 +1233,7 @@ linkea:
 
 out_put:
        if (parent2 != NULL && !IS_ERR(parent2))
 
 out_put:
        if (parent2 != NULL && !IS_ERR(parent2))

-               lu_object_put(env, &parent2->do_lu);
+               lfsck_object_put(env, parent2);

 
 out_done:
        return rc;
 
 out_done:
        return rc;


@@ -1266,10 +1265,9 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
        struct dt_object         *child1 = NULL;
        /* child2's FID is in the name entry MDTxxxx. */
        struct dt_object         *child2 = NULL;
        struct dt_object         *child1 = NULL;
        /* child2's FID is in the name entry MDTxxxx. */
        struct dt_object         *child2 = NULL;

-       struct dt_device         *dev    = lfsck->li_bottom;

        const struct lu_name     *cname;
        char                      name[8];
        const struct lu_name     *cname;
        char                      name[8];

-       int                       node   = lfsck_dev_idx(dev);
+       int                       node   = lfsck_dev_idx(lfsck);

        int                       rc     = 0;
        ENTRY;
 
        int                       rc     = 0;
        ENTRY;
 


@@ -1279,7 +1277,8 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
                RETURN(0);
 
        if (node == 0) {
                RETURN(0);
 
        if (node == 0) {

-               parent = lfsck_object_find_by_dev(env, dev, &LU_LPF_FID);
+               parent = lfsck_object_find_by_dev(env, lfsck->li_bottom,
+                                                 &LU_LPF_FID);

        } else {
                struct lfsck_tgt_desc *ltd;
 
        } else {
                struct lfsck_tgt_desc *ltd;
 


@@ -1298,7 +1297,7 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
        LASSERT(dt_object_exists(parent));
 
        if (unlikely(!dt_try_as_dir(env, parent))) {
        LASSERT(dt_object_exists(parent));
 
        if (unlikely(!dt_try_as_dir(env, parent))) {

-               lu_object_put(env, &parent->do_lu);
+               lfsck_object_put(env, parent);

 
                GOTO(put, rc = -ENOTDIR);
        }
 
                GOTO(put, rc = -ENOTDIR);
        }


@@ -1327,7 +1326,7 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
                        if (rc != 0)
                                GOTO(put, rc);
                } else {
                        if (rc != 0)
                                GOTO(put, rc);
                } else {

-                       child1 = lfsck_object_find_by_dev(env, dev,
+                       child1 = lfsck_object_find_bottom(env, lfsck,

                                                          &bk->lb_lpf_fid);
                        if (IS_ERR(child1)) {
                                child1 = NULL;
                                                          &bk->lb_lpf_fid);
                        if (IS_ERR(child1)) {
                                child1 = NULL;


@@ -1350,7 +1349,7 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
                                if (rc != 0)
                                        GOTO(put, rc);
 
                                if (rc != 0)
                                        GOTO(put, rc);
 

-                               lu_object_put(env, &child1->do_lu);
+                               lfsck_object_put(env, child1);

                                child1 = NULL;
                        } else if (unlikely(!dt_try_as_dir(env, child1))) {
                                GOTO(put, rc = -ENOTDIR);
                                child1 = NULL;
                        } else if (unlikely(!dt_try_as_dir(env, child1))) {
                                GOTO(put, rc = -ENOTDIR);


@@ -1381,7 +1380,7 @@ find_child2:
                goto check_child1;
        }
 
                goto check_child1;
        }
 

-       child2 = lfsck_object_find_by_dev(env, dev, cfid);
+       child2 = lfsck_object_find_bottom(env, lfsck, cfid);

        if (IS_ERR(child2))
                GOTO(put, rc = PTR_ERR(child2));
 
        if (IS_ERR(child2))
                GOTO(put, rc = PTR_ERR(child2));
 


@@ -1415,7 +1414,7 @@ find_child2:
                }
 
                cname = lfsck_name_get_const(env, name, strlen(name));
                }
 
                cname = lfsck_name_get_const(env, name, strlen(name));

-               rc = lfsck_verify_linkea(env, dev, child2, cname, &LU_LPF_FID);
+               rc = lfsck_verify_linkea(env, child2, cname, &LU_LPF_FID);

        }
 
        GOTO(put, rc);
        }
 
        GOTO(put, rc);


@@ -1430,7 +1429,7 @@ check_child1:
 put:
        if (lfsck->li_lpf_obj != NULL) {
                if (unlikely(!dt_try_as_dir(env, lfsck->li_lpf_obj))) {
 put:
        if (lfsck->li_lpf_obj != NULL) {
                if (unlikely(!dt_try_as_dir(env, lfsck->li_lpf_obj))) {

-                       lu_object_put(env, &lfsck->li_lpf_obj->do_lu);
+                       lfsck_object_put(env, lfsck->li_lpf_obj);

                        lfsck->li_lpf_obj = NULL;
                        rc = -ENOTDIR;
                }
                        lfsck->li_lpf_obj = NULL;
                        rc = -ENOTDIR;
                }


@@ -1439,9 +1438,9 @@ put:
        }
 
        if (child2 != NULL && !IS_ERR(child2))
        }
 
        if (child2 != NULL && !IS_ERR(child2))

-               lu_object_put(env, &child2->do_lu);
+               lfsck_object_put(env, child2);

        if (child1 != NULL && !IS_ERR(child1))
        if (child1 != NULL && !IS_ERR(child1))

-               lu_object_put(env, &child1->do_lu);
+               lfsck_object_put(env, child1);

 
        return rc;
 }
 
        return rc;
 }


@@ -1449,12 +1448,11 @@ put:
 static int lfsck_fid_init(struct lfsck_instance *lfsck)
 {
        struct lfsck_bookmark   *bk     = &lfsck->li_bookmark_ram;
 static int lfsck_fid_init(struct lfsck_instance *lfsck)
 {
        struct lfsck_bookmark   *bk     = &lfsck->li_bookmark_ram;

-       struct seq_server_site  *ss;
+       struct seq_server_site  *ss     = lfsck_dev_site(lfsck);

        char                    *prefix;
        int                      rc     = 0;
        ENTRY;
 
        char                    *prefix;
        int                      rc     = 0;
        ENTRY;
 

-       ss = lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);

        if (unlikely(ss == NULL))
                RETURN(-ENXIO);
 
        if (unlikely(ss == NULL))
                RETURN(-ENXIO);
 


@@ -1509,7 +1507,7 @@ void lfsck_instance_cleanup(const struct lu_env *env,
        LASSERT(thread_is_init(thread) || thread_is_stopped(thread));
 
        if (lfsck->li_obj_oit != NULL) {
        LASSERT(thread_is_init(thread) || thread_is_stopped(thread));
 
        if (lfsck->li_obj_oit != NULL) {

-               lu_object_put_nocache(env, &lfsck->li_obj_oit->do_lu);
+               lfsck_object_put(env, lfsck->li_obj_oit);

                lfsck->li_obj_oit = NULL;
        }
 
                lfsck->li_obj_oit = NULL;
        }
 


@@ -1545,22 +1543,22 @@ void lfsck_instance_cleanup(const struct lu_env *env,
        lfsck_tgt_descs_fini(&lfsck->li_mdt_descs);
 
        if (lfsck->li_lfsck_dir != NULL) {
        lfsck_tgt_descs_fini(&lfsck->li_mdt_descs);
 
        if (lfsck->li_lfsck_dir != NULL) {

-               lu_object_put_nocache(env, &lfsck->li_lfsck_dir->do_lu);
+               lfsck_object_put(env, lfsck->li_lfsck_dir);

                lfsck->li_lfsck_dir = NULL;
        }
 
        if (lfsck->li_bookmark_obj != NULL) {
                lfsck->li_lfsck_dir = NULL;
        }
 
        if (lfsck->li_bookmark_obj != NULL) {

-               lu_object_put_nocache(env, &lfsck->li_bookmark_obj->do_lu);
+               lfsck_object_put(env, lfsck->li_bookmark_obj);

                lfsck->li_bookmark_obj = NULL;
        }
 
        if (lfsck->li_lpf_obj != NULL) {
                lfsck->li_bookmark_obj = NULL;
        }
 
        if (lfsck->li_lpf_obj != NULL) {

-               lu_object_put(env, &lfsck->li_lpf_obj->do_lu);
+               lfsck_object_put(env, lfsck->li_lpf_obj);

                lfsck->li_lpf_obj = NULL;
        }
 
        if (lfsck->li_lpf_root_obj != NULL) {
                lfsck->li_lpf_obj = NULL;
        }
 
        if (lfsck->li_lpf_root_obj != NULL) {

-               lu_object_put(env, &lfsck->li_lpf_root_obj->do_lu);
+               lfsck_object_put(env, lfsck->li_lpf_root_obj);

                lfsck->li_lpf_root_obj = NULL;
        }
 
                lfsck->li_lpf_root_obj = NULL;
        }
 


@@ -2125,7 +2123,7 @@ static int lfsck_stop_notify(const struct lu_env *env,
                spin_unlock(&ltds->ltd_lock);
 
                memset(lr, 0, sizeof(*lr));
                spin_unlock(&ltds->ltd_lock);
 
                memset(lr, 0, sizeof(*lr));

-               lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+               lr->lr_index = lfsck_dev_idx(lfsck);

                lr->lr_event = LE_PEER_EXIT;
                lr->lr_active = type;
                lr->lr_status = LS_CO_PAUSED;
                lr->lr_event = LE_PEER_EXIT;
                lr->lr_active = type;
                lr->lr_status = LS_CO_PAUSED;


@@ -2535,7 +2533,7 @@ static int lfsck_stop_all(const struct lu_env *env,
 
        memset(lr, 0, sizeof(*lr));
        lr->lr_event = LE_STOP;
 
        memset(lr, 0, sizeof(*lr));
        lr->lr_event = LE_STOP;

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_index = lfsck_dev_idx(lfsck);

        lr->lr_status = stop->ls_status;
        lr->lr_version = bk->lb_version;
        lr->lr_active = LFSCK_TYPES_ALL;
        lr->lr_status = stop->ls_status;
        lr->lr_version = bk->lb_version;
        lr->lr_active = LFSCK_TYPES_ALL;


@@ -2605,7 +2603,7 @@ static int lfsck_start_all(const struct lu_env *env,
 
        memset(lr, 0, sizeof(*lr));
        lr->lr_event = LE_START;
 
        memset(lr, 0, sizeof(*lr));
        lr->lr_event = LE_START;

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_index = lfsck_dev_idx(lfsck);

        lr->lr_speed = bk->lb_speed_limit;
        lr->lr_version = bk->lb_version;
        lr->lr_active = start->ls_active;
        lr->lr_speed = bk->lb_speed_limit;
        lr->lr_version = bk->lb_version;
        lr->lr_active = start->ls_active;


@@ -3149,7 +3147,7 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
        lfsck->li_local_root_fid = *fid;
        if (master) {
                lfsck->li_master = 1;
        lfsck->li_local_root_fid = *fid;
        if (master) {
                lfsck->li_master = 1;

-               if (lfsck_dev_idx(key) == 0) {
+               if (lfsck_dev_idx(lfsck) == 0) {

                        struct lu_fid *pfid = &lfsck_env_info(env)->lti_fid2;
                        const struct lu_name *cname;
 
                        struct lu_fid *pfid = &lfsck_env_info(env)->lti_fid2;
                        const struct lu_name *cname;
 


@@ -3171,14 +3169,14 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
                        if (rc != 0)
                                GOTO(out, rc);
 
                        if (rc != 0)
                                GOTO(out, rc);
 

-                       lu_object_put(env, &obj->do_lu);
+                       lfsck_object_put(env, obj);

                        obj = dt_locate(env, key, fid);
                        if (IS_ERR(obj))
                                GOTO(out, rc = PTR_ERR(obj));
 
                        cname = lfsck_name_get_const(env, dotlustre,
                                                     strlen(dotlustre));
                        obj = dt_locate(env, key, fid);
                        if (IS_ERR(obj))
                                GOTO(out, rc = PTR_ERR(obj));
 
                        cname = lfsck_name_get_const(env, dotlustre,
                                                     strlen(dotlustre));

-                       rc = lfsck_verify_linkea(env, key, obj, cname,
+                       rc = lfsck_verify_linkea(env, obj, cname,

                                                 &lfsck->li_global_root_fid);
                        if (rc != 0)
                                GOTO(out, rc);
                                                 &lfsck->li_global_root_fid);
                        if (rc != 0)
                                GOTO(out, rc);


@@ -3193,18 +3191,18 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
                        if (rc != 0)
                                GOTO(out, rc);
 
                        if (rc != 0)
                                GOTO(out, rc);
 

-                       lu_object_put(env, &obj->do_lu);
+                       lfsck_object_put(env, obj);

                        obj = dt_locate(env, key, fid);
                        if (IS_ERR(obj))
                                GOTO(out, rc = PTR_ERR(obj));
 
                        cname = lfsck_name_get_const(env, lostfound,
                                                     strlen(lostfound));
                        obj = dt_locate(env, key, fid);
                        if (IS_ERR(obj))
                                GOTO(out, rc = PTR_ERR(obj));
 
                        cname = lfsck_name_get_const(env, lostfound,
                                                     strlen(lostfound));

-                       rc = lfsck_verify_linkea(env, key, obj, cname, pfid);
+                       rc = lfsck_verify_linkea(env, obj, cname, pfid);

                        if (rc != 0)
                                GOTO(out, rc);
 
                        if (rc != 0)
                                GOTO(out, rc);
 

-                       lu_object_put(env, &obj->do_lu);
+                       lfsck_object_put(env, obj);

                        obj = NULL;
                }
        }
                        obj = NULL;
                }
        }


@@ -3253,9 +3251,9 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
                rc = lfsck_add_target_from_orphan(env, lfsck);
 out:
        if (obj != NULL && !IS_ERR(obj))
                rc = lfsck_add_target_from_orphan(env, lfsck);
 out:
        if (obj != NULL && !IS_ERR(obj))

-               lu_object_put(env, &obj->do_lu);
+               lfsck_object_put(env, obj);

        if (root != NULL && !IS_ERR(root))
        if (root != NULL && !IS_ERR(root))

-               lu_object_put(env, &root->do_lu);
+               lfsck_object_put(env, root);

        if (rc != 0)
                lfsck_instance_cleanup(env, lfsck);
        return rc;
        if (rc != 0)
                lfsck_instance_cleanup(env, lfsck);
        return rc;

diff --git a/lustre/lfsck/lfsck_namespace.c b/lustre/lfsck/lfsck_namespace.c
index 72b05d5..73358a5 100644 (file)
--- a/lustre/lfsck/lfsck_namespace.c
+++ b/lustre/lfsck/lfsck_namespace.c
@@ -386,6 +386,7 @@ static int lfsck_namespace_store(const struct lu_env *env,
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct lfsck_assistant_data     *lad    = com->lc_data;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct lfsck_assistant_data     *lad    = com->lc_data;

+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        cfs_bitmap_t                    *bitmap = NULL;
        struct thandle                  *handle;
        __u32                            nbits  = 0;
        cfs_bitmap_t                    *bitmap = NULL;
        struct thandle                  *handle;
        __u32                            nbits  = 0;


@@ -407,7 +408,7 @@ static int lfsck_namespace_store(const struct lu_env *env,
        ns->ln_bitmap_size = nbits;
        lfsck_namespace_cpu_to_le((struct lfsck_namespace *)com->lc_file_disk,
                                  ns);
        ns->ln_bitmap_size = nbits;
        lfsck_namespace_cpu_to_le((struct lfsck_namespace *)com->lc_file_disk,
                                  ns);

-       handle = dt_trans_create(env, lfsck->li_bottom);
+       handle = dt_trans_create(env, dev);

        if (IS_ERR(handle))
                GOTO(log, rc = PTR_ERR(handle));
 
        if (IS_ERR(handle))
                GOTO(log, rc = PTR_ERR(handle));
 


@@ -439,7 +440,7 @@ static int lfsck_namespace_store(const struct lu_env *env,
        }
 #endif
 
        }
 #endif
 

-       rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
+       rc = dt_trans_start_local(env, dev, handle);

        if (rc != 0)
                GOTO(out, rc);
 
        if (rc != 0)
                GOTO(out, rc);
 


@@ -462,7 +463,7 @@ static int lfsck_namespace_store(const struct lu_env *env,
        GOTO(out, rc);
 
 out:
        GOTO(out, rc);
 
 out:

-       dt_trans_stop(env, lfsck->li_bottom, handle);
+       dt_trans_stop(env, dev, handle);

 
 log:
        if (rc != 0)
 
 log:
        if (rc != 0)


@@ -516,7 +517,7 @@ static int lfsck_namespace_load_sub_trace_files(const struct lu_env *env,
                        if (!reset)
                                continue;
 
                        if (!reset)
                                continue;
 

-                       lu_object_put(env, &lsto->lsto_obj->do_lu);
+                       lfsck_object_put(env, lsto->lsto_obj);

                        lsto->lsto_obj = NULL;
                }
 
                        lsto->lsto_obj = NULL;
                }
 


@@ -573,7 +574,7 @@ int lfsck_namespace_trace_update(const struct lu_env *env,
        struct lfsck_instance   *lfsck  = com->lc_lfsck;
        struct dt_object        *obj;
        struct lu_fid           *key    = &lfsck_env_info(env)->lti_fid3;
        struct lfsck_instance   *lfsck  = com->lc_lfsck;
        struct dt_object        *obj;
        struct lu_fid           *key    = &lfsck_env_info(env)->lti_fid3;

-       struct dt_device        *dev    = lfsck->li_bottom;
+       struct dt_device        *dev;

        struct thandle          *th     = NULL;
        int                      idx;
        int                      rc     = 0;
        struct thandle          *th     = NULL;
        int                      idx;
        int                      rc     = 0;


@@ -588,6 +589,7 @@ int lfsck_namespace_trace_update(const struct lu_env *env,
 
        idx = lfsck_sub_trace_file_fid2idx(fid);
        obj = com->lc_sub_trace_objs[idx].lsto_obj;
 
        idx = lfsck_sub_trace_file_fid2idx(fid);
        obj = com->lc_sub_trace_objs[idx].lsto_obj;

+       dev = lfsck_obj2dev(obj);

        mutex_lock(&com->lc_sub_trace_objs[idx].lsto_mutex);
        fid_cpu_to_be(key, fid);
        rc = dt_lookup(env, obj, (struct dt_rec *)&old,
        mutex_lock(&com->lc_sub_trace_objs[idx].lsto_mutex);
        fid_cpu_to_be(key, fid);
        rc = dt_lookup(env, obj, (struct dt_rec *)&old,


@@ -761,7 +763,7 @@ static int lfsck_namespace_links_remove(const struct lu_env *env,
                                        struct dt_object *obj)
 {
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
                                        struct dt_object *obj)
 {
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct thandle                  *th     = NULL;
        int                              rc     = 0;
        ENTRY;
        struct thandle                  *th     = NULL;
        int                              rc     = 0;
        ENTRY;


@@ -918,7 +920,7 @@ static int lfsck_namespace_insert_orphan(const struct lu_env *env,
        const struct lu_fid             *pfid;
        struct lu_fid                    tfid;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        const struct lu_fid             *pfid;
        struct lu_fid                    tfid;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(orphan);

        struct dt_object                *parent;
        struct thandle                  *th     = NULL;
        struct lustre_handle             plh    = { 0 };
        struct dt_object                *parent;
        struct thandle                  *th     = NULL;
        struct lustre_handle             plh    = { 0 };


@@ -1131,20 +1133,34 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        struct lu_attr                  *la     = &info->lti_la;
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lu_attr                  *la     = &info->lti_la;
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

+       /* The child and its name may be on different MDTs. */

        struct dt_device                *dev    = lfsck->li_next;
        struct dt_device                *dev    = lfsck->li_next;

+       struct dt_object                *pobj   = NULL;
+       struct dt_object                *cobj   = NULL;

        struct thandle                  *th     = NULL;
        struct lustre_handle             lh     = { 0 };
        int                              rc     = 0;
        ENTRY;
 
        struct thandle                  *th     = NULL;
        struct lustre_handle             lh     = { 0 };
        int                              rc     = 0;
        ENTRY;
 

-       if (unlikely(!dt_try_as_dir(env, parent)))
+       /* @parent/@child may be based on lfsck->li_bottom,
+        * but here we need the object based on the lfsck->li_next. */
+
+       pobj = lfsck_object_locate(dev, parent);
+       if (IS_ERR(pobj))
+               GOTO(log, rc = PTR_ERR(pobj));
+
+       if (unlikely(!dt_try_as_dir(env, pobj)))

                GOTO(log, rc = -ENOTDIR);
 
                GOTO(log, rc = -ENOTDIR);
 

+       cobj = lfsck_object_locate(dev, child);
+       if (IS_ERR(cobj))
+               GOTO(log, rc = PTR_ERR(cobj));
+

        if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
                GOTO(log, rc = 1);
 
        if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
                GOTO(log, rc = 1);
 

-       /* Hold update lock on the parent to prevent others to access. */
-       rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
+       /* Hold update lock on the pobj to prevent others to access. */
+       rc = lfsck_ibits_lock(env, lfsck, pobj, &lh,

                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);
                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);


@@ -1153,15 +1169,15 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 

-       rec->rec_type = lfsck_object_type(child) & S_IFMT;
-       rec->rec_fid = lfsck_dto2fid(child);
-       rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
+       rec->rec_type = lfsck_object_type(cobj) & S_IFMT;
+       rec->rec_fid = lfsck_dto2fid(cobj);
+       rc = dt_declare_insert(env, pobj, (const struct dt_rec *)rec,

                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 
        if (S_ISDIR(rec->rec_type)) {
                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 
        if (S_ISDIR(rec->rec_type)) {

-               rc = dt_declare_ref_add(env, parent, th);
+               rc = dt_declare_ref_add(env, pobj, th);

                if (rc != 0)
                        GOTO(stop, rc);
        }
                if (rc != 0)
                        GOTO(stop, rc);
        }


@@ -1169,11 +1185,11 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        memset(la, 0, sizeof(*la));
        la->la_ctime = cfs_time_current_sec();
        la->la_valid = LA_CTIME;
        memset(la, 0, sizeof(*la));
        la->la_ctime = cfs_time_current_sec();
        la->la_valid = LA_CTIME;

-       rc = dt_declare_attr_set(env, parent, la, th);
+       rc = dt_declare_attr_set(env, pobj, la, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_declare_attr_set(env, child, la, th);
+       rc = dt_declare_attr_set(env, cobj, la, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -1181,25 +1197,25 @@ static int lfsck_namespace_insert_normal(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_insert(env, parent, (const struct dt_rec *)rec,
+       rc = dt_insert(env, pobj, (const struct dt_rec *)rec,

                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(stop, rc);
 
        if (S_ISDIR(rec->rec_type)) {
                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(stop, rc);
 
        if (S_ISDIR(rec->rec_type)) {

-               dt_write_lock(env, parent, 0);
-               rc = dt_ref_add(env, parent, th);
-               dt_write_unlock(env, parent);
+               dt_write_lock(env, pobj, 0);
+               rc = dt_ref_add(env, pobj, th);
+               dt_write_unlock(env, pobj);

                if (rc != 0)
                        GOTO(stop, rc);
        }
 
        la->la_ctime = cfs_time_current_sec();
                if (rc != 0)
                        GOTO(stop, rc);
        }
 
        la->la_ctime = cfs_time_current_sec();

-       rc = dt_attr_set(env, parent, la, th, BYPASS_CAPA);
+       rc = dt_attr_set(env, pobj, la, th, BYPASS_CAPA);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_attr_set(env, child, la, th, BYPASS_CAPA);
+       rc = dt_attr_set(env, cobj, la, th, BYPASS_CAPA);

 
        GOTO(stop, rc = (rc == 0 ? 1 : rc));
 
 
        GOTO(stop, rc = (rc == 0 ? 1 : rc));
 


@@ -1259,9 +1275,8 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
        struct lu_fid                    tfid;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct lu_fid                    tfid;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_namespace          *ns     = com->lc_file_ram;

-       struct dt_device                *dev;
+       struct dt_device                *dev    = lfsck_obj2dev(orphan);

        struct dt_object                *parent = NULL;
        struct dt_object                *parent = NULL;

-       struct dt_object                *child  = NULL;

        struct thandle                  *th     = NULL;
        struct lustre_handle             lh     = { 0 };
        struct linkea_data               ldata  = { NULL };
        struct thandle                  *th     = NULL;
        struct lustre_handle             lh     = { 0 };
        struct linkea_data               ldata  = { NULL };


@@ -1310,10 +1325,6 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
        if (IS_ERR(dev))
                GOTO(log, rc = PTR_ERR(dev));
 
        if (IS_ERR(dev))
                GOTO(log, rc = PTR_ERR(dev));
 

-       child = lfsck_object_find_by_dev(env, dev, cfid);
-       if (IS_ERR(child))
-               GOTO(log, rc = PTR_ERR(child));
-

        /* Hold update lock on the parent to prevent others to access. */
        rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
                              MDS_INODELOCK_UPDATE, LCK_EX);
        /* Hold update lock on the parent to prevent others to access. */
        rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
                              MDS_INODELOCK_UPDATE, LCK_EX);


@@ -1338,8 +1349,8 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
        la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
                       LA_ATIME | LA_MTIME | LA_CTIME;
 
        la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
                       LA_ATIME | LA_MTIME | LA_CTIME;
 

-       child->do_ops->do_ah_init(env, hint, parent, child,
-                                 la->la_mode & S_IFMT);
+       orphan->do_ops->do_ah_init(env, hint, parent, orphan,
+                                  la->la_mode & S_IFMT);

 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(S_IFDIR);
 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(S_IFDIR);


@@ -1361,39 +1372,39 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
        if (dt_object_remote(orphan))
                th->th_sync = 1;
 
        if (dt_object_remote(orphan))
                th->th_sync = 1;
 

-       rc = dt_declare_create(env, child, la, hint, dof, th);
+       rc = dt_declare_create(env, orphan, la, hint, dof, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       if (unlikely(!dt_try_as_dir(env, child)))
+       if (unlikely(!dt_try_as_dir(env, orphan)))

                GOTO(stop, rc = -ENOTDIR);
 
        rec->rec_type = S_IFDIR;
        rec->rec_fid = cfid;
                GOTO(stop, rc = -ENOTDIR);
 
        rec->rec_type = S_IFDIR;
        rec->rec_fid = cfid;

-       rc = dt_declare_insert(env, child, (const struct dt_rec *)rec,
+       rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,

                               (const struct dt_key *)dot, th);
        if (rc != 0)
                GOTO(stop, rc);
 
        rec->rec_fid = lfsck_dto2fid(parent);
                               (const struct dt_key *)dot, th);
        if (rc != 0)
                GOTO(stop, rc);
 
        rec->rec_fid = lfsck_dto2fid(parent);

-       rc = dt_declare_insert(env, child, (const struct dt_rec *)rec,
+       rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,

                               (const struct dt_key *)dotdot, th);
        if (rc == 0)
                               (const struct dt_key *)dotdot, th);
        if (rc == 0)

-               rc = dt_declare_ref_add(env, child, th);
+               rc = dt_declare_ref_add(env, orphan, th);

 
        if (rc != 0)
                GOTO(stop, rc);
 
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_declare_ref_add(env, child, th);
+       rc = dt_declare_ref_add(env, orphan, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (lmv != NULL) {
                lmv->lmv_magic = LMV_MAGIC;
        if (rc != 0)
                GOTO(stop, rc);
 
        if (lmv != NULL) {
                lmv->lmv_magic = LMV_MAGIC;

-               lmv->lmv_master_mdt_index = lfsck_dev_idx(dev);
+               lmv->lmv_master_mdt_index = lfsck_dev_idx(lfsck);

                lfsck_lmv_header_cpu_to_le(lmv2, lmv);
                lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
                lfsck_lmv_header_cpu_to_le(lmv2, lmv);
                lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));

-               rc = dt_declare_xattr_set(env, child, &lmv_buf,
+               rc = dt_declare_xattr_set(env, orphan, &lmv_buf,

                                          XATTR_NAME_LMV, 0, th);
                if (rc != 0)
                        GOTO(stop, rc);
                                          XATTR_NAME_LMV, 0, th);
                if (rc != 0)
                        GOTO(stop, rc);


@@ -1401,7 +1412,7 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
 
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);
 
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);

-       rc = dt_declare_xattr_set(env, child, &linkea_buf,
+       rc = dt_declare_xattr_set(env, orphan, &linkea_buf,

                                  XATTR_NAME_LINK, 0, th);
        if (rc != 0)
                GOTO(stop, rc);
                                  XATTR_NAME_LINK, 0, th);
        if (rc != 0)
                GOTO(stop, rc);


@@ -1419,38 +1430,38 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       dt_write_lock(env, child, 0);
-       rc = dt_create(env, child, la, hint, dof, th);
+       dt_write_lock(env, orphan, 0);
+       rc = dt_create(env, orphan, la, hint, dof, th);

        if (rc != 0)
                GOTO(unlock2, rc);
 
        rec->rec_fid = cfid;
        if (rc != 0)
                GOTO(unlock2, rc);
 
        rec->rec_fid = cfid;

-       rc = dt_insert(env, child, (const struct dt_rec *)rec,
+       rc = dt_insert(env, orphan, (const struct dt_rec *)rec,

                       (const struct dt_key *)dot, th, BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(unlock2, rc);
 
        rec->rec_fid = lfsck_dto2fid(parent);
                       (const struct dt_key *)dot, th, BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(unlock2, rc);
 
        rec->rec_fid = lfsck_dto2fid(parent);

-       rc = dt_insert(env, child, (const struct dt_rec *)rec,
+       rc = dt_insert(env, orphan, (const struct dt_rec *)rec,

                       (const struct dt_key *)dotdot, th,
                       BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(unlock2, rc);
 
                       (const struct dt_key *)dotdot, th,
                       BYPASS_CAPA, 1);
        if (rc != 0)
                GOTO(unlock2, rc);
 

-       rc = dt_ref_add(env, child, th);
+       rc = dt_ref_add(env, orphan, th);

        if (rc != 0)
                GOTO(unlock2, rc);
 
        if (lmv != NULL) {
        if (rc != 0)
                GOTO(unlock2, rc);
 
        if (lmv != NULL) {

-               rc = dt_xattr_set(env, child, &lmv_buf, XATTR_NAME_LMV, 0,
+               rc = dt_xattr_set(env, orphan, &lmv_buf, XATTR_NAME_LMV, 0,

                                  th, BYPASS_CAPA);
                if (rc != 0)
                        GOTO(unlock2, rc);
        }
 
                                  th, BYPASS_CAPA);
                if (rc != 0)
                        GOTO(unlock2, rc);
        }
 

-       rc = dt_xattr_set(env, child, &linkea_buf,
+       rc = dt_xattr_set(env, orphan, &linkea_buf,

                          XATTR_NAME_LINK, 0, th, BYPASS_CAPA);
                          XATTR_NAME_LINK, 0, th, BYPASS_CAPA);

-       dt_write_unlock(env, child);
+       dt_write_unlock(env, orphan);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -1466,7 +1477,7 @@ static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
        GOTO(stop, rc = (rc == 0 ? 1 : rc));
 
 unlock2:
        GOTO(stop, rc = (rc == 0 ? 1 : rc));
 
 unlock2:

-       dt_write_unlock(env, child);
+       dt_write_unlock(env, orphan);

 
 stop:
        dt_trans_stop(env, dev, th);
 
 stop:
        dt_trans_stop(env, dev, th);


@@ -1480,9 +1491,6 @@ log:
               lfsck_lfsck2name(lfsck), PFID(cfid),
               cname->ln_name != NULL ? cname->ln_name : "<NULL>", rc);
 
               lfsck_lfsck2name(lfsck), PFID(cfid),
               cname->ln_name != NULL ? cname->ln_name : "<NULL>", rc);
 

-       if (child != NULL && !IS_ERR(child))
-               lfsck_object_put(env, child);
-

        if (parent != NULL && !IS_ERR(parent) && parent != lfsck->li_lpf_obj)
                lfsck_object_put(env, parent);
 
        if (parent != NULL && !IS_ERR(parent) && parent != lfsck->li_lpf_obj)
                lfsck_object_put(env, parent);
 


@@ -1521,7 +1529,7 @@ static int lfsck_namespace_shrink_linkea(const struct lu_env *env,
                                         bool next)
 {
        struct lfsck_instance           *lfsck     = com->lc_lfsck;
                                         bool next)
 {
        struct lfsck_instance           *lfsck     = com->lc_lfsck;

-       struct dt_device                *dev       = lfsck->li_bottom;
+       struct dt_device                *dev       = lfsck_obj2dev(obj);

        struct lfsck_bookmark           *bk        = &lfsck->li_bookmark_ram;
        struct thandle                  *th        = NULL;
        struct lustre_handle             lh        = { 0 };
        struct lfsck_bookmark           *bk        = &lfsck->li_bookmark_ram;
        struct thandle                  *th        = NULL;
        struct lustre_handle             lh        = { 0 };


@@ -1738,9 +1746,11 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lu_fid                    tfid;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lu_fid                    tfid;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

+       /* The child and its name may be on different MDTs. */

        struct dt_device                *dev    = lfsck->li_next;
        const char                      *name   = cname->ln_name;
        struct dt_device                *dev    = lfsck->li_next;
        const char                      *name   = cname->ln_name;

-       struct dt_object                *obj    = NULL;
+       struct dt_object                *pobj   = NULL;
+       struct dt_object                *cobj   = NULL;

        struct lustre_handle             plh    = { 0 };
        struct lustre_handle             clh    = { 0 };
        struct linkea_data               ldata  = { NULL };
        struct lustre_handle             plh    = { 0 };
        struct lustre_handle             clh    = { 0 };
        struct linkea_data               ldata  = { NULL };


@@ -1749,7 +1759,18 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
        int                              rc     = 0;
        ENTRY;
 
        int                              rc     = 0;
        ENTRY;
 

-       rc = lfsck_ibits_lock(env, lfsck, parent, &plh,
+       /* @parent/@child may be based on lfsck->li_bottom,
+        * but here we need the object based on the lfsck->li_next. */
+
+       pobj = lfsck_object_locate(dev, parent);
+       if (IS_ERR(pobj))
+               GOTO(log, rc = PTR_ERR(pobj));
+
+       if (unlikely(!dt_try_as_dir(env, pobj)))
+               GOTO(log, rc = -ENOTDIR);
+
+       /* Hold update lock on the pobj to prevent others to access. */
+       rc = lfsck_ibits_lock(env, lfsck, pobj, &plh,

                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);
                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);


@@ -1759,9 +1780,9 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
                goto replace;
        }
 
                goto replace;
        }
 

-       obj = lfsck_object_find(env, lfsck, cfid);
-       if (IS_ERR(obj)) {
-               rc = PTR_ERR(obj);
+       cobj = lfsck_object_find_by_dev(env, dev, cfid);
+       if (IS_ERR(cobj)) {
+               rc = PTR_ERR(cobj);

                if (rc == -ENOENT) {
                        exist = false;
                        goto replace;
                if (rc == -ENOENT) {
                        exist = false;
                        goto replace;


@@ -1770,12 +1791,12 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
                GOTO(log, rc);
        }
 
                GOTO(log, rc);
        }
 

-       if (!dt_object_exists(obj)) {
+       if (!dt_object_exists(cobj)) {

                exist = false;
                goto replace;
        }
 
                exist = false;
                goto replace;
        }
 

-       rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
+       rc = dt_lookup(env, pobj, (struct dt_rec *)&tfid,

                       (const struct dt_key *)name, BYPASS_CAPA);
        if (rc == -ENOENT) {
                exist = false;
                       (const struct dt_key *)name, BYPASS_CAPA);
        if (rc == -ENOENT) {
                exist = false;


@@ -1790,18 +1811,18 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
                GOTO(log, rc = 0);
 
        /* lock the object to be destroyed. */
                GOTO(log, rc = 0);
 
        /* lock the object to be destroyed. */

-       rc = lfsck_ibits_lock(env, lfsck, obj, &clh,
+       rc = lfsck_ibits_lock(env, lfsck, cobj, &clh,

                              MDS_INODELOCK_UPDATE |
                              MDS_INODELOCK_XATTR, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);
 
                              MDS_INODELOCK_UPDATE |
                              MDS_INODELOCK_XATTR, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);
 

-       if (unlikely(lfsck_is_dead_obj(obj))) {
+       if (unlikely(lfsck_is_dead_obj(cobj))) {

                exist = false;
                goto replace;
        }
 
                exist = false;
                goto replace;
        }
 

-       rc = dt_attr_get(env, obj, la, BYPASS_CAPA);
+       rc = dt_attr_get(env, cobj, la, BYPASS_CAPA);

        if (rc != 0)
                GOTO(log, rc);
 
        if (rc != 0)
                GOTO(log, rc);
 


@@ -1811,7 +1832,7 @@ static int lfsck_namespace_replace_cond(const struct lu_env *env,
                GOTO(log, rc);
 
        if (S_ISREG(la->la_mode)) {
                GOTO(log, rc);
 
        if (S_ISREG(la->la_mode)) {

-               rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LOV,
+               rc = dt_xattr_get(env, cobj, &LU_BUF_NULL, XATTR_NAME_LOV,

                                  BYPASS_CAPA);
                /* If someone has created related OST-object(s),
                 * then keep it. */
                                  BYPASS_CAPA);
                /* If someone has created related OST-object(s),
                 * then keep it. */


@@ -1831,7 +1852,7 @@ replace:
        if (rc != 0)
                GOTO(log, rc);
 
        if (rc != 0)
                GOTO(log, rc);
 

-       rc = linkea_links_find(&ldata, cname, lfsck_dto2fid(parent));
+       rc = linkea_links_find(&ldata, cname, lfsck_dto2fid(pobj));

        /* Someone moved the child, no need to replace. */
        if (rc != 0)
                GOTO(log, rc = 0);
        /* Someone moved the child, no need to replace. */
        if (rc != 0)
                GOTO(log, rc = 0);


@@ -1844,37 +1865,38 @@ replace:
                GOTO(log, rc = PTR_ERR(th));
 
        if (exist) {
                GOTO(log, rc = PTR_ERR(th));
 
        if (exist) {

-               rc = dt_declare_destroy(env, obj, th);
+               rc = dt_declare_destroy(env, cobj, th);

                if (rc != 0)
                        GOTO(stop, rc);
        }
 
                if (rc != 0)
                        GOTO(stop, rc);
        }
 

-       rc = dt_declare_delete(env, parent, (const struct dt_key *)name, th);
+       rc = dt_declare_delete(env, pobj, (const struct dt_key *)name, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        rec->rec_type = S_IFDIR;
        rec->rec_fid = lfsck_dto2fid(child);
        if (rc != 0)
                GOTO(stop, rc);
 
        rec->rec_type = S_IFDIR;
        rec->rec_fid = lfsck_dto2fid(child);

-       rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
+       rc = dt_declare_insert(env, pobj, (const struct dt_rec *)rec,

                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 
                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (exist) {
        if (rc != 0)
                GOTO(stop, rc);
 
        if (exist) {

-               rc = dt_destroy(env, obj, th);
+               rc = dt_destroy(env, cobj, th);

                if (rc != 0)
                        GOTO(stop, rc);
        }
 
        /* The old name entry maybe not exist. */
                if (rc != 0)
                        GOTO(stop, rc);
        }
 
        /* The old name entry maybe not exist. */

-       dt_delete(env, parent, (const struct dt_key *)name, th,
-                 BYPASS_CAPA);
+       rc = dt_delete(env, pobj, (const struct dt_key *)name, th, BYPASS_CAPA);
+       if (rc != 0 && rc != -ENOENT)
+               GOTO(stop, rc);

 
 

-       rc = dt_insert(env, parent, (const struct dt_rec *)rec,
+       rc = dt_insert(env, pobj, (const struct dt_rec *)rec,

                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
 
        GOTO(stop, rc = (rc == 0 ? 1 : rc));
                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
 
        GOTO(stop, rc = (rc == 0 ? 1 : rc));


@@ -1885,8 +1907,9 @@ stop:
 log:
        lfsck_ibits_unlock(&clh, LCK_EX);
        lfsck_ibits_unlock(&plh, LCK_EX);
 log:
        lfsck_ibits_unlock(&clh, LCK_EX);
        lfsck_ibits_unlock(&plh, LCK_EX);

-       if (obj != NULL && !IS_ERR(obj))
-               lfsck_object_put(env, obj);
+
+       if (cobj != NULL && !IS_ERR(cobj))
+               lfsck_object_put(env, cobj);

 
        CDEBUG(D_LFSCK, "%s: namespace LFSCK conditionally destroy the "
               "object "DFID" because of conflict with the object "DFID
 
        CDEBUG(D_LFSCK, "%s: namespace LFSCK conditionally destroy the "
               "object "DFID" because of conflict with the object "DFID


@@ -1918,7 +1941,7 @@ int lfsck_namespace_rebuild_linkea(const struct lu_env *env,
                                   struct linkea_data *ldata)
 {
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
                                   struct linkea_data *ldata)
 {
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct thandle                  *th     = NULL;
        struct lu_buf                    linkea_buf;
        int                              rc     = 0;
        struct thandle                  *th     = NULL;
        struct lu_buf                    linkea_buf;
        int                              rc     = 0;


@@ -2003,7 +2026,7 @@ int lfsck_namespace_repair_dirent(const struct lu_env *env,
        const struct lu_fid     *cfid   = lfsck_dto2fid(child);
        struct lu_fid            tfid;
        struct lfsck_instance   *lfsck  = com->lc_lfsck;
        const struct lu_fid     *cfid   = lfsck_dto2fid(child);
        struct lu_fid            tfid;
        struct lfsck_instance   *lfsck  = com->lc_lfsck;

-       struct dt_device        *dev    = lfsck->li_next;
+       struct dt_device        *dev    = lfsck_obj2dev(parent);

        struct thandle          *th     = NULL;
        struct lustre_handle     lh     = { 0 };
        int                      rc     = 0;
        struct thandle          *th     = NULL;
        struct lustre_handle     lh     = { 0 };
        int                      rc     = 0;


@@ -2041,7 +2064,7 @@ int lfsck_namespace_repair_dirent(const struct lu_env *env,
                        GOTO(stop, rc);
        }
 
                        GOTO(stop, rc);
        }
 

-       rc = dt_trans_start(env, dev, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -2146,7 +2169,7 @@ static int lfsck_namespace_repair_unmatched_pairs(const struct lu_env *env,
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct thandle                  *th     = NULL;
        struct linkea_data               ldata  = { NULL };
        struct lu_buf                    linkea_buf;
        struct thandle                  *th     = NULL;
        struct linkea_data               ldata  = { NULL };
        struct lu_buf                    linkea_buf;


@@ -2854,9 +2877,8 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
        struct lu_fid                   *tfid   = &info->lti_fid3;
        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lu_fid                   *tfid   = &info->lti_fid3;
        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck->li_bottom;
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        const struct lu_fid             *cfid   = lfsck_dto2fid(obj);
        const struct lu_fid             *cfid   = lfsck_dto2fid(obj);

-       struct dt_object                *child  = NULL;

        struct thandle                  *th     = NULL;
        struct linkea_data               ldata  = { NULL };
        struct lustre_handle             lh     = { 0 };
        struct thandle                  *th     = NULL;
        struct linkea_data               ldata  = { NULL };
        struct lustre_handle             lh     = { 0 };


@@ -2869,11 +2891,7 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
        LASSERT(!dt_object_remote(obj));
        LASSERT(S_ISREG(lfsck_object_type(obj)));
 
        LASSERT(!dt_object_remote(obj));
        LASSERT(S_ISREG(lfsck_object_type(obj)));
 

-       child = lfsck_object_find_by_dev(env, dev, cfid);
-       if (IS_ERR(child))
-               GOTO(log, rc = PTR_ERR(child));
-
-       rc = lfsck_ibits_lock(env, lfsck, child, &lh,
+       rc = lfsck_ibits_lock(env, lfsck, obj, &lh,

                              MDS_INODELOCK_UPDATE |
                              MDS_INODELOCK_XATTR, LCK_EX);
        if (rc != 0)
                              MDS_INODELOCK_UPDATE |
                              MDS_INODELOCK_XATTR, LCK_EX);
        if (rc != 0)


@@ -2884,7 +2902,7 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
                GOTO(log, rc = PTR_ERR(th));
 
        la->la_valid = LA_NLINK;
                GOTO(log, rc = PTR_ERR(th));
 
        la->la_valid = LA_NLINK;

-       rc = dt_declare_attr_set(env, child, la, th);
+       rc = dt_declare_attr_set(env, obj, la, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 


@@ -2892,7 +2910,7 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       dt_write_lock(env, child, 0);
+       dt_write_lock(env, obj, 0);

        /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT has
         * ever tried to verify some remote MDT-object that resides on this
         * MDT, but this MDT failed to respond such request. So means there
        /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT has
         * ever tried to verify some remote MDT-object that resides on this
         * MDT, but this MDT failed to respond such request. So means there


@@ -2914,11 +2932,11 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
        if (flags & LNTF_SKIP_NLINK)
                GOTO(unlock, rc = 0);
 
        if (flags & LNTF_SKIP_NLINK)
                GOTO(unlock, rc = 0);
 

-       rc = dt_attr_get(env, child, la, BYPASS_CAPA);
+       rc = dt_attr_get(env, obj, la, BYPASS_CAPA);

        if (rc != 0)
                GOTO(unlock, rc = (rc == -ENOENT ? 0 : rc));
 
        if (rc != 0)
                GOTO(unlock, rc = (rc == -ENOENT ? 0 : rc));
 

-       rc = lfsck_links_read2(env, child, &ldata);
+       rc = lfsck_links_read2(env, obj, &ldata);

        if (rc != 0)
                GOTO(unlock, rc = (rc == -ENODATA ? 0 : rc));
 
        if (rc != 0)
                GOTO(unlock, rc = (rc == -ENODATA ? 0 : rc));
 


@@ -2930,21 +2948,18 @@ static int lfsck_namespace_repair_nlink(const struct lu_env *env,
        if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
                GOTO(unlock, rc = 1);
 
        if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
                GOTO(unlock, rc = 1);
 

-       rc = dt_attr_set(env, child, la, th, BYPASS_CAPA);
+       rc = dt_attr_set(env, obj, la, th, BYPASS_CAPA);

 
        GOTO(unlock, rc = (rc == 0 ? 1 : rc));
 
 unlock:
 
        GOTO(unlock, rc = (rc == 0 ? 1 : rc));
 
 unlock:

-       dt_write_unlock(env, child);
+       dt_write_unlock(env, obj);

 
 stop:
        dt_trans_stop(env, dev, th);
 
 log:
        lfsck_ibits_unlock(&lh, LCK_EX);
 
 stop:
        dt_trans_stop(env, dev, th);
 
 log:
        lfsck_ibits_unlock(&lh, LCK_EX);

-       if (child != NULL && !IS_ERR(child))
-               lfsck_object_put(env, child);
-

        CDEBUG(D_LFSCK, "%s: namespace LFSCK repaired the object "DFID"'s "
               "nlink count from %u to %u: rc = %d\n",
               lfsck_lfsck2name(lfsck), PFID(cfid), old, la->la_nlink, rc);
        CDEBUG(D_LFSCK, "%s: namespace LFSCK repaired the object "DFID"'s "
               "nlink count from %u to %u: rc = %d\n",
               lfsck_lfsck2name(lfsck), PFID(cfid), old, la->la_nlink, rc);


@@ -3757,7 +3772,7 @@ out:
        up_write(&com->lc_sem);
 
 put:
        up_write(&com->lc_sem);
 
 put:

-       lu_object_put(env, &root->do_lu);
+       lfsck_object_put(env, root);

 log:
        CDEBUG(D_LFSCK, "%s: namespace LFSCK reset: rc = %d\n",
               lfsck_lfsck2name(lfsck), rc);
 log:
        CDEBUG(D_LFSCK, "%s: namespace LFSCK reset: rc = %d\n",
               lfsck_lfsck2name(lfsck), rc);


@@ -3846,10 +3861,9 @@ static int lfsck_namespace_open_dir(const struct lu_env *env,
        if (llmv->ll_lmv_master) {
                struct lmv_mds_md_v1 *lmv = &llmv->ll_lmv;
 
        if (llmv->ll_lmv_master) {
                struct lmv_mds_md_v1 *lmv = &llmv->ll_lmv;
 

-               if (lmv->lmv_master_mdt_index !=
-                   lfsck_dev_idx(lfsck->li_bottom)) {
+               if (lmv->lmv_master_mdt_index != lfsck_dev_idx(lfsck)) {

                        lmv->lmv_master_mdt_index =
                        lmv->lmv_master_mdt_index =

-                               lfsck_dev_idx(lfsck->li_bottom);
+                               lfsck_dev_idx(lfsck);

                        ns->ln_flags |= LF_INCONSISTENT;
                        llmv->ll_lmv_updated = 1;
                }
                        ns->ln_flags |= LF_INCONSISTENT;
                        llmv->ll_lmv_updated = 1;
                }


@@ -4011,11 +4025,9 @@ static int lfsck_namespace_exec_oit(const struct lu_env *env,
        struct lu_fid            *pfid  = &info->lti_fid2;
        struct lu_name           *cname = &info->lti_name;
        struct lu_seq_range      *range = &info->lti_range;
        struct lu_fid            *pfid  = &info->lti_fid2;
        struct lu_name           *cname = &info->lti_name;
        struct lu_seq_range      *range = &info->lti_range;

-       struct dt_device         *dev   = lfsck->li_bottom;
-       struct seq_server_site   *ss    =
-                               lu_site2seq(dev->dd_lu_dev.ld_site);
+       struct seq_server_site   *ss    = lfsck_dev_site(lfsck);

        struct linkea_data        ldata = { NULL };
        struct linkea_data        ldata = { NULL };

-       __u32                     idx   = lfsck_dev_idx(dev);
+       __u32                     idx   = lfsck_dev_idx(lfsck);

        int                       rc;
        ENTRY;
 
        int                       rc;
        ENTRY;
 


@@ -4564,8 +4576,7 @@ log:
        case LE_SET_LMV_MASTER: {
                struct dt_object        *obj;
 
        case LE_SET_LMV_MASTER: {
                struct dt_object        *obj;
 

-               obj = lfsck_object_find_by_dev(env, lfsck->li_bottom,
-                                              &lr->lr_fid);
+               obj = lfsck_object_find_bottom(env, lfsck, &lr->lr_fid);

                if (IS_ERR(obj))
                        RETURN(PTR_ERR(obj));
 
                if (IS_ERR(obj))
                        RETURN(PTR_ERR(obj));
 


@@ -4725,6 +4736,8 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
        struct dt_object_format         *dof    = &info->lti_dof;
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
        struct dt_object_format         *dof    = &info->lti_dof;
        struct dt_insert_rec            *rec    = &info->lti_dt_rec;
        struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;

+       struct dt_object                *pobj   = NULL;
+       struct dt_object                *cobj   = NULL;

        const struct lu_name            *cname;
        struct linkea_data               ldata  = { NULL };
        struct lustre_handle             lh     = { 0 };
        const struct lu_name            *cname;
        struct linkea_data               ldata  = { NULL };
        struct lustre_handle             lh     = { 0 };


@@ -4732,7 +4745,7 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
        struct lu_buf                    lmv_buf;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
        struct lu_buf                    lmv_buf;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;

-       struct dt_device                *dev    = lfsck_obj2dt_dev(child);
+       struct dt_device                *dev    = lfsck->li_next;

        struct thandle                  *th     = NULL;
        int                              rc     = 0;
        __u16                            type   = lnr->lnr_type;
        struct thandle                  *th     = NULL;
        int                              rc     = 0;
        __u16                            type   = lnr->lnr_type;


@@ -4748,27 +4761,37 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
        if (!create || bk->lb_param & LPF_DRYRUN)
                GOTO(log, rc = 0);
 
        if (!create || bk->lb_param & LPF_DRYRUN)
                GOTO(log, rc = 0);
 

+       /* We may need to create the sub-objects of the @child via LOD,
+        * so make the modification based on lfsck->li_next. */
+
+       pobj = lfsck_object_locate(dev, parent);
+       if (IS_ERR(pobj))
+               GOTO(log, rc = PTR_ERR(pobj));
+
+       if (unlikely(!dt_try_as_dir(env, pobj)))
+               GOTO(log, rc = -ENOTDIR);
+
+       cobj = lfsck_object_locate(dev, child);
+       if (IS_ERR(cobj))
+               GOTO(log, rc = PTR_ERR(cobj));
+

        rc = linkea_data_new(&ldata, &info->lti_linkea_buf2);
        if (rc != 0)
                GOTO(log, rc);
 
        rc = linkea_data_new(&ldata, &info->lti_linkea_buf2);
        if (rc != 0)
                GOTO(log, rc);
 

-       rc = linkea_add_buf(&ldata, cname, lfsck_dto2fid(parent));
+       rc = linkea_add_buf(&ldata, cname, lfsck_dto2fid(pobj));

        if (rc != 0)
                GOTO(log, rc);
 
        if (rc != 0)
                GOTO(log, rc);
 

-       rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
+       rc = lfsck_ibits_lock(env, lfsck, pobj, &lh,

                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);
 
                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                GOTO(log, rc);
 

-       rc = lfsck_namespace_check_exist(env, parent, child, lnr->lnr_name);
+       rc = lfsck_namespace_check_exist(env, pobj, cobj, lnr->lnr_name);

        if (rc != 0)
                GOTO(log, rc);
 
        if (rc != 0)
                GOTO(log, rc);
 

-       th = dt_trans_create(env, dev);
-       if (IS_ERR(th))
-               GOTO(log, rc = PTR_ERR(th));
-

        /* Set the ctime as zero, then others can know it is created for
         * repairing dangling name entry by LFSCK. And if the LFSCK made
         * wrong decision and the real MDT-object has been found later,
        /* Set the ctime as zero, then others can know it is created for
         * repairing dangling name entry by LFSCK. And if the LFSCK made
         * wrong decision and the real MDT-object has been found later,


@@ -4778,8 +4801,8 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
        la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
                        LA_ATIME | LA_MTIME | LA_CTIME;
 
        la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
                        LA_ATIME | LA_MTIME | LA_CTIME;
 

-       child->do_ops->do_ah_init(env, hint, parent, child,
-                                 la->la_mode & S_IFMT);
+       cobj->do_ops->do_ah_init(env, hint, pobj, cobj,
+                                la->la_mode & S_IFMT);

 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(type);
 
        memset(dof, 0, sizeof(*dof));
        dof->dof_type = dt_mode_to_dft(type);


@@ -4787,34 +4810,38 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
         * the MDT-object without stripes (dof->dof_reg.striped = 0). related
         * OST-objects will be created when write open. */
 
         * the MDT-object without stripes (dof->dof_reg.striped = 0). related
         * OST-objects will be created when write open. */
 

+       th = dt_trans_create(env, dev);
+       if (IS_ERR(th))
+               GOTO(log, rc = PTR_ERR(th));
+

        /* 1a. create child. */
        /* 1a. create child. */

-       rc = dt_declare_create(env, child, la, hint, dof, th);
+       rc = dt_declare_create(env, cobj, la, hint, dof, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (S_ISDIR(type)) {
        if (rc != 0)
                GOTO(stop, rc);
 
        if (S_ISDIR(type)) {

-               if (unlikely(!dt_try_as_dir(env, child)))
+               if (unlikely(!dt_try_as_dir(env, cobj)))

                        GOTO(stop, rc = -ENOTDIR);
 
                /* 2a. insert dot into child dir */
                rec->rec_type = S_IFDIR;
                        GOTO(stop, rc = -ENOTDIR);
 
                /* 2a. insert dot into child dir */
                rec->rec_type = S_IFDIR;

-               rec->rec_fid = lfsck_dto2fid(child);
-               rc = dt_declare_insert(env, child,
+               rec->rec_fid = lfsck_dto2fid(cobj);
+               rc = dt_declare_insert(env, cobj,

                                       (const struct dt_rec *)rec,
                                       (const struct dt_key *)dot, th);
                if (rc != 0)
                        GOTO(stop, rc);
 
                /* 3a. insert dotdot into child dir */
                                       (const struct dt_rec *)rec,
                                       (const struct dt_key *)dot, th);
                if (rc != 0)
                        GOTO(stop, rc);
 
                /* 3a. insert dotdot into child dir */

-               rec->rec_fid = lfsck_dto2fid(parent);
-               rc = dt_declare_insert(env, child,
+               rec->rec_fid = lfsck_dto2fid(pobj);
+               rc = dt_declare_insert(env, cobj,

                                       (const struct dt_rec *)rec,
                                       (const struct dt_key *)dotdot, th);
                if (rc != 0)
                        GOTO(stop, rc);
 
                /* 4a. increase child nlink */
                                       (const struct dt_rec *)rec,
                                       (const struct dt_key *)dotdot, th);
                if (rc != 0)
                        GOTO(stop, rc);
 
                /* 4a. increase child nlink */

-               rc = dt_declare_ref_add(env, child, th);
+               rc = dt_declare_ref_add(env, cobj, th);

                if (rc != 0)
                        GOTO(stop, rc);
 
                if (rc != 0)
                        GOTO(stop, rc);
 


@@ -4824,7 +4851,7 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
 
                        idx = lfsck_shard_name_to_index(env,
                                        lnr->lnr_name, lnr->lnr_namelen,
 
                        idx = lfsck_shard_name_to_index(env,
                                        lnr->lnr_name, lnr->lnr_namelen,

-                                       type, lfsck_dto2fid(child));
+                                       type, lfsck_dto2fid(cobj));

                        if (unlikely(idx < 0))
                                GOTO(stop, rc = idx);
 
                        if (unlikely(idx < 0))
                                GOTO(stop, rc = idx);
 


@@ -4834,7 +4861,7 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
 
                        lfsck_lmv_header_cpu_to_le(lmv2, lmv2);
                        lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
 
                        lfsck_lmv_header_cpu_to_le(lmv2, lmv2);
                        lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));

-                       rc = dt_declare_xattr_set(env, child, &lmv_buf,
+                       rc = dt_declare_xattr_set(env, cobj, &lmv_buf,

                                                  XATTR_NAME_LMV, 0, th);
                        if (rc != 0)
                                GOTO(stop, rc);
                                                  XATTR_NAME_LMV, 0, th);
                        if (rc != 0)
                                GOTO(stop, rc);


@@ -4844,49 +4871,46 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
        /* 6a. insert linkEA for child */
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);
        /* 6a. insert linkEA for child */
        lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
                       ldata.ld_leh->leh_len);

-       rc = dt_declare_xattr_set(env, child, &linkea_buf,
+       rc = dt_declare_xattr_set(env, cobj, &linkea_buf,

                                  XATTR_NAME_LINK, 0, th);
        if (rc != 0)
                GOTO(stop, rc);
 
                                  XATTR_NAME_LINK, 0, th);
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc = (rc == -EEXIST ? 1 : rc));
 
        if (rc != 0)
                GOTO(stop, rc = (rc == -EEXIST ? 1 : rc));
 

-       dt_write_lock(env, child, 0);
+       dt_write_lock(env, cobj, 0);

        /* 1b. create child */
        /* 1b. create child */

-       rc = dt_create(env, child, la, hint, dof, th);
+       rc = dt_create(env, cobj, la, hint, dof, th);

        if (rc != 0)
                GOTO(unlock, rc = (rc == -EEXIST ? 1 : rc));
 
        if (S_ISDIR(type)) {
        if (rc != 0)
                GOTO(unlock, rc = (rc == -EEXIST ? 1 : rc));
 
        if (S_ISDIR(type)) {

-               if (unlikely(!dt_try_as_dir(env, child)))
-                       GOTO(unlock, rc = -ENOTDIR);
-

                /* 2b. insert dot into child dir */
                rec->rec_type = S_IFDIR;
                /* 2b. insert dot into child dir */
                rec->rec_type = S_IFDIR;

-               rec->rec_fid = lfsck_dto2fid(child);
-               rc = dt_insert(env, child, (const struct dt_rec *)rec,
+               rec->rec_fid = lfsck_dto2fid(cobj);
+               rc = dt_insert(env, cobj, (const struct dt_rec *)rec,

                               (const struct dt_key *)dot, th, BYPASS_CAPA, 1);
                if (rc != 0)
                        GOTO(unlock, rc);
 
                /* 3b. insert dotdot into child dir */
                               (const struct dt_key *)dot, th, BYPASS_CAPA, 1);
                if (rc != 0)
                        GOTO(unlock, rc);
 
                /* 3b. insert dotdot into child dir */

-               rec->rec_fid = lfsck_dto2fid(parent);
-               rc = dt_insert(env, child, (const struct dt_rec *)rec,
+               rec->rec_fid = lfsck_dto2fid(pobj);
+               rc = dt_insert(env, cobj, (const struct dt_rec *)rec,

                               (const struct dt_key *)dotdot, th,
                               BYPASS_CAPA, 1);
                if (rc != 0)
                        GOTO(unlock, rc);
 
                /* 4b. increase child nlink */
                               (const struct dt_key *)dotdot, th,
                               BYPASS_CAPA, 1);
                if (rc != 0)
                        GOTO(unlock, rc);
 
                /* 4b. increase child nlink */

-               rc = dt_ref_add(env, child, th);
+               rc = dt_ref_add(env, cobj, th);

                if (rc != 0)
                        GOTO(unlock, rc);
 
                /* 5b. generate slave LMV EA. */
                if (lnr->lnr_lmv != NULL && lnr->lnr_lmv->ll_lmv_master) {
                if (rc != 0)
                        GOTO(unlock, rc);
 
                /* 5b. generate slave LMV EA. */
                if (lnr->lnr_lmv != NULL && lnr->lnr_lmv->ll_lmv_master) {

-                       rc = dt_xattr_set(env, child, &lmv_buf, XATTR_NAME_LMV,
+                       rc = dt_xattr_set(env, cobj, &lmv_buf, XATTR_NAME_LMV,

                                          0, th, BYPASS_CAPA);
                        if (rc != 0)
                                GOTO(unlock, rc);
                                          0, th, BYPASS_CAPA);
                        if (rc != 0)
                                GOTO(unlock, rc);


@@ -4894,13 +4918,13 @@ int lfsck_namespace_repair_dangling(const struct lu_env *env,
        }
 
        /* 6b. insert linkEA for child. */
        }
 
        /* 6b. insert linkEA for child. */

-       rc = dt_xattr_set(env, child, &linkea_buf,
+       rc = dt_xattr_set(env, cobj, &linkea_buf,

                          XATTR_NAME_LINK, 0, th, BYPASS_CAPA);
 
        GOTO(unlock, rc);
 
 unlock:
                          XATTR_NAME_LINK, 0, th, BYPASS_CAPA);
 
        GOTO(unlock, rc);
 
 unlock:

-       dt_write_unlock(env, child);
+       dt_write_unlock(env, cobj);

 
 stop:
        dt_trans_stop(env, dev, th);
 
 stop:
        dt_trans_stop(env, dev, th);


@@ -4910,7 +4934,7 @@ log:
        CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found dangling "
               "reference for: parent "DFID", child "DFID", type %u, "
               "name %s. %s: rc = %d\n", lfsck_lfsck2name(lfsck),
        CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found dangling "
               "reference for: parent "DFID", child "DFID", type %u, "
               "name %s. %s: rc = %d\n", lfsck_lfsck2name(lfsck),

-              PFID(lfsck_dto2fid(parent)), PFID(lfsck_dto2fid(child)),
+              PFID(&lnr->lnr_lar.lar_fid), PFID(lfsck_dto2fid(child)),

               type, cname->ln_name,
               create ? "Create the lost OST-object as required" :
                        "Keep the MDT-object there by default", rc);
               type, cname->ln_name,
               create ? "Create the lost OST-object as required" :
                        "Keep the MDT-object there by default", rc);


@@ -5025,7 +5049,7 @@ static int lfsck_namespace_assistant_handler_p1(const struct lu_env *env,
        if (idx < 0)
                GOTO(out, rc = idx);
 
        if (idx < 0)
                GOTO(out, rc = idx);
 

-       if (idx == lfsck_dev_idx(lfsck->li_bottom)) {
+       if (idx == lfsck_dev_idx(lfsck)) {

                if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
                        GOTO(out, rc = 0);
 
                if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
                        GOTO(out, rc = 0);
 


@@ -5100,7 +5124,7 @@ again:
                if (rc != 0)
                        GOTO(stop, rc);
 
                if (rc != 0)
                        GOTO(stop, rc);
 

-               rc = dt_trans_start(env, dev, handle);
+               rc = dt_trans_start_local(env, dev, handle);

                if (rc != 0)
                        GOTO(stop, rc);
 
                if (rc != 0)
                        GOTO(stop, rc);
 


@@ -5333,7 +5357,7 @@ out:
                if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
                     rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
                     rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&
                if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
                     rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
                     rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&

-                   dev != NULL && dev != lfsck->li_next)
+                   dev != NULL && dev != lfsck->li_bottom)

                        lfsck_lad_set_bitmap(env, com, idx);
 
                if (!(bk->lb_param & LPF_FAILOUT))
                        lfsck_lad_set_bitmap(env, com, idx);
 
                if (!(bk->lb_param & LPF_FAILOUT))


@@ -5397,7 +5421,7 @@ out:
                lfsck_object_put(env, obj);
 
 put_dir:
                lfsck_object_put(env, obj);
 
 put_dir:

-       lu_object_put(env, &dir->do_lu);
+       lfsck_object_put(env, dir);

 
        return rc;
 }
 
        return rc;
 }


@@ -5550,7 +5574,7 @@ stop:
        dt_trans_stop(env, dev, th);
 
 out:
        dt_trans_stop(env, dev, th);
 
 out:

-       lu_object_put(env, &child->do_lu);
+       lfsck_object_put(env, child);

 
        return rc;
 }
 
        return rc;
 }


@@ -5580,19 +5604,19 @@ static void lfsck_namespace_scan_local_lpf(const struct lu_env *env,
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct ptlrpc_thread            *thread = &lfsck->li_thread;
        struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct ptlrpc_thread            *thread = &lfsck->li_thread;
        struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;

-       struct dt_device                *dev    = lfsck->li_bottom;

        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct dt_object                *parent;
        const struct dt_it_ops          *iops;
        struct dt_it                    *di;
        struct lfsck_namespace          *ns     = com->lc_file_ram;
        struct dt_object                *parent;
        const struct dt_it_ops          *iops;
        struct dt_it                    *di;

-       struct seq_server_site          *ss     =
-                                       lu_site2seq(dev->dd_lu_dev.ld_site);
+       struct seq_server_site          *ss     = lfsck_dev_site(lfsck);

        __u64                            cookie;
        __u64                            cookie;

+       __u32                            idx    = lfsck_dev_idx(lfsck);

        int                              rc     = 0;
        __u16                            type;
        ENTRY;
 
        int                              rc     = 0;
        __u16                            type;
        ENTRY;
 

-       parent = lfsck_object_find_by_dev(env, dev, &LU_BACKEND_LPF_FID);
+       parent = lfsck_object_find_by_dev(env, lfsck->li_bottom,
+                                         &LU_BACKEND_LPF_FID);

        if (IS_ERR(parent)) {
                CERROR("%s: fail to find backend /lost+found: rc = %ld\n",
                       lfsck_lfsck2name(lfsck), PTR_ERR(parent));
        if (IS_ERR(parent)) {
                CERROR("%s: fail to find backend /lost+found: rc = %ld\n",
                       lfsck_lfsck2name(lfsck), PTR_ERR(parent));


@@ -5652,7 +5676,7 @@ static void lfsck_namespace_scan_local_lpf(const struct lu_env *env,
                                              fid_seq(&ent->lde_fid), range);
                        if (rc != 0)
                                goto skip;
                                              fid_seq(&ent->lde_fid), range);
                        if (rc != 0)
                                goto skip;

-               } else if (lfsck_dev_idx(dev) != 0) {
+               } else if (idx != 0) {

                        /* If the returned FID is IGIF, then there are three
                         * possible cases:
                         *
                        /* If the returned FID is IGIF, then there are three
                         * possible cases:
                         *


@@ -5677,8 +5701,7 @@ static void lfsck_namespace_scan_local_lpf(const struct lu_env *env,
                               "in the backend /lost+found on the MDT %04x, "
                               "to be safe, skip it.\n",
                               lfsck_lfsck2name(lfsck), ent->lde_namelen,
                               "in the backend /lost+found on the MDT %04x, "
                               "to be safe, skip it.\n",
                               lfsck_lfsck2name(lfsck), ent->lde_namelen,

-                              ent->lde_name, PFID(&ent->lde_fid),
-                              lfsck_dev_idx(dev));
+                              ent->lde_name, PFID(&ent->lde_fid), idx);

                        goto skip;
                }
 
                        goto skip;
                }
 


@@ -5718,7 +5741,7 @@ out:
        CDEBUG(D_LFSCK, "%s: stop to scan backend /lost+found: rc = %d\n",
               lfsck_lfsck2name(lfsck), rc);
 
        CDEBUG(D_LFSCK, "%s: stop to scan backend /lost+found: rc = %d\n",
               lfsck_lfsck2name(lfsck), rc);
 

-       lu_object_put(env, &parent->do_lu);
+       lfsck_object_put(env, parent);

 }
 
 /**
 }
 
 /**


@@ -5889,7 +5912,7 @@ lfsck_namespace_double_scan_one_trace_file(const struct lu_env *env,
                        goto checkpoint;
                }
 
                        goto checkpoint;
                }
 

-               target = lfsck_object_find_by_dev(env, lfsck->li_bottom, &fid);
+               target = lfsck_object_find_bottom(env, lfsck, &fid);

                if (IS_ERR(target)) {
                        rc = PTR_ERR(target);
                        goto checkpoint;
                if (IS_ERR(target)) {
                        rc = PTR_ERR(target);
                        goto checkpoint;


@@ -6173,7 +6196,6 @@ struct lfsck_assistant_operations lfsck_namespace_assistant_ops = {
  * entries, then re-generate the linkEA with the given information.
  *
  * \param[in] env      pointer to the thread context
  * entries, then re-generate the linkEA with the given information.
  *
  * \param[in] env      pointer to the thread context

- * \param[in] dev      pointer to the dt_device

  * \param[in] obj      pointer to the dt_object to be handled
  * \param[in] cname    the name for the child in the parent directory
  * \param[in] pfid     the parent directory's FID for the linkEA
  * \param[in] obj      pointer to the dt_object to be handled
  * \param[in] cname    the name for the child in the parent directory
  * \param[in] pfid     the parent directory's FID for the linkEA


@@ -6181,10 +6203,10 @@ struct lfsck_assistant_operations lfsck_namespace_assistant_ops = {
  * \retval             0 for success
  * \retval             negative error number on failure
  */
  * \retval             0 for success
  * \retval             negative error number on failure
  */

-int lfsck_verify_linkea(const struct lu_env *env, struct dt_device *dev,
-                       struct dt_object *obj, const struct lu_name *cname,
-                       const struct lu_fid *pfid)
+int lfsck_verify_linkea(const struct lu_env *env, struct dt_object *obj,
+                       const struct lu_name *cname, const struct lu_fid *pfid)

 {
 {

+       struct dt_device        *dev    = lfsck_obj2dev(obj);

        struct linkea_data       ldata  = { NULL };
        struct lu_buf            linkea_buf;
        struct thandle          *th;
        struct linkea_data       ldata  = { NULL };
        struct lu_buf            linkea_buf;
        struct thandle          *th;


@@ -6288,10 +6310,10 @@ int lfsck_links_get_first(const struct lu_env *env, struct dt_object *obj,
  *
  * \param[in] env      pointer to the thread context
  * \param[in] lfsck    pointer to the lfsck instance
  *
  * \param[in] env      pointer to the thread context
  * \param[in] lfsck    pointer to the lfsck instance

- * \param[in] parent   pointer to the parent directory that holds
+ * \param[in] dir      pointer to the directory that holds

  *                     the name entry
  * \param[in] name     the name for the entry to be updated
  *                     the name entry
  * \param[in] name     the name for the entry to be updated

- * \param[in] pfid     the new PFID for the name entry
+ * \param[in] fid      the new FID for the name entry referenced

  * \param[in] type     the type for the name entry to be updated
  *
  * \retval             0 for success
  * \param[in] type     the type for the name entry to be updated
  *
  * \retval             0 for success


@@ -6299,18 +6321,18 @@ int lfsck_links_get_first(const struct lu_env *env, struct dt_object *obj,
  */
 int lfsck_update_name_entry(const struct lu_env *env,
                            struct lfsck_instance *lfsck,
  */
 int lfsck_update_name_entry(const struct lu_env *env,
                            struct lfsck_instance *lfsck,

-                           struct dt_object *parent, const char *name,
-                           const struct lu_fid *pfid, __u32 type)
+                           struct dt_object *dir, const char *name,
+                           const struct lu_fid *fid, __u32 type)

 {
        struct dt_insert_rec    *rec    = &lfsck_env_info(env)->lti_dt_rec;
 {
        struct dt_insert_rec    *rec    = &lfsck_env_info(env)->lti_dt_rec;

-       struct dt_device        *dev    = lfsck->li_next;
+       struct dt_device        *dev    = lfsck_obj2dev(dir);

        struct lustre_handle     lh     = { 0 };
        struct thandle          *th;
        int                      rc;
        bool                     exists = true;
        ENTRY;
 
        struct lustre_handle     lh     = { 0 };
        struct thandle          *th;
        int                      rc;
        bool                     exists = true;
        ENTRY;
 

-       rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
+       rc = lfsck_ibits_lock(env, lfsck, dir, &lh,

                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                RETURN(rc);
                              MDS_INODELOCK_UPDATE, LCK_EX);
        if (rc != 0)
                RETURN(rc);


@@ -6319,26 +6341,26 @@ int lfsck_update_name_entry(const struct lu_env *env,
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 
        if (IS_ERR(th))
                GOTO(unlock, rc = PTR_ERR(th));
 

-       rc = dt_declare_delete(env, parent, (const struct dt_key *)name, th);
+       rc = dt_declare_delete(env, dir, (const struct dt_key *)name, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        rec->rec_type = type;
        if (rc != 0)
                GOTO(stop, rc);
 
        rec->rec_type = type;

-       rec->rec_fid = pfid;
-       rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
+       rec->rec_fid = fid;
+       rc = dt_declare_insert(env, dir, (const struct dt_rec *)rec,

                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 
                               (const struct dt_key *)name, th);
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_declare_ref_add(env, parent, th);
+       rc = dt_declare_ref_add(env, dir, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_trans_start(env, dev, th);
+       rc = dt_trans_start_local(env, dev, th);

        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_delete(env, parent, (const struct dt_key *)name, th,
+       rc = dt_delete(env, dir, (const struct dt_key *)name, th,

                       BYPASS_CAPA);
        if (rc == -ENOENT) {
                exists = false;
                       BYPASS_CAPA);
        if (rc == -ENOENT) {
                exists = false;


@@ -6348,12 +6370,12 @@ int lfsck_update_name_entry(const struct lu_env *env,
        if (rc != 0)
                GOTO(stop, rc);
 
        if (rc != 0)
                GOTO(stop, rc);
 

-       rc = dt_insert(env, parent, (const struct dt_rec *)rec,
+       rc = dt_insert(env, dir, (const struct dt_rec *)rec,

                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (rc == 0 && S_ISDIR(type) && !exists) {
                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (rc == 0 && S_ISDIR(type) && !exists) {

-               dt_write_lock(env, parent, 0);
-               rc = dt_ref_add(env, parent, th);
-               dt_write_unlock(env, parent);
+               dt_write_lock(env, dir, 0);
+               rc = dt_ref_add(env, dir, th);
+               dt_write_unlock(env, dir);

        }
 
        GOTO(stop, rc);
        }
 
        GOTO(stop, rc);


@@ -6363,10 +6385,9 @@ stop:
 
 unlock:
        lfsck_ibits_unlock(&lh, LCK_EX);
 
 unlock:
        lfsck_ibits_unlock(&lh, LCK_EX);

-

        CDEBUG(D_LFSCK, "%s: update name entry "DFID"/%s with the FID "DFID
               " and the type %o: rc = %d\n", lfsck_lfsck2name(lfsck),
        CDEBUG(D_LFSCK, "%s: update name entry "DFID"/%s with the FID "DFID
               " and the type %o: rc = %d\n", lfsck_lfsck2name(lfsck),

-              PFID(lfsck_dto2fid(parent)), name, PFID(pfid), type, rc);
+              PFID(lfsck_dto2fid(dir)), name, PFID(fid), type, rc);

 
        return rc;
 }
 
        return rc;
 }


@@ -6471,7 +6492,7 @@ int lfsck_namespace_setup(const struct lu_env *env,
 
 out:
        if (root != NULL && !IS_ERR(root))
 
 out:
        if (root != NULL && !IS_ERR(root))

-               lu_object_put(env, &root->do_lu);
+               lfsck_object_put(env, root);

        if (rc != 0) {
                lfsck_component_cleanup(env, com);
                CERROR("%s: fail to init namespace LFSCK component: rc = %d\n",
        if (rc != 0) {
                lfsck_component_cleanup(env, com);
                CERROR("%s: fail to init namespace LFSCK component: rc = %d\n",

diff --git a/lustre/lfsck/lfsck_striped_dir.c b/lustre/lfsck/lfsck_striped_dir.c
index e612140..663bc55 100644 (file)
--- a/lustre/lfsck/lfsck_striped_dir.c
+++ b/lustre/lfsck/lfsck_striped_dir.c
@@ -206,7 +206,7 @@ static int lfsck_disable_master_lmv(const struct lu_env *env,
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct lu_attr                  *la     = &info->lti_la;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lfsck_thread_info        *info   = lfsck_env_info(env);
        struct lu_attr                  *la     = &info->lti_la;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck_obj2dt_dev(obj);
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct thandle                  *th     = NULL;
        int                              rc     = 0;
        ENTRY;
        struct thandle                  *th     = NULL;
        int                              rc     = 0;
        ENTRY;


@@ -342,7 +342,7 @@ static int lfsck_remove_dirent(const struct lu_env *env,
 
        snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2), DFID":%u",
                 PFID(fid), index);
 
        snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2), DFID":%u",
                 PFID(fid), index);

-       obj = lfsck_object_find_by_dev(env, com->lc_lfsck->li_bottom, fid);
+       obj = lfsck_object_find_bottom(env, com->lc_lfsck, fid);

        if (IS_ERR(obj))
                return PTR_ERR(obj);
 
        if (IS_ERR(obj))
                return PTR_ERR(obj);
 


@@ -841,22 +841,12 @@ out:
 int lfsck_read_stripe_lmv(const struct lu_env *env, struct dt_object *obj,
                          struct lmv_mds_md_v1 *lmv)
 {
 int lfsck_read_stripe_lmv(const struct lu_env *env, struct dt_object *obj,
                          struct lmv_mds_md_v1 *lmv)
 {

-       struct dt_object *bottom;
-       int               rc;
+       int rc;

 
 

-       /* Currently, we only store the LMV header on disk. It is the LOD's
-        * duty to iterate the master MDT-object's directory to compose the
-        * integrated LMV EA. But here, we only want to load the LMV header,
-        * so we need to bypass LOD to avoid unnecessary iteration in LOD. */
-       bottom = lu2dt(container_of0(obj->do_lu.lo_header->loh_layers.prev,
-                                    struct lu_object, lo_linkage));
-       if (unlikely(bottom == NULL))
-               return -ENOENT;
-
-       dt_read_lock(env, bottom, 0);
-       rc = dt_xattr_get(env, bottom, lfsck_buf_get(env, lmv, sizeof(*lmv)),
+       dt_read_lock(env, obj, 0);
+       rc = dt_xattr_get(env, obj, lfsck_buf_get(env, lmv, sizeof(*lmv)),

                          XATTR_NAME_LMV, BYPASS_CAPA);
                          XATTR_NAME_LMV, BYPASS_CAPA);

-       dt_read_unlock(env, bottom);
+       dt_read_unlock(env, obj);

        if (rc != sizeof(*lmv))
                return rc > 0 ? -EINVAL : rc;
 
        if (rc != sizeof(*lmv))
                return rc > 0 ? -EINVAL : rc;
 


@@ -1009,7 +999,7 @@ int lfsck_namespace_update_lmv(const struct lu_env *env,
        struct lmv_mds_md_v1            *lmv4   = &info->lti_lmv4;
        struct lu_buf                   *buf    = &info->lti_buf;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lmv_mds_md_v1            *lmv4   = &info->lti_lmv4;
        struct lu_buf                   *buf    = &info->lti_buf;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct dt_device                *dev    = lfsck_obj2dt_dev(obj);
+       struct dt_device                *dev    = lfsck_obj2dev(obj);

        struct thandle                  *th     = NULL;
        struct lustre_handle             lh     = { 0 };
        int                              rc     = 0;
        struct thandle                  *th     = NULL;
        struct lustre_handle             lh     = { 0 };
        int                              rc     = 0;


@@ -1237,7 +1227,7 @@ static int lfsck_namespace_notify_lmv_remote(const struct lu_env *env,
        lr = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
        memset(lr, 0, sizeof(*lr));
        lr->lr_event = event;
        lr = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
        memset(lr, 0, sizeof(*lr));
        lr->lr_event = event;

-       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_index = lfsck_dev_idx(lfsck);

        lr->lr_active = LFSCK_TYPE_NAMESPACE;
        lr->lr_fid = *fid;
        lr->lr_flags = flags;
        lr->lr_active = LFSCK_TYPE_NAMESPACE;
        lr->lr_fid = *fid;
        lr->lr_flags = flags;


@@ -1356,7 +1346,7 @@ int lfsck_namespace_notify_lmv_master_local(const struct lu_env *env,
  *
  * \param[in] env      pointer to the thread context
  * \param[in] com      pointer to the lfsck component
  *
  * \param[in] env      pointer to the thread context
  * \param[in] com      pointer to the lfsck component

- * \param[in] dir      pointer to the object on which the LMV EA will be set
+ * \param[in] obj      pointer to the object on which the LMV EA will be set

  * \param[in] lmv      pointer to the buffer holding the new LMV EA
  * \param[in] cfid     the shard's FID used for verification
  * \param[in] cidx     the shard's index used for verification
  * \param[in] lmv      pointer to the buffer holding the new LMV EA
  * \param[in] cfid     the shard's FID used for verification
  * \param[in] cidx     the shard's index used for verification


@@ -1368,7 +1358,7 @@ int lfsck_namespace_notify_lmv_master_local(const struct lu_env *env,
  */
 static int lfsck_namespace_set_lmv_master(const struct lu_env *env,
                                          struct lfsck_component *com,
  */
 static int lfsck_namespace_set_lmv_master(const struct lu_env *env,
                                          struct lfsck_component *com,

-                                         struct dt_object *dir,
+                                         struct dt_object *obj,

                                          struct lmv_mds_md_v1 *lmv,
                                          const struct lu_fid *cfid,
                                          __u32 cidx, __u32 flags)
                                          struct lmv_mds_md_v1 *lmv,
                                          const struct lu_fid *cfid,
                                          __u32 cidx, __u32 flags)


@@ -1377,20 +1367,12 @@ static int lfsck_namespace_set_lmv_master(const struct lu_env *env,
        struct lmv_mds_md_v1            *lmv3   = &info->lti_lmv3;
        struct lu_seq_range             *range  = &info->lti_range;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;
        struct lmv_mds_md_v1            *lmv3   = &info->lti_lmv3;
        struct lu_seq_range             *range  = &info->lti_range;
        struct lfsck_instance           *lfsck  = com->lc_lfsck;

-       struct seq_server_site          *ss     =
-                       lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
-       struct dt_object                *obj;
+       struct seq_server_site          *ss     = lfsck_dev_site(lfsck);

        struct lustre_handle             lh     = { 0 };
        int                              pidx   = -1;
        int                              rc     = 0;
        ENTRY;
 
        struct lustre_handle             lh     = { 0 };
        int                              pidx   = -1;
        int                              rc     = 0;
        ENTRY;
 

-       /* Find the bottom object to bypass LOD when set LMV EA. */
-       obj = lu2dt(container_of0(dir->do_lu.lo_header->loh_layers.prev,
-                                 struct lu_object, lo_linkage));
-       if (unlikely(obj == NULL))
-               RETURN(-ENOENT);
-

        fld_range_set_mdt(range);
        rc = fld_server_lookup(env, ss->ss_server_fld,
                               fid_seq(lfsck_dto2fid(obj)), range);
        fld_range_set_mdt(range);
        rc = fld_server_lookup(env, ss->ss_server_fld,
                               fid_seq(lfsck_dto2fid(obj)), range);


@@ -1524,7 +1506,7 @@ log:
        CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found bad name hash "
               "on the MDT %x, parent "DFID", name %s, shard_%x "DFID
               ": rc = %d\n",
        CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found bad name hash "
               "on the MDT %x, parent "DFID", name %s, shard_%x "DFID
               ": rc = %d\n",

-              lfsck_lfsck2name(lfsck), lfsck_dev_idx(lfsck->li_bottom),
+              lfsck_lfsck2name(lfsck), lfsck_dev_idx(lfsck),

               PFID(pfid), name, llmv->ll_lmv.lmv_master_mdt_index,
               PFID(lfsck_dto2fid(shard)), rc);
 
               PFID(pfid), name, llmv->ll_lmv.lmv_master_mdt_index,
               PFID(lfsck_dto2fid(shard)), rc);
 


@@ -1714,7 +1696,7 @@ int lfsck_namespace_verify_stripe_slave(const struct lu_env *env,
                GOTO(out, rc);
        }
 
                GOTO(out, rc);
        }
 

-       parent = lfsck_object_find(env, lfsck, pfid);
+       parent = lfsck_object_find_bottom(env, lfsck, pfid);

        if (IS_ERR(parent)) {
                rc = lfsck_namespace_trace_update(env, com, cfid,
                                        LNTF_UNCERTAIN_LMV, true);
        if (IS_ERR(parent)) {
                rc = lfsck_namespace_trace_update(env, com, cfid,
                                        LNTF_UNCERTAIN_LMV, true);


@@ -1837,8 +1819,7 @@ int lfsck_namespace_striped_dir_rescan(const struct lu_env *env,
        struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
        const struct lu_fid             *pfid   = lfsck_dto2fid(dir);
        struct lu_seq_range             *range  = &info->lti_range;
        struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
        const struct lu_fid             *pfid   = lfsck_dto2fid(dir);
        struct lu_seq_range             *range  = &info->lti_range;

-       struct seq_server_site          *ss     =
-                       lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
+       struct seq_server_site          *ss     = lfsck_dev_site(lfsck);

        __u32                            stripe_count;
        __u32                            hash_type;
        int                              rc     = 0;
        __u32                            stripe_count;
        __u32                            hash_type;
        int                              rc     = 0;


@@ -2214,11 +2195,11 @@ int lfsck_namespace_handle_striped_master(const struct lu_env *env,
        if (shard_idx < 0)
                GOTO(fail_lmv, rc = shard_idx);
 
        if (shard_idx < 0)
                GOTO(fail_lmv, rc = shard_idx);
 

-       if (shard_idx == lfsck_dev_idx(lfsck->li_bottom)) {
+       if (shard_idx == lfsck_dev_idx(lfsck)) {

                if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
                        GOTO(out, rc = 0);
 
                if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
                        GOTO(out, rc = 0);
 

-               dev = lfsck->li_next;
+               dev = lfsck->li_bottom;

        } else {
                struct lfsck_tgt_desc *ltd;
 
        } else {
                struct lfsck_tgt_desc *ltd;
 


@@ -2336,7 +2317,7 @@ out:
                if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
                     rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
                     rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&
                if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
                     rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
                     rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&

-                   dev != NULL && dev != lfsck->li_next)
+                   dev != NULL && dev != lfsck->li_bottom)

                        lfsck_lad_set_bitmap(env, com, shard_idx);
 
                if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT))
                        lfsck_lad_set_bitmap(env, com, shard_idx);
 
                if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT))

diff --git a/lustre/ofd/ofd_io.c b/lustre/ofd/ofd_io.c
index 1de6028..269a723 100644 (file)
--- a/lustre/ofd/ofd_io.c
+++ b/lustre/ofd/ofd_io.c
@@ -119,7 +119,7 @@ static void ofd_inconsistency_verify_one(const struct lu_env *env,
                               ", the old stored PFID "DFID"\n",
                               ofd_name(ofd), PFID(&fo->ofo_header.loh_fid),
                               PFID(&oii->oii_pfid), PFID(pfid));
                               ", the old stored PFID "DFID"\n",
                               ofd_name(ofd), PFID(&fo->ofo_header.loh_fid),
                               PFID(&oii->oii_pfid), PFID(pfid));

-               } else {
+               } else if (rc < 0) {

                        CDEBUG(D_LFSCK, "%s: fail to fix the OST PFID xattr "
                               "for "DFID", client given PFID "DFID", local "
                               "stored PFID "DFID": rc = %d\n",
                        CDEBUG(D_LFSCK, "%s: fail to fix the OST PFID xattr "
                               "for "DFID", client given PFID "DFID", local "
                               "stored PFID "DFID": rc = %d\n",

diff --git a/lustre/tests/sanity-lfsck.sh b/lustre/tests/sanity-lfsck.sh
index b89e2d2..9fb7dba 100644 (file)
--- a/lustre/tests/sanity-lfsck.sh
+++ b/lustre/tests/sanity-lfsck.sh
@@ -2832,7 +2832,7 @@ test_22a() {
        echo "#####"
        echo "The parent_A references the child directory via some name entry,"
        echo "but the child directory back references another parent_B via its"
        echo "#####"
        echo "The parent_A references the child directory via some name entry,"
        echo "but the child directory back references another parent_B via its"

-       echo "".." name entry. The parent_B does not exist. Then the namesapce"
+       echo "".." name entry. The parent_B does not exist. Then the namespace"

        echo "LFSCK will repair the child directory's ".." name entry."
        echo "#####"
 
        echo "LFSCK will repair the child directory's ".." name entry."
        echo "#####"
 


@@ -2881,7 +2881,7 @@ test_22b() {
        echo "The parent_A references the child directory via the name entry_B,"
        echo "but the child directory back references another parent_C via its"
        echo "".." name entry. The parent_C exists, but there is no the name"
        echo "The parent_A references the child directory via the name entry_B,"
        echo "but the child directory back references another parent_C via its"
        echo "".." name entry. The parent_C exists, but there is no the name"

-       echo "entry_B under the parent_C. Then the namesapce LFSCK will repair"
+       echo "entry_B under the parent_C. Then the namespace LFSCK will repair"

        echo "the child directory's ".." name entry and its linkEA."
        echo "#####"
 
        echo "the child directory's ".." name entry and its linkEA."
        echo "#####"