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

The LFSCK on the MDT uses LFSCK_NOTIFY RPC to control the LFSCK
on the OSTs (or other MDTs) to start/stop/fail/pause the layout
consistency verification. And uses LFSCK_QUERY RPC to query the
LFSCK status on the OSTs (or other MDTs).

Introduce new connection flag: OBD_CONNECT_LFSCK to indicate that
whether the target server (MDT/OST) supports online LFSCK or not.

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

diff --git a/lustre/include/lustre/lustre_idl.h b/lustre/include/lustre/lustre_idl.h
index d1f867f..4bb3e7c 100644 (file)
--- a/lustre/include/lustre/lustre_idl.h
+++ b/lustre/include/lustre/lustre_idl.h
@@ -1331,6 +1331,7 @@ extern void lustre_swab_ptlrpc_body(struct ptlrpc_body *pb);
 #define OBD_CONNECT_DISP_STRIPE 0x10000000000000ULL/* create stripe disposition*/
 #define OBD_CONNECT_OPEN_BY_FID        0x20000000000000ULL /* open by fid won't pack
                                                       name in request */
+#define OBD_CONNECT_LFSCK      0x40000000000000ULL/* support online LFSCK */
 
 /* XXX README XXX:
  * Please DO NOT add flag values here before first ensuring that this same
@@ -1375,7 +1376,7 @@ extern void lustre_swab_ptlrpc_body(struct ptlrpc_body *pb);
                                OBD_CONNECT_LVB_TYPE | OBD_CONNECT_LAYOUTLOCK |\
                                OBD_CONNECT_PINGLESS | OBD_CONNECT_MAX_EASIZE |\
                                OBD_CONNECT_FLOCK_DEAD | \
-                               OBD_CONNECT_DISP_STRIPE)
+                               OBD_CONNECT_DISP_STRIPE | OBD_CONNECT_LFSCK)
 
 #define OST_CONNECT_SUPPORTED  (OBD_CONNECT_SRVLOCK | OBD_CONNECT_GRANT | \
                                 OBD_CONNECT_REQPORTAL | OBD_CONNECT_VERSION | \
@@ -1393,7 +1394,7 @@ extern void lustre_swab_ptlrpc_body(struct ptlrpc_body *pb);
                                OBD_CONNECT_JOBSTATS | \
                                OBD_CONNECT_LIGHTWEIGHT | OBD_CONNECT_LVB_TYPE|\
                                OBD_CONNECT_LAYOUTLOCK | OBD_CONNECT_FID | \
-                               OBD_CONNECT_PINGLESS)
+                               OBD_CONNECT_PINGLESS | OBD_CONNECT_LFSCK)
 #define ECHO_CONNECT_SUPPORTED (0)
 #define MGS_CONNECT_SUPPORTED  (OBD_CONNECT_VERSION | OBD_CONNECT_AT | \
                                OBD_CONNECT_FULL20 | OBD_CONNECT_IMP_RECOV | \

diff --git a/lustre/lfsck/lfsck_internal.h b/lustre/lfsck/lfsck_internal.h
index 5e2a144..6f1e4e1 100644 (file)
--- a/lustre/lfsck/lfsck_internal.h
+++ b/lustre/lfsck/lfsck_internal.h
@@ -252,6 +252,8 @@ struct lfsck_layout {
 };
 
 struct lfsck_component;
+struct lfsck_tgt_descs;
+struct lfsck_tgt_desc;
 
 struct lfsck_operations {
        int (*lfsck_reset)(const struct lu_env *env,
@@ -303,6 +305,12 @@ struct lfsck_operations {
 
        int (*lfsck_query)(const struct lu_env *env,
                           struct lfsck_component *com);
+
+       int (*lfsck_stop_notify)(const struct lu_env *env,
+                                struct lfsck_component *com,
+                                struct lfsck_tgt_descs *ltds,
+                                struct lfsck_tgt_desc *ltd,
+                                struct ptlrpc_request_set *set);
 };
 
 #define TGT_PTRS               256     /* number of pointers at 1st level */
