LU-1267 lfsck: framework (2) for MDT-OST consistency

The LFSCK can talk with OSP directly, then the LFSCK on the MDT can
control/monitor the specified LFSCK instance on other targets (MDTs
or/and OSTs) without breaking dt_device APIs nor making OSP to know
the LFSCK things, and simplify the handling of remote OST-object or
MDT-object. For that, each OSP will register to the LFSCK will they
are added into the system. The LFSCK maintains such target table in
RAM with the similar logic as the LOD does.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: Ifa14db68925a0cd2afe0c3566382dbb6176d50b2
Reviewed-on: http://review.whamcloud.com/8302
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_lfsck.h b/lustre/include/lustre_lfsck.h
index b20eff9..1be54ec 100644 (file)
--- a/lustre/include/lustre_lfsck.h
+++ b/lustre/include/lustre_lfsck.h
@@ -118,6 +118,12 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
                   void *notify_data, bool master);
 void lfsck_degister(const struct lu_env *env, struct dt_device *key);
 
+int lfsck_add_target(const struct lu_env *env, struct dt_device *key,
+                    struct dt_device *tgt, struct obd_export *exp,
+                    __u32 index, bool for_ost);
+void lfsck_del_target(const struct lu_env *env, struct dt_device *key,
+                     struct dt_device *tgt, __u32 index, bool for_ost);
+
 int lfsck_start(const struct lu_env *env, struct dt_device *key,
                struct lfsck_start_param *lsp);
 int lfsck_stop(const struct lu_env *env, struct dt_device *key,

diff --git a/lustre/lfsck/lfsck_internal.h b/lustre/lfsck/lfsck_internal.h
index 1c16e06..e0366b6 100644 (file)
--- a/lustre/lfsck/lfsck_internal.h
+++ b/lustre/lfsck/lfsck_internal.h
@@ -292,12 +292,57 @@ struct lfsck_operations {
                                   struct lfsck_component *com);
 };
 