@@ -314,8 +322,11 @@ struct lfsck_tgt_desc {
        struct dt_device  *ltd_key;
        struct obd_export *ltd_exp;
        struct list_head   ltd_layout_list;
+       struct list_head   ltd_layout_phase_list;
        atomic_t           ltd_ref;
        __u32              ltd_index;
+       __u32              ltd_layout_gen;
+       unsigned int       ltd_dead:1;
 };
 
 struct lfsck_tgt_desc_idx {
@@ -482,6 +493,13 @@ enum lfsck_linkea_flags {
        LLF_REPAIR_FAILED       = 0x02,
 };
 
+struct lfsck_async_interpret_args {
+       struct lfsck_component          *laia_com;
+       struct lfsck_tgt_descs          *laia_ltds;
+       struct lfsck_tgt_desc           *laia_ltd;
+       struct lfsck_request            *laia_lr;
+};
+
 struct lfsck_thread_args {
        struct lu_env            lta_env;
        struct lfsck_instance   *lta_lfsck;
@@ -507,6 +525,7 @@ struct lfsck_thread_info {
        struct lu_dirent        lti_ent;
        char                    lti_key[NAME_MAX + 16];
        struct lfsck_request    lti_lr;
+       struct lfsck_async_interpret_args lti_laia;
 };
 
 /* lfsck_lib.c */
@@ -541,6 +560,11 @@ int lfsck_post(const struct lu_env *env, struct lfsck_instance *lfsck,
               int result);
 int lfsck_double_scan(const struct lu_env *env, struct lfsck_instance *lfsck);
 void lfsck_quit(const struct lu_env *env, struct lfsck_instance *lfsck);
+int lfsck_async_request(const struct lu_env *env, struct obd_export *exp,
+                       struct lfsck_request *lr,
+                       struct ptlrpc_request_set *set,
+                       ptlrpc_interpterer_t interpterer,
+                       void *args, int request);
 
 /* lfsck_engine.c */
 int lfsck_master_engine(void *args);

diff --git a/lustre/lfsck/lfsck_layout.c b/lustre/lfsck/lfsck_layout.c
index 908ed89..a26a2fa 100644 (file)
--- a/lustre/lfsck/lfsck_layout.c
+++ b/lustre/lfsck/lfsck_layout.c
@@ -82,10 +82,21 @@ struct lfsck_layout_req {
 };
 
 struct lfsck_layout_master_data {
-       struct list_head        llmd_req_list;
        spinlock_t              llmd_lock;
+       struct list_head        llmd_req_list;
+
+       /* list for the ost targets involve layout verification. */
+       struct list_head        llmd_ost_list;
+
+       /* list for the ost targets in phase1 scanning. */
+       struct list_head        llmd_ost_phase1_list;
+
+       /* list for the ost targets in phase1 scanning. */
+       struct list_head        llmd_ost_phase2_list;
+
        struct ptlrpc_thread    llmd_thread;
        atomic_t                llmd_rpcs_in_flight;
+       __u32                   llmd_touch_gen;
        int                     llmd_prefetched;
        int                     llmd_assistant_status;
        int                     llmd_post_result;
@@ -616,29 +627,241 @@ out:
        return rc;
 }
 
+static int lfsck_layout_master_async_interpret(const struct lu_env *env,
+                                              struct ptlrpc_request *req,
+                                              void *args, int rc)
+{
+       struct lfsck_async_interpret_args *laia = args;
+       struct lfsck_component            *com  = laia->laia_com;
+       struct lfsck_layout_master_data   *llmd = com->lc_data;
+       struct lfsck_tgt_descs            *ltds = laia->laia_ltds;
+       struct lfsck_tgt_desc             *ltd  = laia->laia_ltd;
+       struct lfsck_request              *lr   = laia->laia_lr;
+
+       switch (lr->lr_event) {
+       case LE_START:
+               if (rc == 0) {
+                       LASSERT(!list_empty(&ltd->ltd_layout_list));
+
+                       spin_lock(&ltds->ltd_lock);
+                       if (!ltd->ltd_dead) {
+                               list_add_tail(&ltd->ltd_layout_list,
+                                             &llmd->llmd_ost_list);
+                               list_add_tail(&ltd->ltd_layout_phase_list,
+                                             &llmd->llmd_ost_phase1_list);
+                       }
+                       spin_unlock(&ltds->ltd_lock);
+               } else {
+                       struct lfsck_layout *lo = com->lc_file_ram;
+
+                       lo->ll_flags |= LF_INCOMPLETE;
+               }
+               lfsck_tgt_put(ltd);
+               break;
+       case LE_STOP:
+       case LE_PHASE2_DONE:
+               break;
+       case LE_QUERY:
+               spin_lock(&ltds->ltd_lock);
+               if (rc == 0 && !ltd->ltd_dead) {
+                       struct lfsck_reply *reply;
+
+                       reply = req_capsule_server_get(&req->rq_pill,
+                                                      &RMF_LFSCK_REPLY);
+                       switch (reply->lr_status) {
+                       case LS_SCANNING_PHASE1:
+                               break;
+                       case LS_SCANNING_PHASE2:
+                               list_del(&ltd->ltd_layout_phase_list);
+                               list_add_tail(&ltd->ltd_layout_phase_list,
+                                             &llmd->llmd_ost_phase2_list);
+                               break;
+                       default:
+                               list_del_init(&ltd->ltd_layout_phase_list);
+                               list_del_init(&ltd->ltd_layout_list);
+                               break;
+                       }
+               }
+               spin_unlock(&ltds->ltd_lock);
+               lfsck_tgt_put(ltd);
+               break;
+       default:
+               CERROR("%s: unexpected event: rc = %d\n",
+                      lfsck_lfsck2name(com->lc_lfsck), lr->lr_event);
+               break;
+       }
+
+       return 0;
+}
+
 static int lfsck_layout_master_query_others(const struct lu_env *env,
                                            struct lfsck_component *com)
 {
-       /* XXX: to be implemented. */
+       struct lfsck_thread_info          *info  = lfsck_env_info(env);
+       struct lfsck_request              *lr    = &info->lti_lr;
+       struct lfsck_async_interpret_args *laia  = &info->lti_laia;
+       struct lfsck_instance             *lfsck = com->lc_lfsck;
+       struct lfsck_layout_master_data   *llmd  = com->lc_data;
+       struct ptlrpc_request_set         *set;
+       struct lfsck_tgt_descs            *ltds;
+       struct lfsck_tgt_desc             *ltd;
+       __u32                              cnt   = 0;
+       int                                rc    = 0;
+       int                                rc1   = 0;
+       ENTRY;
 
-       return 0;
+       set = ptlrpc_prep_set();
+       if (set == NULL)
+               RETURN(-ENOMEM);
+
+       llmd->llmd_touch_gen++;
+       ltds = &lfsck->li_ost_descs;
+       memset(lr, 0, sizeof(*lr));
+       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_event = LE_QUERY;
+       lr->lr_active = LT_LAYOUT;
+
+       laia->laia_com = com;
+       laia->laia_ltds = ltds;
+       laia->laia_lr = lr;
+       spin_lock(&ltds->ltd_lock);
+       while (!list_empty(&llmd->llmd_ost_phase1_list)) {
+               ltd = list_entry(llmd->llmd_ost_phase1_list.next,
+                                struct lfsck_tgt_desc,
+                                ltd_layout_phase_list);
+               if (ltd->ltd_layout_gen == llmd->llmd_touch_gen)
+                       break;
+
+               ltd->ltd_layout_gen = llmd->llmd_touch_gen;
+               list_del(&ltd->ltd_layout_phase_list);
+               list_add_tail(&ltd->ltd_layout_phase_list,
+                             &llmd->llmd_ost_phase1_list);
+               atomic_inc(&ltd->ltd_ref);
+               laia->laia_ltd = ltd;
+               spin_unlock(&ltds->ltd_lock);
+               rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
+                                        lfsck_layout_master_async_interpret,
+                                        laia, LFSCK_QUERY);
+               if (rc != 0) {
+                       CERROR("%s: fail to query OST %x for layout: rc = %d\n",
+                              lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
+                       lfsck_tgt_put(ltd);
+                       rc1 = rc;
+               } else {
+                       cnt++;
+               }
+               spin_lock(&ltds->ltd_lock);
+       }
+       spin_unlock(&ltds->ltd_lock);
+
+       if (cnt > 0)
+               rc = ptlrpc_set_wait(set);
+       ptlrpc_set_destroy(set);
+
+       RETURN(rc1 != 0 ? rc1 : rc);
 }
 
 static inline bool
 lfsck_layout_master_to_orphan(struct lfsck_layout_master_data *llmd)
 {
-       /* XXX: to be implemented. */
-
-       return 1;
+       return !list_empty(&llmd->llmd_ost_phase2_list) ||
+              list_empty(&llmd->llmd_ost_phase1_list);
 }
 
 static int lfsck_layout_master_notify_others(const struct lu_env *env,
                                             struct lfsck_component *com,
                                             struct lfsck_request *lr)
 {
-       /* XXX: to be implemented. */
+       struct lfsck_thread_info          *info  = lfsck_env_info(env);
+       struct lfsck_async_interpret_args *laia  = &info->lti_laia;
+       struct lfsck_instance             *lfsck = com->lc_lfsck;
+       struct lfsck_layout_master_data   *llmd  = com->lc_data;
+       struct lfsck_layout               *lo    = com->lc_file_ram;
+       struct ptlrpc_request_set         *set;
+       struct lfsck_tgt_descs            *ltds;
+       struct lfsck_tgt_desc             *ltd;
+       __u32                              idx;
+       __u32                              cnt   = 0;
+       int                                rc    = 0;
+       ENTRY;
 
-       return 0;
+       set = ptlrpc_prep_set();
+       if (set == NULL)
+               RETURN(-ENOMEM);
+
+       lr->lr_active = LT_LAYOUT;
+       laia->laia_com = com;
+       laia->laia_lr = lr;
+       switch (lr->lr_event) {
+       case LE_START:
+               ltds = &lfsck->li_ost_descs;
+               laia->laia_ltds = ltds;
+               down_read(&ltds->ltd_rw_sem);
+               cfs_foreach_bit(ltds->ltd_tgts_bitmap, idx) {
+                       ltd = lfsck_tgt_get(ltds, idx);
+                       LASSERT(ltd != NULL);
+
+                       laia->laia_ltd = ltd;
+                       rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
+                                       lfsck_layout_master_async_interpret,
+                                       laia, LFSCK_NOTIFY);
+                       if (rc != 0) {
+                               CERROR("%s: fail to notify OST %x for layout "
+                                      "start: rc = %d\n",
+                                      lfsck_lfsck2name(lfsck), idx, rc);
+                               lfsck_tgt_put(ltd);
+                               lo->ll_flags |= LF_INCOMPLETE;
+                       } else {
+                               cnt++;
+                       }
+               }
+               up_read(&ltds->ltd_rw_sem);
+               break;
+       case LE_STOP:
+       case LE_PHASE2_DONE:
+               ltds = &lfsck->li_ost_descs;
+               laia->laia_ltds = ltds;
+               spin_lock(&ltds->ltd_lock);
+               while (!list_empty(&llmd->llmd_ost_list)) {
+                       ltd = list_entry(llmd->llmd_ost_list.next,
+                                        struct lfsck_tgt_desc,
+                                        ltd_layout_list);
+                       list_del_init(&ltd->ltd_layout_phase_list);
+                       list_del_init(&ltd->ltd_layout_list);
+                       laia->laia_ltd = ltd;
+                       spin_unlock(&ltds->ltd_lock);
+                       rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
+                                       lfsck_layout_master_async_interpret,
+                                       laia, LFSCK_NOTIFY);
+                       if (rc != 0)
+                               CERROR("%s: fail to notify OST %x for layout "
+                                      "stop/done: rc = %d\n",
+                                      lfsck_lfsck2name(lfsck),
+                                      ltd->ltd_index, rc);
+                       else
+                               cnt++;
+                       spin_lock(&ltds->ltd_lock);
+               }
+               spin_unlock(&ltds->ltd_lock);
+               break;
+       case LE_PHASE1_DONE:
+               break;
+       default:
+               CERROR("%s: unexpected LFSCK event: rc = %d\n",
+                      lfsck_lfsck2name(lfsck), lr->lr_event);
+               rc = -EINVAL;
+               break;
+       }
+
+       if (cnt > 0)
+               rc = ptlrpc_set_wait(set);
+       ptlrpc_set_destroy(set);
+
+       if (rc == 0 && lr->lr_event == LE_START &&
+           list_empty(&llmd->llmd_ost_list))
+               rc = -ENODEV;
+
+       RETURN(rc);
 }
 
 static int lfsck_layout_double_scan_result(const struct lu_env *env,
@@ -688,6 +911,15 @@ static int lfsck_layout_double_scan_result(const struct lu_env *env,
        return rc;
 }
 
+static int lfsck_layout_scan_orphan(const struct lu_env *env,
+                                   struct lfsck_component *com,
+                                   struct lfsck_tgt_desc *ltd)
+{
+       /* XXX: To be extended in other patch. */
+
+       return 0;
+}
+
 static int lfsck_layout_assistant(void *args)
 {
        struct lfsck_thread_args        *lta     = args;
@@ -714,7 +946,6 @@ static int lfsck_layout_assistant(void *args)
                       LSV_ASYNC_WINDOWS;
        lr->lr_speed = bk->lb_speed_limit;
        lr->lr_version = bk->lb_version;
-       lr->lr_active = LT_LAYOUT;
        lr->lr_param = bk->lb_param;
        lr->lr_async_windows = bk->lb_async_windows;
        if (pos->lp_oit_cookie <= 1)
@@ -791,7 +1022,6 @@ static int lfsck_layout_assistant(void *args)
 
                        memset(lr, 0, sizeof(*lr));
                        lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
-                       lr->lr_active = LT_LAYOUT;
                        lr->lr_event = LE_PHASE1_DONE;
                        lr->lr_status = llmd->llmd_post_result;
                        rc = lfsck_layout_master_notify_others(env, com, lr);
@@ -811,6 +1041,10 @@ static int lfsck_layout_assistant(void *args)
                        wake_up_all(&mthread->t_ctl_waitq);
 
                        while (llmd->llmd_in_double_scan) {
+                               struct lfsck_tgt_descs  *ltds =
+                                                       &lfsck->li_ost_descs;
+                               struct lfsck_tgt_desc   *ltd;
+
                                rc = lfsck_layout_master_query_others(env, com);
                                if (lfsck_layout_master_to_orphan(llmd))
                                        goto orphan;
@@ -840,9 +1074,35 @@ static int lfsck_layout_assistant(void *args)
                                        GOTO(cleanup2, rc);
 
 orphan:
-                               /* XXX: real double scan for ost orphans. */
-
-                               GOTO(cleanup2, rc = 1);
+                               spin_lock(&ltds->ltd_lock);
+                               while (!list_empty(
+                                               &llmd->llmd_ost_phase2_list)) {
+                                       ltd = list_entry(
+                                             llmd->llmd_ost_phase2_list.next,
+                                             struct lfsck_tgt_desc,
+                                             ltd_layout_phase_list);
+                                       list_del_init(
+                                               &ltd->ltd_layout_phase_list);
+                                       spin_unlock(&ltds->ltd_lock);
+
+                                       rc = lfsck_layout_scan_orphan(env, com,
+                                                                     ltd);
+                                       if (rc != 0 &&
+                                           bk->lb_param & LPF_FAILOUT)
+                                               GOTO(cleanup2, rc);
+
+                                       if (unlikely(llmd->llmd_exit ||
+                                               !thread_is_running(mthread)))
+                                               GOTO(cleanup2, rc = 0);
+
+                                       spin_lock(&ltds->ltd_lock);
+                               }
+
+                               if (list_empty(&llmd->llmd_ost_phase1_list)) {
+                                       spin_unlock(&ltds->ltd_lock);
+                                       GOTO(cleanup2, rc = 1);
+                               }
+                               spin_unlock(&ltds->ltd_lock);
                        }
                }
        }
@@ -872,7 +1132,6 @@ cleanup1:
 cleanup2:
        memset(lr, 0, sizeof(*lr));
        lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
-       lr->lr_active = LT_LAYOUT;
        if (rc > 0) {
                lr->lr_event = LE_PHASE2_DONE;
                lr->lr_status = rc;
@@ -1635,7 +1894,11 @@ static int lfsck_layout_slave_double_scan(const struct lu_env *env,
 static void lfsck_layout_master_data_release(const struct lu_env *env,
                                             struct lfsck_component *com)
 {
-       struct lfsck_layout_master_data *llmd = com->lc_data;
+       struct lfsck_layout_master_data *llmd   = com->lc_data;
+       struct lfsck_instance           *lfsck  = com->lc_lfsck;
+       struct lfsck_tgt_descs          *ltds;
+       struct lfsck_tgt_desc           *ltd;
+       struct lfsck_tgt_desc           *next;
 
        LASSERT(llmd != NULL);
        LASSERT(thread_is_init(&llmd->llmd_thread) ||
@@ -1644,6 +1907,23 @@ static void lfsck_layout_master_data_release(const struct lu_env *env,
        LASSERT(atomic_read(&llmd->llmd_rpcs_in_flight) == 0);
 
        com->lc_data = NULL;
+
+       ltds = &lfsck->li_ost_descs;
+       spin_lock(&ltds->ltd_lock);
+       list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_phase1_list,
+                                ltd_layout_phase_list) {
+               list_del_init(&ltd->ltd_layout_phase_list);
+       }
+       list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_phase2_list,
+                                ltd_layout_phase_list) {
+               list_del_init(&ltd->ltd_layout_phase_list);
+       }
+       list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_list,
+                                ltd_layout_list) {
+               list_del_init(&ltd->ltd_layout_list);
+       }
+       spin_unlock(&ltds->ltd_lock);
+
        OBD_FREE_PTR(llmd);
 }
 
@@ -1708,6 +1988,42 @@ static int lfsck_layout_query(const struct lu_env *env,
        return lo->ll_status;
 }
 
+static int lfsck_layout_master_stop_notify(const struct lu_env *env,
+                                          struct lfsck_component *com,
+                                          struct lfsck_tgt_descs *ltds,
+                                          struct lfsck_tgt_desc *ltd,
+                                          struct ptlrpc_request_set *set)
+{
+       struct lfsck_thread_info          *info  = lfsck_env_info(env);
+       struct lfsck_async_interpret_args *laia  = &info->lti_laia;
+       struct lfsck_request              *lr    = &info->lti_lr;
+       struct lfsck_instance             *lfsck = com->lc_lfsck;
+       int                                rc;
+
+       LASSERT(list_empty(&ltd->ltd_layout_list));
+       LASSERT(list_empty(&ltd->ltd_layout_phase_list));
+
+       memset(lr, 0, sizeof(*lr));
+       lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
+       lr->lr_event = LE_STOP;
+       lr->lr_active = LT_LAYOUT;
+       lr->lr_status = LS_CO_STOPPED;
+
+       laia->laia_com = com;
+       laia->laia_ltds = ltds;
+       laia->laia_ltd = ltd;
+       laia->laia_lr = lr;
+
+       rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
+                                lfsck_layout_master_async_interpret,
+                                laia, LFSCK_NOTIFY);
+       if (rc != 0)
+               CERROR("%s: Fail to notify OST %x for stop: rc = %d\n",
+                      lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
+
+       return rc;
+}
+
 static struct lfsck_operations lfsck_layout_master_ops = {
        .lfsck_reset            = lfsck_layout_reset,
        .lfsck_fail             = lfsck_layout_fail,
@@ -1722,6 +2038,7 @@ static struct lfsck_operations lfsck_layout_master_ops = {
        .lfsck_quit             = lfsck_layout_master_quit,
        .lfsck_in_notify        = lfsck_layout_master_in_notify,
        .lfsck_query            = lfsck_layout_query,
+       .lfsck_stop_notify      = lfsck_layout_master_stop_notify,
 };
 
 static struct lfsck_operations lfsck_layout_slave_ops = {
@@ -1767,6 +2084,9 @@ int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
                        GOTO(out, rc = -ENOMEM);
 
                INIT_LIST_HEAD(&llmd->llmd_req_list);
+               INIT_LIST_HEAD(&llmd->llmd_ost_list);
+               INIT_LIST_HEAD(&llmd->llmd_ost_phase1_list);
+               INIT_LIST_HEAD(&llmd->llmd_ost_phase2_list);
                spin_lock_init(&llmd->llmd_lock);
                init_waitqueue_head(&llmd->llmd_thread.t_ctl_waitq);
                atomic_set(&llmd->llmd_rpcs_in_flight, 0);

diff --git a/lustre/lfsck/lfsck_lib.c b/lustre/lfsck/lfsck_lib.c
index d59a07a..dc4616e 100644 (file)
--- a/lustre/lfsck/lfsck_lib.c
+++ b/lustre/lfsck/lfsck_lib.c
@@ -137,6 +137,7 @@ static void lfsck_tgt_descs_fini(struct lfsck_tgt_descs *ltds)
                ltd = LTD_TGT(ltds, idx);
                if (likely(ltd != NULL)) {
                        LASSERT(list_empty(&ltd->ltd_layout_list));
+                       LASSERT(list_empty(&ltd->ltd_layout_phase_list));
 
                        ltds->ltd_tgtnr--;
                        cfs_bitmap_clear(ltds->ltd_tgts_bitmap, idx);
@@ -981,7 +982,49 @@ int lfsck_double_scan(const struct lu_env *env, struct lfsck_instance *lfsck)
                     atomic_read(&lfsck->li_double_scan_count) == 0,
                     &lwi);
 
-       return (rc1 != 0 ? rc1 : rc);
+       return rc1 != 0 ? rc1 : rc;
+}
+
+int lfsck_stop_notify(const struct lu_env *env, struct lfsck_instance *lfsck,
+                     struct lfsck_tgt_descs *ltds, struct lfsck_tgt_desc *ltd)
+{
+       struct ptlrpc_request_set *set;
+       struct lfsck_component    *com;
+       int                        cnt = 0;
+       int                        rc  = 0;
+       int                        rc1 = 0;
+
+       set = ptlrpc_prep_set();
+       if (set == NULL)
+               return -ENOMEM;
+
+       list_for_each_entry(com, &lfsck->li_list_scan, lc_link) {
+               if (com->lc_ops->lfsck_stop_notify != NULL) {
+                       rc = com->lc_ops->lfsck_stop_notify(env, com, ltds,
+                                                           ltd, set);
+                       if (rc != 0)
+                               rc1 = rc;
+                       else
+                               cnt++;
+               }
+       }
+
+       list_for_each_entry(com, &lfsck->li_list_double_scan, lc_link) {
+               if (com->lc_ops->lfsck_stop_notify != NULL) {
+                       rc = com->lc_ops->lfsck_stop_notify(env, com, ltds,
+                                                           ltd, set);
+                       if (rc != 0)
+                               rc1 = rc;
+                       else
+                               cnt++;
+               }
+       }
+
+       if (cnt > 0)
+               rc = ptlrpc_set_wait(set);
+       ptlrpc_set_destroy(set);
+
+       return rc1 != 0 ? rc1 : rc;
 }
 
 void lfsck_quit(const struct lu_env *env, struct lfsck_instance *lfsck)
@@ -1002,6 +1045,57 @@ void lfsck_quit(const struct lu_env *env, struct lfsck_instance *lfsck)
        }
 }
 