+#define TGT_PTRS               256     /* number of pointers at 1st level */
+#define TGT_PTRS_PER_BLOCK     256     /* number of pointers at 2nd level */
+
+struct lfsck_tgt_desc {
+       struct list_head   ltd_orphan_list;
+       struct dt_device  *ltd_tgt;
+       struct dt_device  *ltd_key;
+       struct obd_export *ltd_exp;
+       struct list_head   ltd_layout_list;
+       atomic_t           ltd_ref;
+       __u32              ltd_index;
+};
+
+struct lfsck_tgt_desc_idx {
+       struct lfsck_tgt_desc *ldi_tgts[TGT_PTRS_PER_BLOCK];
+};
+
+struct lfsck_tgt_descs {
+       /* list of known TGTs */
+       struct lfsck_tgt_desc_idx       *ltd_tgts_idx[TGT_PTRS];
+
+       /* bitmap of TGTs available */
+       cfs_bitmap_t                    *ltd_tgts_bitmap;
+
+       /* for lfsck_tgt_desc::ltd_xxx_list */
+       spinlock_t                       ltd_lock;
+
+       /* for tgts table accessing and changes */
+       struct rw_semaphore              ltd_rw_sem;
+
+       /* Temporary list for orphan targets. */
+       struct list_head                 ltd_orphan;
+
+       /* number of registered TGTs */
+       int                              ltd_tgtnr;
+};
+
+#define LTD_TGT(ltd, index)    \
+       ((ltd)->ltd_tgts_idx[(index) / TGT_PTRS_PER_BLOCK]->\
+        ldi_tgts[(index) % TGT_PTRS_PER_BLOCK])
+
+#define OST_TGT(lfsck, index)   LTD_TGT(&lfsck->li_ost_descs, index)
+#define MDT_TGT(lfsck, index)   LTD_TGT(&lfsck->li_mdt_descs, index)
+
 struct lfsck_component {
        /* into lfsck_instance::li_list_(scan,double_scan,idle} */
        cfs_list_t               lc_link;
 
        /* into lfsck_instance::li_list_dir */
        cfs_list_t               lc_link_dir;
+
        struct rw_semaphore      lc_sem;
        atomic_t                 lc_ref;
 
@@ -382,6 +427,12 @@ struct lfsck_instance {
        /* It for directory traversal */
        struct dt_it             *li_di_dir;
 
+       /* Description of OST */
+       struct lfsck_tgt_descs    li_ost_descs;
+
+       /* Description of MDT */
+       struct lfsck_tgt_descs    li_mdt_descs;
+
        /* namespace-based directory traversal position. */
        __u64                     li_cookie_dir;
 
@@ -636,6 +687,24 @@ static inline void lfsck_object_put(const struct lu_env *env,
        lu_object_put(env, &obj->do_lu);
 }
 
+static inline struct lfsck_tgt_desc *lfsck_tgt_get(struct lfsck_tgt_descs *ltds,
+                                                  __u32 index)
+{
+       struct lfsck_tgt_desc *ltd;
+
+       ltd = LTD_TGT(ltds, index);
+       if (ltd != NULL)
+               atomic_inc(&ltd->ltd_ref);
+
+       return ltd;
+}
+
+static inline void lfsck_tgt_put(struct lfsck_tgt_desc *ltd)
+{
+       if (atomic_dec_and_test(&ltd->ltd_ref))
+               OBD_FREE_PTR(ltd);
+}
+
 static inline struct lfsck_component *
 lfsck_component_get(struct lfsck_component *com)
 {

diff --git a/lustre/lfsck/lfsck_lib.c b/lustre/lfsck/lfsck_lib.c
index bc703ea..d4811bd 100644 (file)
--- a/lustre/lfsck/lfsck_lib.c
+++ b/lustre/lfsck/lfsck_lib.c
@@ -58,6 +58,8 @@ LU_CONTEXT_KEY_DEFINE(lfsck, LCT_MD_THREAD | LCT_DT_THREAD);
 LU_KEY_INIT_GENERIC(lfsck);
 
 static CFS_LIST_HEAD(lfsck_instance_list);
+static struct list_head lfsck_ost_orphan_list;
+static struct list_head lfsck_mdt_orphan_list;
 static DEFINE_SPINLOCK(lfsck_instance_lock);
 
 static const char *lfsck_status_names[] = {
@@ -96,6 +98,177 @@ const char *lfsck_status2names(enum lfsck_status status)
        return lfsck_status_names[status];
 }
 
+static int lfsck_tgt_descs_init(struct lfsck_tgt_descs *ltds)
+{
+       spin_lock_init(&ltds->ltd_lock);
+       init_rwsem(&ltds->ltd_rw_sem);
+       INIT_LIST_HEAD(&ltds->ltd_orphan);
+       ltds->ltd_tgts_bitmap = CFS_ALLOCATE_BITMAP(BITS_PER_LONG);
+       if (ltds->ltd_tgts_bitmap == NULL)
+               return -ENOMEM;
+
+       return 0;
+}
+
+static void lfsck_tgt_descs_fini(struct lfsck_tgt_descs *ltds)
+{
+       struct lfsck_tgt_desc   *ltd;
+       struct lfsck_tgt_desc   *next;
+       int                      idx;
+
+       down_write(&ltds->ltd_rw_sem);
+
+       list_for_each_entry_safe(ltd, next, &ltds->ltd_orphan,
+                                ltd_orphan_list) {
+               list_del_init(&ltd->ltd_orphan_list);
+               lfsck_tgt_put(ltd);
+       }
+
+       if (unlikely(ltds->ltd_tgts_bitmap == NULL)) {
+               up_write(&ltds->ltd_rw_sem);
+
+               return;
+       }
+
+       cfs_foreach_bit(ltds->ltd_tgts_bitmap, idx) {
+               ltd = LTD_TGT(ltds, idx);
+               if (likely(ltd != NULL)) {
+                       LASSERT(list_empty(&ltd->ltd_layout_list));
+
+                       ltds->ltd_tgtnr--;
+                       cfs_bitmap_clear(ltds->ltd_tgts_bitmap, idx);
+                       LTD_TGT(ltds, idx) = NULL;
+                       lfsck_tgt_put(ltd);
+               }
+       }
+
+       LASSERTF(ltds->ltd_tgtnr == 0, "tgt count unmatched: %d\n",
+                ltds->ltd_tgtnr);
+
+       for (idx = 0; idx < TGT_PTRS; idx++) {
+               if (ltds->ltd_tgts_idx[idx] != NULL) {
+                       OBD_FREE_PTR(ltds->ltd_tgts_idx[idx]);
+                       ltds->ltd_tgts_idx[idx] = NULL;
+               }
+       }
+
+       CFS_FREE_BITMAP(ltds->ltd_tgts_bitmap);
+       ltds->ltd_tgts_bitmap = NULL;
+       up_write(&ltds->ltd_rw_sem);
+}
+
+static int __lfsck_add_target(const struct lu_env *env,
+                             struct lfsck_instance *lfsck,
+                             struct lfsck_tgt_desc *ltd,
+                             bool for_ost, bool locked)
+{
+       struct lfsck_tgt_descs *ltds;
+       __u32                   index = ltd->ltd_index;
+       int                     rc    = 0;
+       ENTRY;
+
+       if (for_ost)
+               ltds = &lfsck->li_ost_descs;
+       else
+               ltds = &lfsck->li_mdt_descs;
+
+       if (!locked)
+               down_write(&ltds->ltd_rw_sem);
+
+       LASSERT(ltds->ltd_tgts_bitmap != NULL);
+
+       if (index >= ltds->ltd_tgts_bitmap->size) {
+               __u32 newsize = max((__u32)ltds->ltd_tgts_bitmap->size,
+                                   (__u32)BITS_PER_LONG);
+               cfs_bitmap_t *old_bitmap = ltds->ltd_tgts_bitmap;
+               cfs_bitmap_t *new_bitmap;
+
+               while (newsize < index + 1)
+                       newsize <<= 1;
+
+               new_bitmap = CFS_ALLOCATE_BITMAP(newsize);
+               if (new_bitmap == NULL)
+                       GOTO(unlock, rc = -ENOMEM);
+
+               if (ltds->ltd_tgtnr > 0)
+                       cfs_bitmap_copy(new_bitmap, old_bitmap);
+               ltds->ltd_tgts_bitmap = new_bitmap;
+               CFS_FREE_BITMAP(old_bitmap);
+       }
+
+       if (cfs_bitmap_check(ltds->ltd_tgts_bitmap, index)) {
+               CERROR("%s: the device %s (%u) is registered already\n",
+                      lfsck_lfsck2name(lfsck),
+                      ltd->ltd_tgt->dd_lu_dev.ld_obd->obd_name, index);
+               GOTO(unlock, rc = -EEXIST);
+       }
+
+       if (ltds->ltd_tgts_idx[index / TGT_PTRS_PER_BLOCK] == NULL) {
+               OBD_ALLOC_PTR(ltds->ltd_tgts_idx[index / TGT_PTRS_PER_BLOCK]);
+               if (ltds->ltd_tgts_idx[index / TGT_PTRS_PER_BLOCK] == NULL)
+                       GOTO(unlock, rc = -ENOMEM);
+       }
+
+       LTD_TGT(ltds, index) = ltd;
+       cfs_bitmap_set(ltds->ltd_tgts_bitmap, index);
+       ltds->ltd_tgtnr++;
+
+       GOTO(unlock, rc = 0);
+
+unlock:
+       if (!locked)
+               up_write(&ltds->ltd_rw_sem);
+
+       return rc;
+}
+
+static int lfsck_add_target_from_orphan(const struct lu_env *env,
+                                       struct lfsck_instance *lfsck)
+{
+       struct lfsck_tgt_descs  *ltds    = &lfsck->li_ost_descs;
+       struct lfsck_tgt_desc   *ltd;
+       struct lfsck_tgt_desc   *next;
+       struct list_head        *head    = &lfsck_ost_orphan_list;
+       int                      rc;
+       bool                     for_ost = true;
+
+again:
+       spin_lock(&lfsck_instance_lock);
+       list_for_each_entry_safe(ltd, next, head, ltd_orphan_list) {
+               if (ltd->ltd_key == lfsck->li_bottom) {
+                       list_del_init(&ltd->ltd_orphan_list);
+                       list_add_tail(&ltd->ltd_orphan_list,
+                                     &ltds->ltd_orphan);
+               }
+       }
+       spin_unlock(&lfsck_instance_lock);
+
+       down_write(&ltds->ltd_rw_sem);
+       while (!list_empty(&ltds->ltd_orphan)) {
+               ltd = list_entry(ltds->ltd_orphan.next,
+                                struct lfsck_tgt_desc,
+                                ltd_orphan_list);
+               list_del_init(&ltd->ltd_orphan_list);
+               rc = __lfsck_add_target(env, lfsck, ltd, for_ost, true);
+               /* Do not hold the semaphore for too long time. */
+               up_write(&ltds->ltd_rw_sem);
+               if (rc != 0)
+                       return rc;
+
+               down_write(&ltds->ltd_rw_sem);
+       }
+       up_write(&ltds->ltd_rw_sem);
+
+       if (for_ost) {
+               ltds = &lfsck->li_mdt_descs;
+               head = &lfsck_mdt_orphan_list;
+               for_ost = false;
+               goto again;
+       }
+
+       return 0;
+}
+
 static inline struct lfsck_component *
 __lfsck_component_find(struct lfsck_instance *lfsck, __u16 type, cfs_list_t *list)
 {
@@ -153,6 +326,9 @@ void lfsck_instance_cleanup(const struct lu_env *env,
        LASSERT(list_empty(&lfsck->li_link));
        LASSERT(thread_is_init(thread) || thread_is_stopped(thread));
 
+       lfsck_tgt_descs_fini(&lfsck->li_ost_descs);
+       lfsck_tgt_descs_fini(&lfsck->li_mdt_descs);
+
        if (lfsck->li_obj_oit != NULL) {
                lu_object_put_nocache(env, &lfsck->li_obj_oit->do_lu);
                lfsck->li_obj_oit = NULL;
@@ -196,26 +372,37 @@ void lfsck_instance_cleanup(const struct lu_env *env,
        OBD_FREE_PTR(lfsck);
 }
 
-static inline struct lfsck_instance *lfsck_instance_find(struct dt_device *key,
-                                                        bool ref, bool unlink)
+static inline struct lfsck_instance *
+__lfsck_instance_find(struct dt_device *key, bool ref, bool unlink)
 {
        struct lfsck_instance *lfsck;
 
-       spin_lock(&lfsck_instance_lock);
        cfs_list_for_each_entry(lfsck, &lfsck_instance_list, li_link) {
                if (lfsck->li_bottom == key) {
                        if (ref)
                                lfsck_instance_get(lfsck);
                        if (unlink)
                                list_del_init(&lfsck->li_link);
-                       spin_unlock(&lfsck_instance_lock);
+
                        return lfsck;
                }
        }
-       spin_unlock(&lfsck_instance_lock);
+
        return NULL;
 }
 
+static inline struct lfsck_instance *lfsck_instance_find(struct dt_device *key,
+                                                        bool ref, bool unlink)
+{
+       struct lfsck_instance *lfsck;
+
+       spin_lock(&lfsck_instance_lock);
+       lfsck = __lfsck_instance_find(key, ref, unlink);
+       spin_unlock(&lfsck_instance_lock);
+
+       return lfsck;
+}
+
 static inline int lfsck_instance_add(struct lfsck_instance *lfsck)
 {
        struct lfsck_instance *tmp;
@@ -1152,6 +1339,14 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
        lfsck->li_next = next;
        lfsck->li_bottom = key;
 
+       rc = lfsck_tgt_descs_init(&lfsck->li_ost_descs);
+       if (rc != 0)
+               GOTO(out, rc);
+
+       rc = lfsck_tgt_descs_init(&lfsck->li_mdt_descs);
+       if (rc != 0)
+               GOTO(out, rc);
+
        fid->f_seq = FID_SEQ_LOCAL_NAME;
        fid->f_oid = 1;
        fid->f_ver = 0;
@@ -1216,6 +1411,8 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
 
 add:
        rc = lfsck_instance_add(lfsck);
+       if (rc == 0)
+               rc = lfsck_add_target_from_orphan(env, lfsck);
 out:
        if (root != NULL && !IS_ERR(root))
                lu_object_put(env, &root->do_lu);
@@ -1235,19 +1432,164 @@ void lfsck_degister(const struct lu_env *env, struct dt_device *key)
 }
 EXPORT_SYMBOL(lfsck_degister);
 
+int lfsck_add_target(const struct lu_env *env, struct dt_device *key,
+                    struct dt_device *tgt, struct obd_export *exp,
+                    __u32 index, bool for_ost)
+{
+       struct lfsck_instance   *lfsck;
+       struct lfsck_tgt_desc   *ltd;
+       int                      rc;
+       ENTRY;
+
+       OBD_ALLOC_PTR(ltd);
+       if (ltd == NULL)
+               RETURN(-ENOMEM);
+
+       ltd->ltd_tgt = tgt;
+       ltd->ltd_key = key;
+       ltd->ltd_exp = exp;
+       INIT_LIST_HEAD(&ltd->ltd_orphan_list);
+       INIT_LIST_HEAD(&ltd->ltd_layout_list);
+       atomic_set(&ltd->ltd_ref, 1);
+       ltd->ltd_index = index;
+
+       spin_lock(&lfsck_instance_lock);
+       lfsck = __lfsck_instance_find(key, true, false);
+       if (lfsck == NULL) {
+               if (for_ost)
+                       list_add_tail(&ltd->ltd_orphan_list,
+                                     &lfsck_ost_orphan_list);
+               else
+                       list_add_tail(&ltd->ltd_orphan_list,
+                                     &lfsck_mdt_orphan_list);
+               spin_unlock(&lfsck_instance_lock);
+
+               RETURN(0);
+       }
+       spin_unlock(&lfsck_instance_lock);
+
+       rc = __lfsck_add_target(env, lfsck, ltd, for_ost, false);
+       if (rc != 0)
+               lfsck_tgt_put(ltd);
+
+       lfsck_instance_put(env, lfsck);
+
+       RETURN(rc);
+}
+EXPORT_SYMBOL(lfsck_add_target);
+
+void lfsck_del_target(const struct lu_env *env, struct dt_device *key,
+                     struct dt_device *tgt, __u32 index, bool for_ost)
+{
+       struct lfsck_instance   *lfsck;
+       struct lfsck_tgt_descs  *ltds;
+       struct lfsck_tgt_desc   *ltd;
+       struct list_head        *head;
+       bool                     found = false;
+
+       if (for_ost)
+               head = &lfsck_ost_orphan_list;
+       else
+               head = &lfsck_mdt_orphan_list;
+
+       spin_lock(&lfsck_instance_lock);
+       list_for_each_entry(ltd, head, ltd_orphan_list) {
+               if (ltd->ltd_tgt == tgt) {
+                       list_del_init(&ltd->ltd_orphan_list);
+                       spin_unlock(&lfsck_instance_lock);
+                       lfsck_tgt_put(ltd);
+
+                       return;
+               }
+       }
+
+       lfsck = __lfsck_instance_find(key, true, false);
+       spin_unlock(&lfsck_instance_lock);
+       if (unlikely(lfsck == NULL))
+               return;
+
+       if (for_ost)
+               ltds = &lfsck->li_ost_descs;
+       else
+               ltds = &lfsck->li_mdt_descs;
+
+       down_write(&ltds->ltd_rw_sem);
+
+       LASSERT(ltds->ltd_tgts_bitmap != NULL);
+
+       if (unlikely(index >= ltds->ltd_tgts_bitmap->size))
+               goto unlock;
+
+       ltd = LTD_TGT(ltds, index);
+       if (unlikely(ltd == NULL))
+               goto unlock;
+
+       found = true;
+       if (!list_empty(&ltd->ltd_layout_list)) {
+               spin_lock(&ltds->ltd_lock);
+               list_del_init(&ltd->ltd_layout_list);
+               spin_unlock(&ltds->ltd_lock);
+       }
+
+       LASSERT(ltds->ltd_tgtnr > 0);
+
+       ltds->ltd_tgtnr--;
+       cfs_bitmap_clear(ltds->ltd_tgts_bitmap, index);
+       LTD_TGT(ltds, index) = NULL;
+       lfsck_tgt_put(ltd);
+
+unlock:
+       if (!found) {
+               if (for_ost)
+                       head = &lfsck->li_ost_descs.ltd_orphan;
+               else
+                       head = &lfsck->li_ost_descs.ltd_orphan;
+
+               list_for_each_entry(ltd, head, ltd_orphan_list) {
+                       if (ltd->ltd_tgt == tgt) {
+                               list_del_init(&ltd->ltd_orphan_list);
+                               lfsck_tgt_put(ltd);
+                               break;
+                       }
+               }
+       }
+
+       up_write(&ltds->ltd_rw_sem);
+       lfsck_instance_put(env, lfsck);
+}
+EXPORT_SYMBOL(lfsck_del_target);
+
 static int __init lfsck_init(void)
 {
        int rc;
 
+       INIT_LIST_HEAD(&lfsck_ost_orphan_list);
+       INIT_LIST_HEAD(&lfsck_mdt_orphan_list);
        lfsck_key_init_generic(&lfsck_thread_key, NULL);
        rc = lu_context_key_register(&lfsck_thread_key);
+
        return rc;
 }
 
 static void __exit lfsck_exit(void)
 {
+       struct lfsck_tgt_desc *ltd;
+       struct lfsck_tgt_desc *next;
+
        LASSERT(cfs_list_empty(&lfsck_instance_list));
 
+       list_for_each_entry_safe(ltd, next, &lfsck_ost_orphan_list,
+                                ltd_orphan_list) {
+               list_del_init(&ltd->ltd_orphan_list);
+               lfsck_tgt_put(ltd);
+       }
+
+       list_for_each_entry_safe(ltd, next, &lfsck_mdt_orphan_list,
+                                ltd_orphan_list) {
+               list_del_init(&ltd->ltd_orphan_list);
+               lfsck_tgt_put(ltd);
+       }
+
        lu_context_key_degister(&lfsck_thread_key);
 }
 

diff --git a/lustre/lod/lod_dev.c b/lustre/lod/lod_dev.c
index e36aee1..02a1e1e 100644 (file)
--- a/lustre/lod/lod_dev.c
+++ b/lustre/lod/lod_dev.c
@@ -290,7 +290,7 @@ static int lod_process_config(const struct lu_env *env,
                } else {
                        rc = lod_del_device(env, lod,
                                            &lod->lod_ost_descs,
-                                           arg1, index, gen);
+                                           arg1, index, gen, true);
                }
 
                break;
@@ -750,12 +750,12 @@ static struct lu_device *lod_device_fini(const struct lu_env *env,
 
        lod_procfs_fini(lod);
 
-       rc = lod_fini_tgt(lod, &lod->lod_ost_descs);
+       rc = lod_fini_tgt(env, lod, &lod->lod_ost_descs, true);
        if (rc)
                CERROR("%s:can not fini ost descs %d\n",
                        lod2obd(lod)->obd_name, rc);
 
-       rc = lod_fini_tgt(lod, &lod->lod_mdt_descs);
+       rc = lod_fini_tgt(env, lod, &lod->lod_mdt_descs, false);
        if (rc)
                CERROR("%s:can not fini mdt descs %d\n",
                        lod2obd(lod)->obd_name, rc);

diff --git a/lustre/lod/lod_internal.h b/lustre/lod/lod_internal.h
index ab4b274..6d99ad5 100644 (file)
--- a/lustre/lod/lod_internal.h
+++ b/lustre/lod/lod_internal.h
@@ -331,8 +331,9 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
                   char *type, int active);
 int lod_del_device(const struct lu_env *env, struct lod_device *lod,
                   struct lod_tgt_descs *ltd, char *osp, unsigned idx,
-                  unsigned gen);
-int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd);
+                  unsigned gen, bool for_ost);
+int lod_fini_tgt(const struct lu_env *env, struct lod_device *lod,
+                struct lod_tgt_descs *ltd, bool for_ost);
 int lod_load_striping(const struct lu_env *env, struct lod_object *mo);
 int lod_get_lov_ea(const struct lu_env *env, struct lod_object *mo);
 void lod_fix_desc(struct lov_desc *desc);

diff --git a/lustre/lod/lod_lov.c b/lustre/lod/lod_lov.c
index e60ba8b..2c93223 100644 (file)
--- a/lustre/lod/lod_lov.c
+++ b/lustre/lod/lod_lov.c
@@ -35,6 +35,7 @@
 
 #include <obd_class.h>
 #include <obd_lov.h>
+#include <lustre_lfsck.h>
 
 #include "lod_internal.h"
 
@@ -184,6 +185,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        struct lod_tgt_desc     *tgt_desc;
        struct lod_tgt_descs    *ltd;
        struct obd_uuid         obd_uuid;
+       bool                    for_ost;
        ENTRY;
 
        CDEBUG(D_CONFIG, "osp:%s idx:%d gen:%d\n", osp, index, gen);
@@ -212,6 +214,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        data->ocd_index = index;
 
        if (strcmp(LUSTRE_OSC_NAME, type) == 0) {
+               for_ost = true;
                data->ocd_connect_flags |= OBD_CONNECT_AT |
                                           OBD_CONNECT_FULL20 |
                                           OBD_CONNECT_INDEX |
@@ -232,6 +235,7 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        } else {
                struct obd_import *imp = obd->u.cli.cl_import;
 
+               for_ost = false;
                data->ocd_ibits_known = MDS_INODELOCK_UPDATE;
                data->ocd_connect_flags |= OBD_CONNECT_ACL |
                                           OBD_CONNECT_MDS_CAPA |
@@ -347,7 +351,12 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
        if (lod->lod_recovery_completed)
                ldev->ld_ops->ldo_recovery_complete(env, ldev);
 
-       RETURN(0);
+       rc = lfsck_add_target(env, lod->lod_child, d, exp, index, for_ost);
+       if (rc != 0)
+               CERROR("Fail to add LFSCK target: name = %s, type = %s, "
+                      "index = %u, rc = %d\n", osp, type, index, rc);
+
+       RETURN(rc);
 
 out_pool:
        lod_ost_pool_remove(&lod->lod_pool_info, index);
@@ -365,17 +374,23 @@ out_free:
 /*
  * helper function to schedule OST removal from the device table
  */
-static void __lod_del_device(struct lod_tgt_descs *ltd,
-                            unsigned idx)
+static void __lod_del_device(const struct lu_env *env, struct lod_device *lod,
+                            struct lod_tgt_descs *ltd, unsigned idx,
+                            bool for_ost)
 {
        LASSERT(LTD_TGT(ltd, idx));
+
+       lfsck_del_target(env, lod->lod_child, LTD_TGT(ltd, idx)->ltd_tgt,
+                        idx, for_ost);
+
        if (LTD_TGT(ltd, idx)->ltd_reap == 0) {
                LTD_TGT(ltd, idx)->ltd_reap = 1;
                ltd->ltd_death_row++;
        }
 }
 
-int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd)
+int lod_fini_tgt(const struct lu_env *env, struct lod_device *lod,
+                struct lod_tgt_descs *ltd, bool for_ost)
 {
        int idx;
 
@@ -384,7 +399,7 @@ int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd)
        lod_getref(ltd);
        mutex_lock(&ltd->ltd_mutex);
        cfs_foreach_bit(ltd->ltd_tgt_bitmap, idx)
-               __lod_del_device(ltd, idx);
+               __lod_del_device(env, lod, ltd, idx, for_ost);
        mutex_unlock(&ltd->ltd_mutex);
        lod_putref(lod, ltd);
        CFS_FREE_BITMAP(ltd->ltd_tgt_bitmap);
@@ -411,7 +426,7 @@ int lod_fini_tgt(struct lod_device *lod, struct lod_tgt_descs *ltd)
  */
 int lod_del_device(const struct lu_env *env, struct lod_device *lod,
                   struct lod_tgt_descs *ltd, char *osp, unsigned idx,
-                  unsigned gen)
+                  unsigned gen, bool for_ost)
 {
        struct obd_device *obd;
        int                rc = 0;
@@ -454,7 +469,7 @@ int lod_del_device(const struct lu_env *env, struct lod_device *lod,
                GOTO(out, rc = -EINVAL);
        }
 
-       __lod_del_device(ltd, idx);
+       __lod_del_device(env, lod, ltd, idx, for_ost);
        EXIT;
 out:
        mutex_unlock(&ltd->ltd_mutex);