+int lfsck_async_request(const struct lu_env *env, struct obd_export *exp,
+                       struct lfsck_request *lr,
+                       struct ptlrpc_request_set *set,
+                       ptlrpc_interpterer_t interpreter,
+                       void *args, int request)
+{
+       struct lfsck_async_interpret_args *laia;
+       struct ptlrpc_request             *req;
+       struct lfsck_request              *tmp;
+       struct req_format                 *format;
+       int                                rc;
+
+       if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
+               return -EOPNOTSUPP;
+
+       switch (request) {
+       case LFSCK_NOTIFY:
+               format = &RQF_LFSCK_NOTIFY;
+               break;
+       case LFSCK_QUERY:
+               format = &RQF_LFSCK_QUERY;
+               break;
+       default:
+               CERROR("%s: unknown async request: opc = %d\n",
+                      exp->exp_obd->obd_name, request);
+               return -EINVAL;
+       }
+
+       req = ptlrpc_request_alloc(class_exp2cliimp(exp), format);
+       if (req == NULL)
+               return -ENOMEM;
+
+       rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, request);
+       if (rc != 0) {
+               ptlrpc_request_free(req);
+
+               return rc;
+       }
+
+       tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
+       *tmp = *lr;
+       ptlrpc_request_set_replen(req);
+
+       laia = ptlrpc_req_async_args(req);
+       *laia = *(struct lfsck_async_interpret_args *)args;
+       req->rq_interpret_reply = interpreter;
+       ptlrpc_set_add_req(set, req);
+
+       return 0;
+}
+
 /* external interfaces */
 
 int lfsck_get_speed(struct dt_device *key, void *buf, int len)
@@ -1617,6 +1711,7 @@ int lfsck_add_target(const struct lu_env *env, struct dt_device *key,
        ltd->ltd_exp = exp;
        INIT_LIST_HEAD(&ltd->ltd_orphan_list);
        INIT_LIST_HEAD(&ltd->ltd_layout_list);
+       INIT_LIST_HEAD(&ltd->ltd_layout_phase_list);
        atomic_set(&ltd->ltd_ref, 1);
        ltd->ltd_index = index;
 
@@ -1653,6 +1748,7 @@ void lfsck_del_target(const struct lu_env *env, struct dt_device *key,
        struct lfsck_tgt_desc   *ltd;
        struct list_head        *head;
        bool                     found = false;
+       bool                     stop  = false;
 
        if (for_ost)
                head = &lfsck_ost_orphan_list;
@@ -1692,11 +1788,18 @@ void lfsck_del_target(const struct lu_env *env, struct dt_device *key,
                goto unlock;
 
        found = true;
+       spin_lock(&ltds->ltd_lock);
+       ltd->ltd_dead = 1;
        if (!list_empty(&ltd->ltd_layout_list)) {
-               spin_lock(&ltds->ltd_lock);
                list_del_init(&ltd->ltd_layout_list);
-               spin_unlock(&ltds->ltd_lock);
+               stop = true;
+       } else {
+               LASSERT(list_empty(&ltd->ltd_layout_phase_list));
        }
+       spin_unlock(&ltds->ltd_lock);
+
+       if (stop && lfsck->li_master)
+               lfsck_stop_notify(env, lfsck, ltds, ltd);
 
        LASSERT(ltds->ltd_tgtnr > 0);
 

diff --git a/lustre/lod/lod_lov.c b/lustre/lod/lod_lov.c
index 2c93223..693d94b 100644 (file)
--- a/lustre/lod/lod_lov.c
+++ b/lustre/lod/lod_lov.c
@@ -228,7 +228,8 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
                                           OBD_CONNECT_FID |
                                           OBD_CONNECT_LVB_TYPE |
                                           OBD_CONNECT_VERSION |
-                                          OBD_CONNECT_PINGLESS;
+                                          OBD_CONNECT_PINGLESS |
+                                          OBD_CONNECT_LFSCK;
 
                data->ocd_group = tgt_index;
                ltd = &lod->lod_ost_descs;
@@ -244,7 +245,8 @@ int lod_add_device(const struct lu_env *env, struct lod_device *lod,
                                           OBD_CONNECT_MDS_MDS |
                                           OBD_CONNECT_FID |
                                           OBD_CONNECT_AT |
-                                          OBD_CONNECT_FULL20;
+                                          OBD_CONNECT_FULL20 |
+                                          OBD_CONNECT_LFSCK;
                /* XXX set MDS-MDS flags, remove this when running this
                 * on client*/
                data->ocd_connect_flags |= OBD_CONNECT_MDS_MDS;

diff --git a/lustre/ptlrpc/wiretest.c b/lustre/ptlrpc/wiretest.c
index 94eebc2..5e074d9 100644 (file)
--- a/lustre/ptlrpc/wiretest.c
+++ b/lustre/ptlrpc/wiretest.c
@@ -1171,6 +1171,8 @@ void lustre_assert_wire_constants(void)
                 OBD_CONNECT_FLOCK_DEAD);
        LASSERTF(OBD_CONNECT_OPEN_BY_FID == 0x20000000000000ULL, "found 0x%.16llxULL\n",
                 OBD_CONNECT_OPEN_BY_FID);
+       LASSERTF(OBD_CONNECT_LFSCK == 0x40000000000000ULL, "found 0x%.16llxULL\n",
+                OBD_CONNECT_LFSCK);
        LASSERTF(OBD_CKSUM_CRC32 == 0x00000001UL, "found 0x%.8xUL\n",
                (unsigned)OBD_CKSUM_CRC32);
        LASSERTF(OBD_CKSUM_ADLER == 0x00000002UL, "found 0x%.8xUL\n",

diff --git a/lustre/utils/wirecheck.c b/lustre/utils/wirecheck.c
index ea07f4e..5a999a1 100644 (file)
--- a/lustre/utils/wirecheck.c
+++ b/lustre/utils/wirecheck.c
@@ -524,6 +524,7 @@ check_obd_connect_data(void)
        CHECK_DEFINE_64X(OBD_CONNECT_PINGLESS);
        CHECK_DEFINE_64X(OBD_CONNECT_FLOCK_DEAD);
        CHECK_DEFINE_64X(OBD_CONNECT_OPEN_BY_FID);
+       CHECK_DEFINE_64X(OBD_CONNECT_LFSCK);
 
        CHECK_VALUE_X(OBD_CKSUM_CRC32);
        CHECK_VALUE_X(OBD_CKSUM_ADLER);

diff --git a/lustre/utils/wiretest.c b/lustre/utils/wiretest.c
index 0b54043..347e987 100644 (file)
--- a/lustre/utils/wiretest.c
+++ b/lustre/utils/wiretest.c
@@ -1180,6 +1180,8 @@ void lustre_assert_wire_constants(void)
                 OBD_CONNECT_FLOCK_DEAD);
        LASSERTF(OBD_CONNECT_OPEN_BY_FID == 0x20000000000000ULL, "found 0x%.16llxULL\n",
                 OBD_CONNECT_OPEN_BY_FID);
+       LASSERTF(OBD_CONNECT_LFSCK == 0x40000000000000ULL, "found 0x%.16llxULL\n",
+                OBD_CONNECT_LFSCK);
        LASSERTF(OBD_CKSUM_CRC32 == 0x00000001UL, "found 0x%.8xUL\n",
                (unsigned)OBD_CKSUM_CRC32);
        LASSERTF(OBD_CKSUM_ADLER == 0x00000002UL, "found 0x%.8xUL\n",