lustre/mdd/mdd_lfsck.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License version 2 for more details.  A copy is
  14  * included in the COPYING file that accompanied this code.
  15
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2012, Intel Corporation.
  24  */
  25 /*
  26  * lustre/mdd/mdd_lfsck.c
  27  *
  28  * Top-level entry points into mdd module
  29  *
  30  * LFSCK controller, which scans the whole device through low layer
  31  * iteration APIs, drives all lfsck compeonents, controls the speed.
  32  *
  33  * Author: Fan Yong <yong.fan@whamcloud.com>
  34  */
  35
  36 #ifndef EXPORT_SYMTAB
  37 # define EXPORT_SYMTAB
  38 #endif
  39 #define DEBUG_SUBSYSTEM S_MDS
  40
  41 #include <lustre/lustre_idl.h>
  42 #include <lustre_fid.h>
  43 #include <obd_support.h>
  44
  45 #include "mdd_internal.h"
  46 #include "mdd_lfsck.h"
  47
  48 #define HALF_SEC                        (CFS_HZ >> 1)
  49 #define LFSCK_CHECKPOINT_INTERVAL       60
  50 #define MDS_DIR_DUMMY_START             0xffffffffffffffffULL
  51
  52 const char lfsck_bookmark_name[] = "lfsck_bookmark";
  53 const char lfsck_namespace_name[] = "lfsck_namespace";
  54
  55 /* misc functions */
  56
  57 static inline struct mdd_device *mdd_lfsck2mdd(struct md_lfsck *lfsck)
  58 {
  59         return container_of0(lfsck, struct mdd_device, mdd_lfsck);
  60 }
  61
  62 static inline char *mdd_lfsck2name(struct md_lfsck *lfsck)
  63 {
  64         struct mdd_device *mdd = mdd_lfsck2mdd(lfsck);
  65
  66         return mdd2obd_dev(mdd)->obd_name;
  67 }
  68
  69 static inline void mdd_lfsck_component_get(struct lfsck_component *com)
  70 {
  71         atomic_inc(&com->lc_ref);
  72 }
  73
  74 static inline void mdd_lfsck_component_put(const struct lu_env *env,
  75                                            struct lfsck_component *com)
  76 {
  77         if (atomic_dec_and_test(&com->lc_ref)) {
  78                 if (com->lc_obj != NULL)
  79                         lu_object_put(env, &com->lc_obj->do_lu);
  80                 if (com->lc_file_ram != NULL)
  81                         OBD_FREE(com->lc_file_ram, com->lc_file_size);
  82                 if (com->lc_file_disk != NULL)
  83                         OBD_FREE(com->lc_file_disk, com->lc_file_size);
  84                 OBD_FREE_PTR(com);
  85         }
  86 }
  87
  88 static inline struct lfsck_component *
  89 __mdd_lfsck_component_find(struct md_lfsck *lfsck, __u16 type, cfs_list_t *list)
  90 {
  91         struct lfsck_component *com;
  92
  93         cfs_list_for_each_entry(com, list, lc_link) {
  94                 if (com->lc_type == type)
  95                         return com;
  96         }
  97         return NULL;
  98 }
  99
 100 static void mdd_lfsck_component_cleanup(const struct lu_env *env,
 101                                         struct lfsck_component *com)
 102 {
 103         if (!cfs_list_empty(&com->lc_link))
 104                 cfs_list_del_init(&com->lc_link);
 105         if (!cfs_list_empty(&com->lc_link_dir))
 106                 cfs_list_del_init(&com->lc_link_dir);
 107
 108         mdd_lfsck_component_put(env, com);
 109 }
 110
 111 static void mdd_lfsck_pos_fill(const struct lu_env *env, struct md_lfsck *lfsck,
 112                                struct lfsck_position *pos, bool oit_processed,
 113                                bool dir_processed)
 114 {
 115         const struct dt_it_ops *iops = &lfsck->ml_obj_oit->do_index_ops->dio_it;
 116
 117         spin_lock(&lfsck->ml_lock);
 118         if (unlikely(lfsck->ml_di_oit == NULL)) {
 119                 spin_unlock(&lfsck->ml_lock);
 120                 memset(pos, 0, sizeof(*pos));
 121                 return;
 122         }
 123
 124         pos->lp_oit_cookie = iops->store(env, lfsck->ml_di_oit);
 125
 126         LASSERT(pos->lp_oit_cookie > 0);
 127
 128         if (!oit_processed)
 129                 pos->lp_oit_cookie--;
 130
 131         if (lfsck->ml_di_dir != NULL) {
 132                 struct dt_object *dto = lfsck->ml_obj_dir;
 133
 134                 pos->lp_dir_parent = *lu_object_fid(&dto->do_lu);
 135                 pos->lp_dir_cookie = dto->do_index_ops->dio_it.store(env,
 136                                                         lfsck->ml_di_dir);
 137
 138                 LASSERT(pos->lp_dir_cookie != MDS_DIR_DUMMY_START);
 139
 140                 if (pos->lp_dir_cookie == MDS_DIR_END_OFF)
 141                         LASSERT(dir_processed);
 142
 143                 /* For the dir which just to be processed,
 144                  * lp_dir_cookie will become MDS_DIR_DUMMY_START,
 145                  * which can be correctly handled by mdd_lfsck_prep. */
 146                 if (!dir_processed)
 147                         pos->lp_dir_cookie--;
 148         } else {
 149                 fid_zero(&pos->lp_dir_parent);
 150                 pos->lp_dir_cookie = 0;
 151         }
 152         spin_unlock(&lfsck->ml_lock);
 153 }
 154
 155 static inline void mdd_lfsck_pos_set_zero(struct lfsck_position *pos)
 156 {
 157         memset(pos, 0, sizeof(*pos));
 158 }
 159
 160 static inline int mdd_lfsck_pos_is_zero(const struct lfsck_position *pos)
 161 {
 162         return pos->lp_oit_cookie == 0 && fid_is_zero(&pos->lp_dir_parent);
 163 }
 164
 165 static inline int mdd_lfsck_pos_is_eq(const struct lfsck_position *pos1,
 166                                       const struct lfsck_position *pos2)
 167 {
 168         if (pos1->lp_oit_cookie < pos2->lp_oit_cookie)
 169                 return -1;
 170
 171         if (pos1->lp_oit_cookie > pos2->lp_oit_cookie)
 172                 return 1;
 173
 174         if (fid_is_zero(&pos1->lp_dir_parent) &&
 175             !fid_is_zero(&pos2->lp_dir_parent))
 176                 return -1;
 177
 178         if (!fid_is_zero(&pos1->lp_dir_parent) &&
 179             fid_is_zero(&pos2->lp_dir_parent))
 180                 return 1;
 181
 182         if (fid_is_zero(&pos1->lp_dir_parent) &&
 183             fid_is_zero(&pos2->lp_dir_parent))
 184                 return 0;
 185
 186         LASSERT(lu_fid_eq(&pos1->lp_dir_parent, &pos2->lp_dir_parent));
 187
 188         if (pos1->lp_dir_cookie < pos2->lp_dir_cookie)
 189                 return -1;
 190
 191         if (pos1->lp_dir_cookie > pos2->lp_dir_cookie)
 192                 return 1;
 193
 194         return 0;
 195 }
 196
 197 static void mdd_lfsck_close_dir(const struct lu_env *env,
 198                                 struct md_lfsck *lfsck)
 199 {
 200         struct dt_object        *dir_obj  = lfsck->ml_obj_dir;
 201         const struct dt_it_ops  *dir_iops = &dir_obj->do_index_ops->dio_it;
 202         struct dt_it            *dir_di   = lfsck->ml_di_dir;
 203
 204         spin_lock(&lfsck->ml_lock);
 205         lfsck->ml_di_dir = NULL;
 206         spin_unlock(&lfsck->ml_lock);
 207
 208         dir_iops->put(env, dir_di);
 209         dir_iops->fini(env, dir_di);
 210         lfsck->ml_obj_dir = NULL;
 211         lu_object_put(env, &dir_obj->do_lu);
 212 }
 213
 214 static void __mdd_lfsck_set_speed(struct md_lfsck *lfsck, __u32 limit)
 215 {
 216         lfsck->ml_bookmark_ram.lb_speed_limit = limit;
 217         if (limit != LFSCK_SPEED_NO_LIMIT) {
 218                 if (limit > CFS_HZ) {
 219                         lfsck->ml_sleep_rate = limit / CFS_HZ;
 220                         lfsck->ml_sleep_jif = 1;
 221                 } else {
 222                         lfsck->ml_sleep_rate = 1;
 223                         lfsck->ml_sleep_jif = CFS_HZ / limit;
 224                 }
 225         } else {
 226                 lfsck->ml_sleep_jif = 0;
 227                 lfsck->ml_sleep_rate = 0;
 228         }
 229 }
 230
 231 static void mdd_lfsck_control_speed(struct md_lfsck *lfsck)
 232 {
 233         struct ptlrpc_thread *thread = &lfsck->ml_thread;
 234         struct l_wait_info    lwi;
 235
 236         if (lfsck->ml_sleep_jif > 0 &&
 237             lfsck->ml_new_scanned >= lfsck->ml_sleep_rate) {
 238                 spin_lock(&lfsck->ml_lock);
 239                 if (likely(lfsck->ml_sleep_jif > 0 &&
 240                            lfsck->ml_new_scanned >= lfsck->ml_sleep_rate)) {
 241                         lwi = LWI_TIMEOUT_INTR(lfsck->ml_sleep_jif, NULL,
 242                                                LWI_ON_SIGNAL_NOOP, NULL);
 243                         spin_unlock(&lfsck->ml_lock);
 244
 245                         l_wait_event(thread->t_ctl_waitq,
 246                                      !thread_is_running(thread),
 247                                      &lwi);
 248                         lfsck->ml_new_scanned = 0;
 249                 } else {
 250                         spin_unlock(&lfsck->ml_lock);
 251                 }
 252         }
 253 }
 254
 255 /* lfsck_bookmark file ops */
 256
 257 static void inline mdd_lfsck_bookmark_to_cpu(struct lfsck_bookmark *des,
 258                                              struct lfsck_bookmark *src)
 259 {
 260         des->lb_magic = le32_to_cpu(src->lb_magic);
 261         des->lb_version = le16_to_cpu(src->lb_version);
 262         des->lb_param = le16_to_cpu(src->lb_param);
 263         des->lb_speed_limit = le32_to_cpu(src->lb_speed_limit);
 264 }
 265
 266 static void inline mdd_lfsck_bookmark_to_le(struct lfsck_bookmark *des,
 267                                             struct lfsck_bookmark *src)
 268 {
 269         des->lb_magic = cpu_to_le32(src->lb_magic);
 270         des->lb_version = cpu_to_le16(src->lb_version);
 271         des->lb_param = cpu_to_le16(src->lb_param);
 272         des->lb_speed_limit = cpu_to_le32(src->lb_speed_limit);
 273 }
 274
 275 static int mdd_lfsck_bookmark_load(const struct lu_env *env,
 276                                    struct md_lfsck *lfsck)
 277 {
 278         loff_t pos = 0;
 279         int    len = sizeof(struct lfsck_bookmark);
 280         int    rc;
 281
 282         rc = dt_record_read(env, lfsck->ml_bookmark_obj,
 283                             mdd_buf_get(env, &lfsck->ml_bookmark_disk, len),
 284                             &pos);
 285         if (rc == 0) {
 286                 struct lfsck_bookmark *bm = &lfsck->ml_bookmark_ram;
 287
 288                 mdd_lfsck_bookmark_to_cpu(bm, &lfsck->ml_bookmark_disk);
 289                 if (bm->lb_magic != LFSCK_BOOKMARK_MAGIC) {
 290                         CWARN("%.16s: invalid lfsck_bookmark magic "
 291                               "0x%x != 0x%x\n", mdd_lfsck2name(lfsck),
 292                               bm->lb_magic, LFSCK_BOOKMARK_MAGIC);
 293                         /* Process it as new lfsck_bookmark. */
 294                         rc = -ENODATA;
 295                 }
 296         } else {
 297                 if (rc == -EFAULT && pos == 0)
 298                         /* return -ENODATA for empty lfsck_bookmark. */
 299                         rc = -ENODATA;
 300                 else
 301                         CERROR("%.16s: fail to load lfsck_bookmark, "
 302                                "expected = %d, rc = %d\n",
 303                                mdd_lfsck2name(lfsck), len, rc);
 304         }
 305         return rc;
 306 }
 307
 308 static int mdd_lfsck_bookmark_store(const struct lu_env *env,
 309                                     struct md_lfsck *lfsck)
 310 {
 311         struct mdd_device *mdd    = mdd_lfsck2mdd(lfsck);
 312         struct thandle    *handle;
 313         struct dt_object  *obj    = lfsck->ml_bookmark_obj;
 314         loff_t             pos    = 0;
 315         int                len    = sizeof(struct lfsck_bookmark);
 316         int                rc;
 317         ENTRY;
 318
 319         mdd_lfsck_bookmark_to_le(&lfsck->ml_bookmark_disk,
 320                                  &lfsck->ml_bookmark_ram);
 321         handle = dt_trans_create(env, mdd->mdd_bottom);
 322         if (IS_ERR(handle)) {
 323                 rc = PTR_ERR(handle);
 324                 CERROR("%.16s: fail to create trans for storing "
 325                        "lfsck_bookmark: %d\n,", mdd_lfsck2name(lfsck), rc);
 326                 RETURN(rc);
 327         }
 328
 329         rc = dt_declare_record_write(env, obj, len, 0, handle);
 330         if (rc != 0) {
 331                 CERROR("%.16s: fail to declare trans for storing "
 332                        "lfsck_bookmark: %d\n,", mdd_lfsck2name(lfsck), rc);
 333                 GOTO(out, rc);
 334         }
 335
 336         rc = dt_trans_start_local(env, mdd->mdd_bottom, handle);
 337         if (rc != 0) {
 338                 CERROR("%.16s: fail to start trans for storing "
 339                        "lfsck_bookmark: %d\n,", mdd_lfsck2name(lfsck), rc);
 340                 GOTO(out, rc);
 341         }
 342
 343         rc = dt_record_write(env, obj,
 344                              mdd_buf_get(env, &lfsck->ml_bookmark_disk, len),
 345                              &pos, handle);
 346         if (rc != 0)
 347                 CERROR("%.16s: fail to store lfsck_bookmark, expected = %d, "
 348                        "rc = %d\n", mdd_lfsck2name(lfsck), len, rc);
 349
 350         GOTO(out, rc);
 351
 352 out:
 353         dt_trans_stop(env, mdd->mdd_bottom, handle);
 354         return rc;
 355 }
 356
 357 static int mdd_lfsck_bookmark_init(const struct lu_env *env,
 358                                    struct md_lfsck *lfsck)
 359 {
 360         struct lfsck_bookmark *mb = &lfsck->ml_bookmark_ram;
 361         int rc;
 362
 363         memset(mb, 0, sizeof(mb));
 364         mb->lb_magic = LFSCK_BOOKMARK_MAGIC;
 365         mb->lb_version = LFSCK_VERSION_V1;
 366         mutex_lock(&lfsck->ml_mutex);
 367         rc = mdd_lfsck_bookmark_store(env, lfsck);
 368         mutex_unlock(&lfsck->ml_mutex);
 369         return rc;
 370 }
 371
 372 /* lfsck_namespace file ops */
 373
 374 static void inline mdd_lfsck_position_to_cpu(struct lfsck_position *des,
 375                                              struct lfsck_position *src)
 376 {
 377         des->lp_oit_cookie = le64_to_cpu(src->lp_oit_cookie);
 378         fid_le_to_cpu(&des->lp_dir_parent, &src->lp_dir_parent);
 379         des->lp_dir_cookie = le64_to_cpu(src->lp_dir_cookie);
 380 }
 381
 382 static void inline mdd_lfsck_position_to_le(struct lfsck_position *des,
 383                                              struct lfsck_position *src)
 384 {
 385         des->lp_oit_cookie = cpu_to_le64(src->lp_oit_cookie);
 386         fid_cpu_to_le(&des->lp_dir_parent, &src->lp_dir_parent);
 387         des->lp_dir_cookie = cpu_to_le64(src->lp_dir_cookie);
 388 }
 389
 390 static void inline mdd_lfsck_namespace_to_cpu(struct lfsck_namespace *des,
 391                                               struct lfsck_namespace *src)
 392 {
 393         des->ln_magic = le32_to_cpu(src->ln_magic);
 394         des->ln_status = le32_to_cpu(src->ln_status);
 395         des->ln_flags = le32_to_cpu(src->ln_flags);
 396         des->ln_success_count = le32_to_cpu(src->ln_success_count);
 397         des->ln_run_time_phase1 = le32_to_cpu(src->ln_run_time_phase1);
 398         des->ln_run_time_phase2 = le32_to_cpu(src->ln_run_time_phase2);
 399         des->ln_time_last_complete = le64_to_cpu(src->ln_time_last_complete);
 400         des->ln_time_latest_start = le64_to_cpu(src->ln_time_latest_start);
 401         des->ln_time_last_checkpoint =
 402                                 le64_to_cpu(src->ln_time_last_checkpoint);
 403         mdd_lfsck_position_to_cpu(&des->ln_pos_latest_start,
 404                                   &src->ln_pos_latest_start);
 405         mdd_lfsck_position_to_cpu(&des->ln_pos_last_checkpoint,
 406                                   &src->ln_pos_last_checkpoint);
 407         mdd_lfsck_position_to_cpu(&des->ln_pos_first_inconsistent,
 408                                   &src->ln_pos_first_inconsistent);
 409         des->ln_items_checked = le64_to_cpu(src->ln_items_checked);
 410         des->ln_items_repaired = le64_to_cpu(src->ln_items_repaired);
 411         des->ln_items_failed = le64_to_cpu(src->ln_items_failed);
 412         des->ln_dirs_checked = le64_to_cpu(src->ln_dirs_checked);
 413         des->ln_mlinked_checked = le64_to_cpu(src->ln_mlinked_checked);
 414         des->ln_objs_checked_phase2 = le64_to_cpu(src->ln_objs_checked_phase2);
 415         des->ln_objs_repaired_phase2 =
 416                                 le64_to_cpu(src->ln_objs_repaired_phase2);
 417         des->ln_objs_failed_phase2 = le64_to_cpu(src->ln_objs_failed_phase2);
 418         des->ln_objs_nlink_repaired = le64_to_cpu(src->ln_objs_nlink_repaired);
 419         des->ln_objs_lost_found = le64_to_cpu(src->ln_objs_lost_found);
 420         fid_le_to_cpu(&des->ln_fid_latest_scanned_phase2,
 421                       &src->ln_fid_latest_scanned_phase2);
 422 }
 423
 424 static void inline mdd_lfsck_namespace_to_le(struct lfsck_namespace *des,
 425                                              struct lfsck_namespace *src)
 426 {
 427         des->ln_magic = cpu_to_le32(src->ln_magic);
 428         des->ln_status = cpu_to_le32(src->ln_status);
 429         des->ln_flags = cpu_to_le32(src->ln_flags);
 430         des->ln_success_count = cpu_to_le32(src->ln_success_count);
 431         des->ln_run_time_phase1 = cpu_to_le32(src->ln_run_time_phase1);
 432         des->ln_run_time_phase2 = cpu_to_le32(src->ln_run_time_phase2);
 433         des->ln_time_last_complete = cpu_to_le64(src->ln_time_last_complete);
 434         des->ln_time_latest_start = cpu_to_le64(src->ln_time_latest_start);
 435         des->ln_time_last_checkpoint =
 436                                 cpu_to_le64(src->ln_time_last_checkpoint);
 437         mdd_lfsck_position_to_le(&des->ln_pos_latest_start,
 438                                  &src->ln_pos_latest_start);
 439         mdd_lfsck_position_to_le(&des->ln_pos_last_checkpoint,
 440                                  &src->ln_pos_last_checkpoint);
 441         mdd_lfsck_position_to_le(&des->ln_pos_first_inconsistent,
 442                                  &src->ln_pos_first_inconsistent);
 443         des->ln_items_checked = cpu_to_le64(src->ln_items_checked);
 444         des->ln_items_repaired = cpu_to_le64(src->ln_items_repaired);
 445         des->ln_items_failed = cpu_to_le64(src->ln_items_failed);
 446         des->ln_dirs_checked = cpu_to_le64(src->ln_dirs_checked);
 447         des->ln_mlinked_checked = cpu_to_le64(src->ln_mlinked_checked);
 448         des->ln_objs_checked_phase2 = cpu_to_le64(src->ln_objs_checked_phase2);
 449         des->ln_objs_repaired_phase2 =
 450                                 cpu_to_le64(src->ln_objs_repaired_phase2);
 451         des->ln_objs_failed_phase2 = cpu_to_le64(src->ln_objs_failed_phase2);
 452         des->ln_objs_nlink_repaired = cpu_to_le64(src->ln_objs_nlink_repaired);
 453         des->ln_objs_lost_found = cpu_to_le64(src->ln_objs_lost_found);
 454         fid_cpu_to_le(&des->ln_fid_latest_scanned_phase2,
 455                       &src->ln_fid_latest_scanned_phase2);
 456 }
 457
 458 /**
 459  * \retval +ve: the lfsck_namespace is broken, the caller should reset it.
 460  * \retval 0: succeed.
 461  * \retval -ve: failed cases.
 462  */
 463 static int mdd_lfsck_namespace_load(const struct lu_env *env,
 464                                     struct lfsck_component *com)
 465 {
 466         int len = com->lc_file_size;
 467         int rc;
 468
 469         rc = dt_xattr_get(env, com->lc_obj,
 470                           mdd_buf_get(env, com->lc_file_disk, len),
 471                           XATTR_NAME_LFSCK_NAMESPACE, BYPASS_CAPA);
 472         if (rc == len) {
 473                 struct lfsck_namespace *ns = com->lc_file_ram;
 474
 475                 mdd_lfsck_namespace_to_cpu(ns,
 476                                 (struct lfsck_namespace *)com->lc_file_disk);
 477                 if (ns->ln_magic != LFSCK_NAMESPACE_MAGIC) {
 478                         CWARN("%.16s: invalid lfsck_namespace magic "
 479                               "0x%x != 0x%x\n",
 480                               mdd_lfsck2name(com->lc_lfsck),
 481                               ns->ln_magic, LFSCK_NAMESPACE_MAGIC);
 482                         rc = 1;
 483                 } else {
 484                         rc = 0;
 485                 }
 486         } else if (rc != -ENODATA) {
 487                 CERROR("%.16s: fail to load lfsck_namespace, expected = %d, "
 488                        "rc = %d\n", mdd_lfsck2name(com->lc_lfsck), len, rc);
 489                 if (rc >= 0)
 490                         rc = 1;
 491         }
 492         return rc;
 493 }
 494
 495 static int mdd_lfsck_namespace_store(const struct lu_env *env,
 496                                      struct lfsck_component *com, bool init)
 497 {
 498         struct dt_object  *obj    = com->lc_obj;
 499         struct md_lfsck   *lfsck  = com->lc_lfsck;
 500         struct mdd_device *mdd    = mdd_lfsck2mdd(lfsck);
 501         struct thandle    *handle;
 502         int                len    = com->lc_file_size;
 503         int                rc;
 504         ENTRY;
 505
 506         mdd_lfsck_namespace_to_le((struct lfsck_namespace *)com->lc_file_disk,
 507                                   (struct lfsck_namespace *)com->lc_file_ram);
 508         handle = dt_trans_create(env, mdd->mdd_bottom);
 509         if (IS_ERR(handle)) {
 510                 rc = PTR_ERR(handle);
 511                 CERROR("%.16s: fail to create trans for storing "
 512                        "lfsck_namespace: %d\n,", mdd_lfsck2name(lfsck), rc);
 513                 RETURN(rc);
 514         }
 515
 516         rc = dt_declare_xattr_set(env, obj,
 517                                   mdd_buf_get(env, com->lc_file_disk, len),
 518                                   XATTR_NAME_LFSCK_NAMESPACE, 0, handle);
 519         if (rc != 0) {
 520                 CERROR("%.16s: fail to declare trans for storing "
 521                        "lfsck_namespace: %d\n,", mdd_lfsck2name(lfsck), rc);
 522                 GOTO(out, rc);
 523         }
 524
 525         rc = dt_trans_start_local(env, mdd->mdd_bottom, handle);
 526         if (rc != 0) {
 527                 CERROR("%.16s: fail to start trans for storing "
 528                        "lfsck_namespace: %d\n,", mdd_lfsck2name(lfsck), rc);
 529                 GOTO(out, rc);
 530         }
 531
 532         rc = dt_xattr_set(env, obj,
 533                           mdd_buf_get(env, com->lc_file_disk, len),
 534                           XATTR_NAME_LFSCK_NAMESPACE,
 535                           init ? LU_XATTR_CREATE : LU_XATTR_REPLACE,
 536                           handle, BYPASS_CAPA);
 537         if (rc != 0)
 538                 CERROR("%.16s: fail to store lfsck_namespace, len = %d, "
 539                        "rc = %d\n", mdd_lfsck2name(lfsck), len, rc);
 540
 541         GOTO(out, rc);
 542
 543 out:
 544         dt_trans_stop(env, mdd->mdd_bottom, handle);
 545         return rc;
 546 }
 547
 548 static int mdd_lfsck_namespace_init(const struct lu_env *env,
 549                                     struct lfsck_component *com)
 550 {
 551         struct lfsck_namespace *ns = (struct lfsck_namespace *)com->lc_file_ram;
 552         int rc;
 553
 554         memset(ns, 0, sizeof(*ns));
 555         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
 556         ns->ln_status = LS_INIT;
 557         down_write(&com->lc_sem);
 558         rc = mdd_lfsck_namespace_store(env, com, true);
 559         up_write(&com->lc_sem);
 560         return rc;
 561 }
 562
 563 static int mdd_declare_lfsck_namespace_unlink(const struct lu_env *env,
 564                                               struct mdd_device *mdd,
 565                                               struct dt_object *p,
 566                                               struct dt_object *c,
 567                                               const char *name,
 568                                               struct thandle *handle)
 569 {
 570         int rc;
 571
 572         rc = dt_declare_delete(env, p, (const struct dt_key *)name, handle);
 573         if (rc != 0)
 574                 return rc;
 575
 576         rc = dt_declare_ref_del(env, c, handle);
 577         if (rc != 0)
 578                 return rc;
 579
 580         rc = dt_declare_destroy(env, c, handle);
 581         return rc;
 582 }
 583
 584 static int mdd_lfsck_namespace_unlink(const struct lu_env *env,
 585                                       struct mdd_device *mdd,
 586                                       struct lfsck_component *com)
 587 {
 588         struct mdd_thread_info  *info   = mdd_env_info(env);
 589         struct lu_fid           *fid    = &info->mti_fid;
 590         struct dt_object        *child  = com->lc_obj;
 591         struct dt_object        *parent;
 592         struct thandle          *handle;
 593         int                      rc;
 594         ENTRY;
 595
 596         parent = dt_store_resolve(env, mdd->mdd_bottom, "", fid);
 597         if (IS_ERR(parent))
 598                 RETURN(rc = PTR_ERR(parent));
 599
 600         if (dt_try_as_dir(env, parent))
 601                 GOTO(out, rc = -ENOTDIR);
 602
 603         handle = dt_trans_create(env, mdd->mdd_bottom);
 604         if (IS_ERR(handle))
 605                 GOTO(out, rc = PTR_ERR(handle));
 606
 607         rc = mdd_declare_lfsck_namespace_unlink(env, mdd, parent, child,
 608                                                 lfsck_namespace_name, handle);
 609         if (rc != 0)
 610                 GOTO(stop, rc);
 611
 612         rc = dt_trans_start_local(env, mdd->mdd_bottom, handle);
 613         if (rc != 0)
 614                 GOTO(stop, rc);
 615
 616         rc = dt_delete(env, parent, (struct dt_key *)lfsck_namespace_name,
 617                        handle, BYPASS_CAPA);
 618         if (rc != 0)
 619                 GOTO(stop, rc);
 620
 621         rc = child->do_ops->do_ref_del(env, child, handle);
 622         if (rc != 0) {
 623                 lu_local_obj_fid(fid, LFSCK_NAMESPACE_OID);
 624                 rc = dt_insert(env, parent,
 625                                (const struct dt_rec*)fid,
 626                                (const struct dt_key *)lfsck_namespace_name,
 627                                handle, BYPASS_CAPA, 1);
 628
 629                 GOTO(stop, rc);
 630         }
 631
 632
 633         rc = dt_destroy(env, child, handle);
 634         if (rc == 0) {
 635                 lu_object_put(env, &child->do_lu);
 636                 com->lc_obj = NULL;
 637         }
 638
 639         GOTO(stop, rc);
 640
 641 stop:
 642         dt_trans_stop(env, mdd->mdd_bottom, handle);
 643
 644 out:
 645         lu_object_put(env, &parent->do_lu);
 646         return rc;
 647 }
 648
 649 /* namespace APIs */
 650
 651 static int mdd_lfsck_namespace_reset(const struct lu_env *env,
 652                                      struct lfsck_component *com, bool init)
 653 {
 654         struct mdd_thread_info  *info = mdd_env_info(env);
 655         struct lu_fid           *fid  = &info->mti_fid;
 656         struct lfsck_namespace  *ns   = (struct lfsck_namespace *)com->lc_file_ram;
 657         struct mdd_device       *mdd  = mdd_lfsck2mdd(com->lc_lfsck);
 658         struct md_object        *mdo;
 659         struct dt_object        *dto;
 660         int                      rc;
 661         ENTRY;
 662
 663         down_write(&com->lc_sem);
 664         if (init) {
 665                 memset(ns, 0, sizeof(*ns));
 666         } else {
 667                 __u32 count = ns->ln_success_count;
 668                 __u64 last_time = ns->ln_time_last_complete;
 669
 670                 memset(ns, 0, sizeof(*ns));
 671                 ns->ln_success_count = count;
 672                 ns->ln_time_last_complete = last_time;
 673         }
 674         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
 675         ns->ln_status = LS_INIT;
 676
 677         rc = mdd_lfsck_namespace_unlink(env, mdd, com);
 678         if (rc != 0)
 679                 GOTO(out, rc);
 680
 681         lu_local_obj_fid(fid, LFSCK_NAMESPACE_OID);
 682         mdo = llo_store_create_index(env, &mdd->mdd_md_dev, mdd->mdd_bottom, "",
 683                                      lfsck_namespace_name, fid,
 684                                      &dt_lfsck_features);
 685         if (IS_ERR(mdo))
 686                 GOTO(out, rc = PTR_ERR(mdo));
 687
 688         lu_object_put(env, &mdo->mo_lu);
 689         dto = dt_store_open(env, mdd->mdd_bottom, "", lfsck_namespace_name, fid);
 690         if (IS_ERR(dto))
 691                 GOTO(out, rc = PTR_ERR(dto));
 692
 693         com->lc_obj = dto;
 694         rc = dto->do_ops->do_index_try(env, dto, &dt_lfsck_features);
 695         if (rc != 0)
 696                 GOTO(out, rc);
 697
 698         rc = mdd_lfsck_namespace_store(env, com, true);
 699
 700         GOTO(out, rc);
 701
 702 out:
 703         up_write(&com->lc_sem);
 704         return rc;
 705 }
 706
 707 static void
 708 mdd_lfsck_namespace_fail(const struct lu_env *env, struct lfsck_component *com,
 709                          bool oit, bool new_checked)
 710 {
 711         struct lfsck_namespace *ns = (struct lfsck_namespace *)com->lc_file_ram;
 712
 713         down_write(&com->lc_sem);
 714         if (new_checked)
 715                 com->lc_new_checked++;
 716         ns->ln_items_failed++;
 717         if (mdd_lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent))
 718                 mdd_lfsck_pos_fill(env, com->lc_lfsck,
 719                                    &ns->ln_pos_first_inconsistent, oit, !oit);
 720         up_write(&com->lc_sem);
 721 }
 722
 723 static int mdd_lfsck_namespace_checkpoint(const struct lu_env *env,
 724                                           struct lfsck_component *com,
 725                                           bool init)
 726 {
 727         struct md_lfsck         *lfsck = com->lc_lfsck;
 728         struct lfsck_namespace  *ns    =
 729                                 (struct lfsck_namespace *)com->lc_file_ram;
 730         int                      rc;
 731
 732         if (com->lc_new_checked == 0 && !init)
 733                 return 0;
 734
 735         down_write(&com->lc_sem);
 736
 737         ns->ln_pos_last_checkpoint = lfsck->ml_pos_current;
 738         if (init) {
 739                 ns->ln_time_last_checkpoint = ns->ln_time_latest_start;
 740                 ns->ln_pos_latest_start = lfsck->ml_pos_current;
 741         } else {
 742                 ns->ln_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
 743                                 HALF_SEC - lfsck->ml_time_last_checkpoint);
 744                 ns->ln_time_last_checkpoint = cfs_time_current_sec();
 745                 ns->ln_items_checked += com->lc_new_checked;
 746                 com->lc_new_checked = 0;
 747         }
 748
 749         rc = mdd_lfsck_namespace_store(env, com, false);
 750
 751         up_write(&com->lc_sem);
 752         return rc;
 753 }
 754
 755 static int mdd_lfsck_namespace_prep(const struct lu_env *env,
 756                                     struct lfsck_component *com)
 757 {
 758         struct md_lfsck         *lfsck  = com->lc_lfsck;
 759         struct lfsck_namespace  *ns     =
 760                                 (struct lfsck_namespace *)com->lc_file_ram;
 761         struct lfsck_position   *pos    = &com->lc_pos_start;
 762
 763         if (ns->ln_status == LS_COMPLETED) {
 764                 int rc;
 765
 766                 rc = mdd_lfsck_namespace_reset(env, com, false);
 767                 if (rc != 0)
 768                         return rc;
 769         }
 770
 771         down_write(&com->lc_sem);
 772
 773         ns->ln_time_latest_start = cfs_time_current_sec();
 774
 775         spin_lock(&lfsck->ml_lock);
 776         if (ns->ln_flags & LF_SCANNED_ONCE) {
 777                 if (!lfsck->ml_drop_dryrun ||
 778                     mdd_lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent)) {
 779                         ns->ln_status = LS_SCANNING_PHASE2;
 780                         cfs_list_del_init(&com->lc_link);
 781                         cfs_list_add_tail(&com->lc_link,
 782                                           &lfsck->ml_list_double_scan);
 783                         if (!cfs_list_empty(&com->lc_link_dir))
 784                                 cfs_list_del_init(&com->lc_link_dir);
 785                         mdd_lfsck_pos_set_zero(pos);
 786                 } else {
 787                         ns->ln_status = LS_SCANNING_PHASE1;
 788                         ns->ln_run_time_phase1 = 0;
 789                         ns->ln_run_time_phase2 = 0;
 790                         ns->ln_items_checked = 0;
 791                         ns->ln_items_repaired = 0;
 792                         ns->ln_items_failed = 0;
 793                         ns->ln_dirs_checked = 0;
 794                         ns->ln_mlinked_checked = 0;
 795                         ns->ln_objs_checked_phase2 = 0;
 796                         ns->ln_objs_repaired_phase2 = 0;
 797                         ns->ln_objs_failed_phase2 = 0;
 798                         ns->ln_objs_nlink_repaired = 0;
 799                         ns->ln_objs_lost_found = 0;
 800                         fid_zero(&ns->ln_fid_latest_scanned_phase2);
 801                         if (cfs_list_empty(&com->lc_link_dir))
 802                                 cfs_list_add_tail(&com->lc_link_dir,
 803                                                   &lfsck->ml_list_dir);
 804                         *pos = ns->ln_pos_first_inconsistent;
 805                 }
 806         } else {
 807                 ns->ln_status = LS_SCANNING_PHASE1;
 808                 if (cfs_list_empty(&com->lc_link_dir))
 809                         cfs_list_add_tail(&com->lc_link_dir,
 810                                           &lfsck->ml_list_dir);
 811                 if (!lfsck->ml_drop_dryrun ||
 812                     mdd_lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent)) {
 813                         *pos = ns->ln_pos_last_checkpoint;
 814                         pos->lp_oit_cookie++;
 815                         if (!fid_is_zero(&pos->lp_dir_parent)) {
 816                                 if (pos->lp_dir_cookie == MDS_DIR_END_OFF) {
 817                                         fid_zero(&pos->lp_dir_parent);
 818                                 } else {
 819                                         pos->lp_dir_cookie++;
 820                                 }
 821                         }
 822                 } else {
 823                         *pos = ns->ln_pos_first_inconsistent;
 824                 }
 825         }
 826         spin_unlock(&lfsck->ml_lock);
 827
 828         up_write(&com->lc_sem);
 829         return 0;
 830 }
 831
 832 static int mdd_lfsck_namespace_exec_oit(const struct lu_env *env,
 833                                         struct lfsck_component *com,
 834                                         struct mdd_object *obj)
 835 {
 836         down_write(&com->lc_sem);
 837         com->lc_new_checked++;
 838         if (S_ISDIR(mdd_object_type(obj)))
 839                 ((struct lfsck_namespace *)com->lc_file_ram)->ln_dirs_checked++;
 840         up_write(&com->lc_sem);
 841         return 0;
 842 }
 843
 844 /* XXX: to be implemented in other patch.  */
 845 static int mdd_lfsck_namespace_exec_dir(const struct lu_env *env,
 846                                         struct lfsck_component *com,
 847                                         struct mdd_object *obj,
 848                                         struct lu_dirent *ent)
 849 {
 850         struct lfsck_namespace     *ns       =
 851                                 (struct lfsck_namespace *)com->lc_file_ram;
 852         const struct lu_name       *cname;
 853         int                         repaired;
 854
 855         cname = mdd_name_get_const(env, ent->lde_name, ent->lde_namelen);
 856         down_write(&com->lc_sem);
 857         com->lc_new_checked++;
 858
 859         if (ent->lde_attrs & LUDA_UPGRADE) {
 860                 ns->ln_flags |= LF_UPGRADE;
 861                 repaired = 1;
 862         } else if (ent->lde_attrs & LUDA_REPAIR) {
 863                 ns->ln_flags |= LF_INCONSISTENT;
 864                 repaired = 1;
 865         } else {
 866                 repaired = 0;
 867         }
 868
 869         ns->ln_items_repaired += repaired;
 870         up_write(&com->lc_sem);
 871         return 0;
 872 }
 873
 874 static int mdd_lfsck_namespace_post(const struct lu_env *env,
 875                                     struct lfsck_component *com,
 876                                     int result)
 877 {
 878         struct md_lfsck         *lfsck = com->lc_lfsck;
 879         struct lfsck_namespace  *ns    =
 880                                 (struct lfsck_namespace *)com->lc_file_ram;
 881         int                      rc;
 882
 883         down_write(&com->lc_sem);
 884
 885         spin_lock(&lfsck->ml_lock);
 886         if (result > 0) {
 887                 ns->ln_status = LS_SCANNING_PHASE2;
 888                 ns->ln_flags |= LF_SCANNED_ONCE;
 889                 ns->ln_flags &= ~LF_UPGRADE;
 890                 cfs_list_del_init(&com->lc_link);
 891                 cfs_list_del_init(&com->lc_link_dir);
 892                 cfs_list_add_tail(&com->lc_link, &lfsck->ml_list_double_scan);
 893         } else if (result == 0) {
 894                 if (lfsck->ml_paused) {
 895                         ns->ln_status = LS_PAUSED;
 896                 } else {
 897                         ns->ln_status = LS_STOPPED;
 898                         cfs_list_del_init(&com->lc_link);
 899                         cfs_list_del_init(&com->lc_link_dir);
 900                         cfs_list_add_tail(&com->lc_link, &lfsck->ml_list_idle);
 901                 }
 902         } else {
 903                 ns->ln_status = LS_FAILED;
 904                 cfs_list_del_init(&com->lc_link);
 905                 cfs_list_del_init(&com->lc_link_dir);
 906                 cfs_list_add_tail(&com->lc_link, &lfsck->ml_list_idle);
 907         }
 908         spin_unlock(&lfsck->ml_lock);
 909
 910         ns->ln_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
 911                                 HALF_SEC - lfsck->ml_time_last_checkpoint);
 912         ns->ln_time_last_checkpoint = cfs_time_current_sec();
 913         ns->ln_items_checked += com->lc_new_checked;
 914         com->lc_new_checked = 0;
 915
 916         rc = mdd_lfsck_namespace_store(env, com, false);
 917
 918         up_write(&com->lc_sem);
 919         return rc;
 920 }
 921
 922 /* XXX: to be implemented in other patch.  */
 923 static int
 924 mdd_lfsck_namespace_dump(const struct lu_env *env, struct lfsck_component *com,
 925                          char *buf, int len)
 926 {
 927         return 0;
 928 }
 929
 930 /* XXX: to be implemented in other patch.  */
 931 static int mdd_lfsck_namespace_double_scan(const struct lu_env *env,
 932                                            struct lfsck_component *com)
 933 {
 934         struct md_lfsck         *lfsck  = com->lc_lfsck;
 935         struct lfsck_bookmark   *bk     = &lfsck->ml_bookmark_ram;
 936         struct lfsck_namespace  *ns     =
 937                                 (struct lfsck_namespace *)com->lc_file_ram;
 938         int                      rc;
 939
 940         down_write(&com->lc_sem);
 941
 942         ns->ln_time_last_checkpoint = cfs_time_current_sec();
 943         com->lc_new_checked = 0;
 944         com->lc_journal = 0;
 945
 946         ns->ln_status = LS_COMPLETED;
 947         if (!(bk->lb_param & LPF_DRYRUN))
 948                 ns->ln_flags &=
 949                 ~(LF_SCANNED_ONCE | LF_INCONSISTENT | LF_UPGRADE);
 950         ns->ln_time_last_complete = ns->ln_time_last_checkpoint;
 951         ns->ln_success_count++;
 952
 953         spin_lock(&lfsck->ml_lock);
 954         cfs_list_del_init(&com->lc_link);
 955         cfs_list_add_tail(&com->lc_link, &lfsck->ml_list_idle);
 956         spin_unlock(&lfsck->ml_lock);
 957
 958         rc = mdd_lfsck_namespace_store(env, com, false);
 959
 960         up_write(&com->lc_sem);
 961         return rc;
 962 }
 963
 964 static struct lfsck_operations mdd_lfsck_namespace_ops = {
 965         .lfsck_reset            = mdd_lfsck_namespace_reset,
 966         .lfsck_fail             = mdd_lfsck_namespace_fail,
 967         .lfsck_checkpoint       = mdd_lfsck_namespace_checkpoint,
 968         .lfsck_prep             = mdd_lfsck_namespace_prep,
 969         .lfsck_exec_oit         = mdd_lfsck_namespace_exec_oit,
 970         .lfsck_exec_dir         = mdd_lfsck_namespace_exec_dir,
 971         .lfsck_post             = mdd_lfsck_namespace_post,
 972         .lfsck_dump             = mdd_lfsck_namespace_dump,
 973         .lfsck_double_scan      = mdd_lfsck_namespace_double_scan,
 974 };
 975
 976 /* LFSCK component setup/cleanup functions */
 977
 978 static int mdd_lfsck_namespace_setup(const struct lu_env *env,
 979                                      struct md_lfsck *lfsck)
 980 {
 981         struct mdd_device      *mdd = mdd_lfsck2mdd(lfsck);
 982         struct lfsck_component *com;
 983         struct lfsck_namespace *ns;
 984         struct dt_object       *obj;
 985         int                     rc;
 986         ENTRY;
 987
 988         OBD_ALLOC_PTR(com);
 989         if (com == NULL)
 990                 RETURN(-ENOMEM);
 991
 992         CFS_INIT_LIST_HEAD(&com->lc_link);
 993         CFS_INIT_LIST_HEAD(&com->lc_link_dir);
 994         init_rwsem(&com->lc_sem);
 995         atomic_set(&com->lc_ref, 1);
 996         com->lc_lfsck = lfsck;
 997         com->lc_type = LT_NAMESPACE;
 998         com->lc_ops = &mdd_lfsck_namespace_ops;
 999         com->lc_file_size = sizeof(struct lfsck_namespace);
1000         OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
1001         if (com->lc_file_ram == NULL)
1002                 GOTO(out, rc = -ENOMEM);
1003
1004         OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
1005         if (com->lc_file_disk == NULL)
1006                 GOTO(out, rc = -ENOMEM);
1007
1008         obj = dt_store_open(env, mdd->mdd_bottom, "", lfsck_namespace_name,
1009                             &mdd_env_info(env)->mti_fid);
1010         if (IS_ERR(obj))
1011                 GOTO(out, rc = PTR_ERR(obj));
1012
1013         com->lc_obj = obj;
1014         rc = obj->do_ops->do_index_try(env, obj, &dt_lfsck_features);
1015         if (rc != 0)
1016                 GOTO(out, rc);
1017
1018         rc = mdd_lfsck_namespace_load(env, com);
1019         if (rc > 0)
1020                 rc = mdd_lfsck_namespace_reset(env, com, true);
1021         else if (rc == -ENODATA)
1022                 rc = mdd_lfsck_namespace_init(env, com);
1023         if (rc != 0)
1024                 GOTO(out, rc);
1025
1026         ns = (struct lfsck_namespace *)com->lc_file_ram;
1027         switch (ns->ln_status) {
1028         case LS_INIT:
1029         case LS_COMPLETED:
1030         case LS_FAILED:
1031         case LS_STOPPED:
1032                 cfs_list_add_tail(&com->lc_link, &lfsck->ml_list_idle);
1033                 break;
1034         default:
1035                 CERROR("%s: unknown status: %u\n",
1036                        mdd_lfsck2name(lfsck), ns->ln_status);
1037                 /* fall through */
1038         case LS_SCANNING_PHASE1:
1039         case LS_SCANNING_PHASE2:
1040                 /* No need to store the status to disk right now.
1041                  * If the system crashed before the status stored,
1042                  * it will be loaded back when next time. */
1043                 ns->ln_status = LS_CRASHED;
1044                 /* fall through */
1045         case LS_PAUSED:
1046         case LS_CRASHED:
1047                 cfs_list_add_tail(&com->lc_link, &lfsck->ml_list_scan);
1048                 cfs_list_add_tail(&com->lc_link_dir, &lfsck->ml_list_dir);
1049                 break;
1050         }
1051
1052         GOTO(out, rc = 0);
1053
1054 out:
1055         if (rc != 0)
1056                 mdd_lfsck_component_cleanup(env, com);
1057         return rc;
1058 }
1059
1060 /* helper functions for framework */
1061
1062 static int object_is_client_visible(const struct lu_env *env,
1063                                     struct mdd_device *mdd,
1064                                     struct mdd_object *obj)
1065 {
1066         struct lu_fid *fid   = &mdd_env_info(env)->mti_fid;
1067         int            depth = 0;
1068         int            rc;
1069
1070         LASSERT(S_ISDIR(mdd_object_type(obj)));
1071
1072         while (1) {
1073                 if (mdd_is_root(mdd, mdo2fid(obj))) {
1074                         if (depth > 0)
1075                                 mdd_object_put(env, obj);
1076                         return 1;
1077                 }
1078
1079                 mdd_read_lock(env, obj, MOR_TGT_CHILD);
1080                 if (unlikely(mdd_is_dead_obj(obj))) {
1081                         mdd_read_unlock(env, obj);
1082                         if (depth > 0)
1083                                 mdd_object_put(env, obj);
1084                         return 0;
1085                 }
1086
1087                 rc = dt_xattr_get(env, mdd_object_child(obj),
1088                                   mdd_buf_get(env, NULL, 0), XATTR_NAME_LINK,
1089                                   BYPASS_CAPA);
1090                 mdd_read_unlock(env, obj);
1091                 if (rc >= 0) {
1092                         if (depth > 0)
1093                                 mdd_object_put(env, obj);
1094                         return 1;
1095                 }
1096
1097                 if (rc < 0 && rc != -ENODATA) {
1098                         if (depth > 0)
1099                                 mdd_object_put(env, obj);
1100                         return rc;
1101                 }
1102
1103                 rc = mdd_parent_fid(env, obj, fid);
1104                 if (depth > 0)
1105                         mdd_object_put(env, obj);
1106                 if (rc != 0)
1107                         return rc;
1108
1109                 if (unlikely(lu_fid_eq(fid, &mdd->mdd_local_root_fid)))
1110                         return 0;
1111
1112                 obj = mdd_object_find(env, mdd, fid);
1113                 if (obj == NULL)
1114                         return 0;
1115                 else if (IS_ERR(obj))
1116                         return PTR_ERR(obj);
1117
1118                 /* XXX: need more processing for remote object in the future. */
1119                 if (!mdd_object_exists(obj) || mdd_object_remote(obj)) {
1120                         mdd_object_put(env, obj);
1121                         return 0;
1122                 }
1123
1124                 depth++;
1125         }
1126         return 0;
1127 }
1128
1129 static void mdd_lfsck_unpack_ent(struct lu_dirent *ent)
1130 {
1131         fid_le_to_cpu(&ent->lde_fid, &ent->lde_fid);
1132         ent->lde_hash = le64_to_cpu(ent->lde_hash);
1133         ent->lde_reclen = le16_to_cpu(ent->lde_reclen);
1134         ent->lde_namelen = le16_to_cpu(ent->lde_namelen);
1135         ent->lde_attrs = le32_to_cpu(ent->lde_attrs);
1136
1137         /* Make sure the name is terminated with '0'.
1138          * The data (type) after ent::lde_name maybe
1139          * broken, but we do not care. */
1140         ent->lde_name[ent->lde_namelen] = 0;
1141 }
1142
1143 /* LFSCK wrap functions */
1144
1145 static void mdd_lfsck_fail(const struct lu_env *env, struct md_lfsck *lfsck,
1146                            bool oit, bool new_checked)
1147 {
1148         struct lfsck_component *com;
1149
1150         cfs_list_for_each_entry(com, &lfsck->ml_list_scan, lc_link) {
1151                 com->lc_ops->lfsck_fail(env, com, oit, new_checked);
1152         }
1153 }
1154
1155 static int mdd_lfsck_checkpoint(const struct lu_env *env,
1156                                 struct md_lfsck *lfsck, bool oit)
1157 {
1158         struct lfsck_component *com;
1159         int                     rc;
1160
1161         if (likely(cfs_time_beforeq(cfs_time_current(),
1162                                     lfsck->ml_time_next_checkpoint)))
1163                 return 0;
1164
1165         mdd_lfsck_pos_fill(env, lfsck, &lfsck->ml_pos_current, oit, !oit);
1166         cfs_list_for_each_entry(com, &lfsck->ml_list_scan, lc_link) {
1167                 rc = com->lc_ops->lfsck_checkpoint(env, com, false);
1168                 if (rc != 0)
1169                         return rc;;
1170         }
1171
1172         lfsck->ml_time_last_checkpoint = cfs_time_current();
1173         lfsck->ml_time_next_checkpoint = lfsck->ml_time_last_checkpoint +
1174                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
1175         return 0;
1176 }
1177
1178 static int mdd_lfsck_prep(struct lu_env *env, struct md_lfsck *lfsck)
1179 {
1180         struct mdd_device      *mdd     = mdd_lfsck2mdd(lfsck);
1181         struct mdd_object      *obj     = NULL;
1182         struct dt_object       *dt_obj;
1183         struct lfsck_component *com;
1184         struct lfsck_component *next;
1185         struct lfsck_position  *pos     = NULL;
1186         const struct dt_it_ops *iops    =
1187                                 &lfsck->ml_obj_oit->do_index_ops->dio_it;
1188         struct dt_it           *di;
1189         int                     rc;
1190         ENTRY;
1191
1192         LASSERT(lfsck->ml_obj_dir == NULL);
1193         LASSERT(lfsck->ml_di_dir == NULL);
1194
1195         cfs_list_for_each_entry_safe(com, next, &lfsck->ml_list_scan, lc_link) {
1196                 com->lc_new_checked = 0;
1197                 if (lfsck->ml_bookmark_ram.lb_param & LPF_DRYRUN)
1198                         com->lc_journal = 0;
1199
1200                 rc = com->lc_ops->lfsck_prep(env, com);
1201                 if (rc != 0)
1202                         RETURN(rc);
1203
1204                 if ((pos == NULL) ||
1205                     (!mdd_lfsck_pos_is_zero(&com->lc_pos_start) &&
1206                      mdd_lfsck_pos_is_eq(pos, &com->lc_pos_start) > 0))
1207                         pos = &com->lc_pos_start;
1208         }
1209
1210         /* Init otable-based iterator. */
1211         if (pos == NULL) {
1212                 rc = iops->load(env, lfsck->ml_di_oit, 0);
1213                 GOTO(out, rc = (rc >= 0 ? 0 : rc));
1214         }
1215
1216         rc = iops->load(env, lfsck->ml_di_oit, pos->lp_oit_cookie);
1217         if (rc < 0)
1218                 GOTO(out, rc);
1219
1220         if (fid_is_zero(&pos->lp_dir_parent))
1221                 GOTO(out, rc = 0);
1222
1223         /* Find the directory for namespace-based traverse. */
1224         obj = mdd_object_find(env, mdd, &pos->lp_dir_parent);
1225         if (obj == NULL)
1226                 GOTO(out, rc = 0);
1227         else if (IS_ERR(obj))
1228                 RETURN(PTR_ERR(obj));
1229
1230         /* XXX: need more processing for remote object in the future. */
1231         if (!mdd_object_exists(obj) || mdd_object_remote(obj) ||
1232             unlikely(!S_ISDIR(mdd_object_type(obj))))
1233                 GOTO(out, rc = 0);
1234
1235         if (unlikely(mdd_is_dead_obj(obj)))
1236                 GOTO(out, rc = 0);
1237
1238         dt_obj = mdd_object_child(obj);
1239         if (unlikely(!dt_try_as_dir(env, dt_obj)))
1240                 GOTO(out, rc = -ENOTDIR);
1241
1242         /* Init the namespace-based directory traverse. */
1243         iops = &dt_obj->do_index_ops->dio_it;
1244         di = iops->init(env, dt_obj, lfsck->ml_args_dir, BYPASS_CAPA);
1245         if (IS_ERR(di))
1246                 GOTO(out, rc = PTR_ERR(di));
1247
1248         rc = iops->load(env, di, pos->lp_dir_cookie);
1249         if (rc == 0)
1250                 rc = iops->next(env, di);
1251         else if (rc > 0)
1252                 rc = 0;
1253
1254         if (rc != 0) {
1255                 iops->put(env, di);
1256                 iops->fini(env, di);
1257                 GOTO(out, rc);
1258         }
1259
1260         lfsck->ml_obj_dir = dt_obj;
1261         spin_lock(&lfsck->ml_lock);
1262         lfsck->ml_di_dir = di;
1263         spin_unlock(&lfsck->ml_lock);
1264         obj = NULL;
1265
1266         GOTO(out, rc = 0);
1267
1268 out:
1269         if (obj != NULL)
1270                 mdd_object_put(env, obj);
1271
1272         if (rc != 0)
1273                 return (rc > 0 ? 0 : rc);
1274
1275         mdd_lfsck_pos_fill(env, lfsck, &lfsck->ml_pos_current, false, false);
1276         cfs_list_for_each_entry(com, &lfsck->ml_list_scan, lc_link) {
1277                 rc = com->lc_ops->lfsck_checkpoint(env, com, true);
1278                 if (rc != 0)
1279                         break;
1280         }
1281
1282         lfsck->ml_time_last_checkpoint = cfs_time_current();
1283         lfsck->ml_time_next_checkpoint = lfsck->ml_time_last_checkpoint +
1284                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
1285         return rc;
1286 }
1287
1288 static int mdd_lfsck_exec_oit(const struct lu_env *env, struct md_lfsck *lfsck,
1289                               struct mdd_object *obj)
1290 {
1291         struct lfsck_component *com;
1292         struct dt_object       *dt_obj;
1293         const struct dt_it_ops *iops;
1294         struct dt_it           *di;
1295         int                     rc;
1296         ENTRY;
1297
1298         LASSERT(lfsck->ml_obj_dir == NULL);
1299
1300         cfs_list_for_each_entry(com, &lfsck->ml_list_scan, lc_link) {
1301                 rc = com->lc_ops->lfsck_exec_oit(env, com, obj);
1302                 if (rc != 0)
1303                         RETURN(rc);
1304         }
1305
1306         if (!S_ISDIR(mdd_object_type(obj)) ||
1307             cfs_list_empty(&lfsck->ml_list_dir))
1308                RETURN(0);
1309
1310         rc = object_is_client_visible(env, mdd_lfsck2mdd(lfsck), obj);
1311         if (rc <= 0)
1312                 GOTO(out, rc);
1313
1314         if (unlikely(mdd_is_dead_obj(obj)))
1315                 GOTO(out, rc = 0);
1316
1317         dt_obj = mdd_object_child(obj);
1318         if (unlikely(!dt_try_as_dir(env, dt_obj)))
1319                 GOTO(out, rc = -ENOTDIR);
1320
1321         iops = &dt_obj->do_index_ops->dio_it;
1322         di = iops->init(env, dt_obj, lfsck->ml_args_dir, BYPASS_CAPA);
1323         if (IS_ERR(di))
1324                 GOTO(out, rc = PTR_ERR(di));
1325
1326         rc = iops->load(env, di, 0);
1327         if (rc == 0)
1328                 rc = iops->next(env, di);
1329         else if (rc > 0)
1330                 rc = 0;
1331
1332         if (rc != 0) {
1333                 iops->put(env, di);
1334                 iops->fini(env, di);
1335                 GOTO(out, rc);
1336         }
1337
1338         mdd_object_get(obj);
1339         lfsck->ml_obj_dir = dt_obj;
1340         spin_lock(&lfsck->ml_lock);
1341         lfsck->ml_di_dir = di;
1342         spin_unlock(&lfsck->ml_lock);
1343
1344         GOTO(out, rc = 0);
1345
1346 out:
1347         if (rc < 0)
1348                 mdd_lfsck_fail(env, lfsck, false, false);
1349         return (rc > 0 ? 0 : rc);
1350 }
1351
1352 static int mdd_lfsck_exec_dir(const struct lu_env *env, struct md_lfsck *lfsck,
1353                               struct mdd_object *obj, struct lu_dirent *ent)
1354 {
1355         struct lfsck_component *com;
1356         int                     rc;
1357
1358         cfs_list_for_each_entry(com, &lfsck->ml_list_scan, lc_link) {
1359                 rc = com->lc_ops->lfsck_exec_dir(env, com, obj, ent);
1360                 if (rc != 0)
1361                         return rc;
1362         }
1363         return 0;
1364 }
1365
1366 static int mdd_lfsck_post(const struct lu_env *env, struct md_lfsck *lfsck,
1367                           int result)
1368 {
1369         struct lfsck_component *com;
1370         struct lfsck_component *next;
1371         int                     rc;
1372
1373         mdd_lfsck_pos_fill(env, lfsck, &lfsck->ml_pos_current, true, true);
1374         cfs_list_for_each_entry_safe(com, next, &lfsck->ml_list_scan, lc_link) {
1375                 rc = com->lc_ops->lfsck_post(env, com, result);
1376                 if (rc != 0)
1377                         return rc;
1378         }
1379
1380         lfsck->ml_time_last_checkpoint = cfs_time_current();
1381         lfsck->ml_time_next_checkpoint = lfsck->ml_time_last_checkpoint +
1382                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
1383         return result;
1384 }
1385
1386 static int mdd_lfsck_double_scan(const struct lu_env *env,
1387                                  struct md_lfsck *lfsck)
1388 {
1389         struct lfsck_component *com;
1390         struct lfsck_component *next;
1391         int                     rc;
1392
1393         cfs_list_for_each_entry_safe(com, next, &lfsck->ml_list_double_scan,
1394                                      lc_link) {
1395                 if (lfsck->ml_bookmark_ram.lb_param & LPF_DRYRUN)
1396                         com->lc_journal = 0;
1397
1398                 rc = com->lc_ops->lfsck_double_scan(env, com);
1399                 if (rc != 0)
1400                         return rc;
1401         }
1402         return 0;
1403 }
1404
1405 /* LFSCK engines */
1406
1407 static int mdd_lfsck_dir_engine(const struct lu_env *env,
1408                                 struct md_lfsck *lfsck)
1409 {
1410         struct mdd_thread_info  *info   = mdd_env_info(env);
1411         struct mdd_device       *mdd    = mdd_lfsck2mdd(lfsck);
1412         const struct dt_it_ops  *iops   =
1413                         &lfsck->ml_obj_dir->do_index_ops->dio_it;
1414         struct dt_it            *di     = lfsck->ml_di_dir;
1415         struct lu_dirent        *ent    = &info->mti_ent;
1416         struct lu_fid           *fid    = &info->mti_fid;
1417         struct lfsck_bookmark   *bk     = &lfsck->ml_bookmark_ram;
1418         struct ptlrpc_thread    *thread = &lfsck->ml_thread;
1419         int                      rc;
1420         ENTRY;
1421
1422         do {
1423                 struct mdd_object *child;
1424
1425                 lfsck->ml_new_scanned++;
1426                 rc = iops->rec(env, di, (struct dt_rec *)ent,
1427                                lfsck->ml_args_dir);
1428                 if (rc != 0) {
1429                         mdd_lfsck_fail(env, lfsck, false, true);
1430                         if (bk->lb_param & LPF_FAILOUT)
1431                                 RETURN(rc);
1432                         else
1433                                 goto checkpoint;
1434                 }
1435
1436                 mdd_lfsck_unpack_ent(ent);
1437                 if (ent->lde_attrs & LUDA_IGNORE)
1438                         goto checkpoint;
1439
1440                 *fid = ent->lde_fid;
1441                 child = mdd_object_find(env, mdd, fid);
1442                 if (child == NULL) {
1443                         goto checkpoint;
1444                 } else if (IS_ERR(child)) {
1445                         mdd_lfsck_fail(env, lfsck, false, true);
1446                         if (bk->lb_param & LPF_FAILOUT)
1447                                 RETURN(PTR_ERR(child));
1448                         else
1449                                 goto checkpoint;
1450                 }
1451
1452                 /* XXX: need more processing for remote object in the future. */
1453                 if (mdd_object_exists(child) && !mdd_object_remote(child))
1454                         rc = mdd_lfsck_exec_dir(env, lfsck, child, ent);
1455                 mdd_object_put(env, child);
1456                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
1457                         RETURN(rc);
1458
1459 checkpoint:
1460                 rc = mdd_lfsck_checkpoint(env, lfsck, false);
1461                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
1462                         RETURN(rc);
1463
1464                 /* Rate control. */
1465                 mdd_lfsck_control_speed(lfsck);
1466                 if (unlikely(!thread_is_running(thread)))
1467                         RETURN(0);
1468
1469                 rc = iops->next(env, di);
1470         } while (rc == 0);
1471
1472         if (rc > 0 && !lfsck->ml_oit_over)
1473                 mdd_lfsck_close_dir(env, lfsck);
1474
1475         RETURN(rc);
1476 }
1477
1478 static int mdd_lfsck_oit_engine(const struct lu_env *env,
1479                                 struct md_lfsck *lfsck)
1480 {
1481         struct mdd_thread_info  *info   = mdd_env_info(env);
1482         struct mdd_device       *mdd    = mdd_lfsck2mdd(lfsck);
1483         const struct dt_it_ops  *iops   =
1484                                 &lfsck->ml_obj_oit->do_index_ops->dio_it;
1485         struct dt_it            *di     = lfsck->ml_di_oit;
1486         struct lu_fid           *fid    = &info->mti_fid;
1487         struct lfsck_bookmark   *bk     = &lfsck->ml_bookmark_ram;
1488         struct ptlrpc_thread    *thread = &lfsck->ml_thread;
1489         int                      rc;
1490         ENTRY;
1491
1492         do {
1493                 struct mdd_object *target;
1494
1495                 if (lfsck->ml_di_dir != NULL) {
1496                         rc = mdd_lfsck_dir_engine(env, lfsck);
1497                         if (rc <= 0)
1498                                 RETURN(rc);
1499                 }
1500
1501                 if (unlikely(lfsck->ml_oit_over))
1502                         RETURN(1);
1503
1504                 lfsck->ml_new_scanned++;
1505                 rc = iops->rec(env, di, (struct dt_rec *)fid, 0);
1506                 if (rc != 0) {
1507                         mdd_lfsck_fail(env, lfsck, true, true);
1508                         if (bk->lb_param & LPF_FAILOUT)
1509                                 RETURN(rc);
1510                         else
1511                                 goto checkpoint;
1512                 }
1513
1514                 target = mdd_object_find(env, mdd, fid);
1515                 if (target == NULL) {
1516                         goto checkpoint;
1517                 } else if (IS_ERR(target)) {
1518                         mdd_lfsck_fail(env, lfsck, true, true);
1519                         if (bk->lb_param & LPF_FAILOUT)
1520                                 RETURN(PTR_ERR(target));
1521                         else
1522                                 goto checkpoint;
1523                 }
1524
1525                 /* XXX: In fact, low layer otable-based iteration should not
1526                  *      return agent object. But before LU-2646 resolved, we
1527                  *      need more processing for agent object. */
1528                 if (mdd_object_exists(target) && !mdd_object_remote(target))
1529                         rc = mdd_lfsck_exec_oit(env, lfsck, target);
1530                 mdd_object_put(env, target);
1531                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
1532                         RETURN(rc);
1533
1534 checkpoint:
1535                 rc = mdd_lfsck_checkpoint(env, lfsck, true);
1536                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
1537                         RETURN(rc);
1538
1539                 /* Rate control. */
1540                 mdd_lfsck_control_speed(lfsck);
1541
1542                 rc = iops->next(env, di);
1543                 if (rc > 0)
1544                         lfsck->ml_oit_over = 1;
1545
1546                 if (unlikely(!thread_is_running(thread)))
1547                         RETURN(0);
1548         } while (rc == 0 || lfsck->ml_di_dir != NULL);
1549
1550         RETURN(rc);
1551 }
1552
1553 static int mdd_lfsck_main(void *args)
1554 {
1555         struct lu_env            env;
1556         struct md_lfsck         *lfsck    = (struct md_lfsck *)args;
1557         struct ptlrpc_thread    *thread   = &lfsck->ml_thread;
1558         struct dt_object        *oit_obj  = lfsck->ml_obj_oit;
1559         const struct dt_it_ops  *oit_iops = &oit_obj->do_index_ops->dio_it;
1560         struct dt_it            *oit_di;
1561         int                      rc;
1562         ENTRY;
1563
1564         cfs_daemonize("lfsck");
1565         rc = lu_env_init(&env, LCT_MD_THREAD | LCT_DT_THREAD);
1566         if (rc != 0) {
1567                 CERROR("%s: LFSCK, fail to init env, rc = %d\n",
1568                        mdd_lfsck2name(lfsck), rc);
1569                 GOTO(noenv, rc);
1570         }
1571
1572         oit_di = oit_iops->init(&env, oit_obj, lfsck->ml_args_oit, BYPASS_CAPA);
1573         if (IS_ERR(oit_di)) {
1574                 rc = PTR_ERR(oit_di);
1575                 CERROR("%s: LFSCK, fail to init iteration, rc = %d\n",
1576                        mdd_lfsck2name(lfsck), rc);
1577                 GOTO(fini_env, rc);
1578         }
1579
1580         spin_lock(&lfsck->ml_lock);
1581         lfsck->ml_di_oit = oit_di;
1582         spin_unlock(&lfsck->ml_lock);
1583         rc = mdd_lfsck_prep(&env, lfsck);
1584         if (rc != 0)
1585                 GOTO(fini_oit, rc);
1586
1587         CDEBUG(D_LFSCK, "LFSCK entry: oit_flags = 0x%x, dir_flags = 0x%x, "
1588                "oit_cookie = "LPU64", dir_cookie = "LPU64", parent = "DFID
1589                ", pid = %d\n", lfsck->ml_args_oit, lfsck->ml_args_dir,
1590                lfsck->ml_pos_current.lp_oit_cookie,
1591                lfsck->ml_pos_current.lp_dir_cookie,
1592                PFID(&lfsck->ml_pos_current.lp_dir_parent),
1593                cfs_curproc_pid());
1594
1595         spin_lock(&lfsck->ml_lock);
1596         thread_set_flags(thread, SVC_RUNNING);
1597         spin_unlock(&lfsck->ml_lock);
1598         cfs_waitq_broadcast(&thread->t_ctl_waitq);
1599
1600         if (!cfs_list_empty(&lfsck->ml_list_scan) ||
1601             cfs_list_empty(&lfsck->ml_list_double_scan))
1602                 rc = mdd_lfsck_oit_engine(&env, lfsck);
1603         else
1604                 rc = 1;
1605
1606         CDEBUG(D_LFSCK, "LFSCK exit: oit_flags = 0x%x, dir_flags = 0x%x, "
1607                "oit_cookie = "LPU64", dir_cookie = "LPU64", parent = "DFID
1608                ", pid = %d, rc = %d\n", lfsck->ml_args_oit, lfsck->ml_args_dir,
1609                lfsck->ml_pos_current.lp_oit_cookie,
1610                lfsck->ml_pos_current.lp_dir_cookie,
1611                PFID(&lfsck->ml_pos_current.lp_dir_parent),
1612                cfs_curproc_pid(), rc);
1613
1614         if (lfsck->ml_paused && cfs_list_empty(&lfsck->ml_list_scan))
1615                 oit_iops->put(&env, oit_di);
1616
1617         rc = mdd_lfsck_post(&env, lfsck, rc);
1618         if (lfsck->ml_di_dir != NULL)
1619                 mdd_lfsck_close_dir(&env, lfsck);
1620
1621 fini_oit:
1622         spin_lock(&lfsck->ml_lock);
1623         lfsck->ml_di_oit = NULL;
1624         spin_unlock(&lfsck->ml_lock);
1625
1626         oit_iops->fini(&env, oit_di);
1627         if (rc == 1) {
1628                 if (!cfs_list_empty(&lfsck->ml_list_double_scan))
1629                         rc = mdd_lfsck_double_scan(&env, lfsck);
1630                 else
1631                         rc = 0;
1632         }
1633
1634         /* XXX: Purge the pinned objects in the future. */
1635
1636 fini_env:
1637         lu_env_fini(&env);
1638
1639 noenv:
1640         spin_lock(&lfsck->ml_lock);
1641         thread_set_flags(thread, SVC_STOPPED);
1642         cfs_waitq_broadcast(&thread->t_ctl_waitq);
1643         spin_unlock(&lfsck->ml_lock);
1644         return rc;
1645 }
1646
1647 /* external interfaces */
1648
1649 int mdd_lfsck_set_speed(const struct lu_env *env, struct md_lfsck *lfsck,
1650                         __u32 limit)
1651 {
1652         int rc;
1653
1654         mutex_lock(&lfsck->ml_mutex);
1655         __mdd_lfsck_set_speed(lfsck, limit);
1656         rc = mdd_lfsck_bookmark_store(env, lfsck);
1657         mutex_unlock(&lfsck->ml_mutex);
1658         return rc;
1659 }
1660
1661 int mdd_lfsck_start(const struct lu_env *env, struct md_lfsck *lfsck,
1662                     struct lfsck_start *start)
1663 {
1664         struct lfsck_bookmark  *bk     = &lfsck->ml_bookmark_ram;
1665         struct ptlrpc_thread   *thread = &lfsck->ml_thread;
1666         struct lfsck_component *com;
1667         struct l_wait_info      lwi    = { 0 };
1668         bool                    dirty  = false;
1669         int                     rc     = 0;
1670         __u16                   valid  = 0;
1671         __u16                   flags  = 0;
1672         ENTRY;
1673
1674         if (lfsck->ml_obj_oit == NULL)
1675                 RETURN(-ENOTSUPP);
1676
1677         /* start == NULL means auto trigger paused LFSCK. */
1678         if (start == NULL && cfs_list_empty(&lfsck->ml_list_scan))
1679                 RETURN(0);
1680
1681         mutex_lock(&lfsck->ml_mutex);
1682         spin_lock(&lfsck->ml_lock);
1683         if (!thread_is_init(thread) && !thread_is_stopped(thread)) {
1684                 spin_unlock(&lfsck->ml_lock);
1685                 mutex_unlock(&lfsck->ml_mutex);
1686                 RETURN(-EALREADY);
1687         }
1688
1689         spin_unlock(&lfsck->ml_lock);
1690
1691         lfsck->ml_paused = 0;
1692         lfsck->ml_oit_over = 0;
1693         lfsck->ml_drop_dryrun = 0;
1694         lfsck->ml_new_scanned = 0;
1695
1696         /* For auto trigger. */
1697         if (start == NULL)
1698                 goto trigger;
1699
1700         start->ls_version = bk->lb_version;
1701         if (start->ls_valid & LSV_SPEED_LIMIT) {
1702                 __mdd_lfsck_set_speed(lfsck, start->ls_speed_limit);
1703                 dirty = true;
1704         }
1705
1706         if (start->ls_valid & LSV_ERROR_HANDLE) {
1707                 valid |= DOIV_ERROR_HANDLE;
1708                 if (start->ls_flags & LPF_FAILOUT)
1709                         flags |= DOIF_FAILOUT;
1710
1711                 if ((start->ls_flags & LPF_FAILOUT) &&
1712                     !(bk->lb_param & LPF_FAILOUT)) {
1713                         bk->lb_param |= LPF_FAILOUT;
1714                         dirty = true;
1715                 } else if (!(start->ls_flags & LPF_FAILOUT) &&
1716                            (bk->lb_param & LPF_FAILOUT)) {
1717                         bk->lb_param &= ~LPF_FAILOUT;
1718                         dirty = true;
1719                 }
1720         }
1721
1722         if (start->ls_valid & LSV_DRYRUN) {
1723                 if ((start->ls_flags & LPF_DRYRUN) &&
1724                     !(bk->lb_param & LPF_DRYRUN)) {
1725                         bk->lb_param |= LPF_DRYRUN;
1726                         dirty = true;
1727                 } else if (!(start->ls_flags & LPF_DRYRUN) &&
1728                            (bk->lb_param & LPF_DRYRUN)) {
1729                         bk->lb_param &= ~LPF_DRYRUN;
1730                         lfsck->ml_drop_dryrun = 1;
1731                         dirty = true;
1732                 }
1733         }
1734
1735         if (dirty) {
1736                 rc = mdd_lfsck_bookmark_store(env, lfsck);
1737                 if (rc != 0)
1738                         GOTO(out, rc);
1739         }
1740
1741         if (start->ls_flags & LPF_RESET)
1742                 flags |= DOIF_RESET;
1743
1744         if (start->ls_active != 0) {
1745                 struct lfsck_component *next;
1746                 __u16 type = 1;
1747
1748                 if (start->ls_active == LFSCK_TYPES_ALL)
1749                         start->ls_active = LFSCK_TYPES_SUPPORTED;
1750
1751                 if (start->ls_active & ~LFSCK_TYPES_SUPPORTED) {
1752                         start->ls_active &= ~LFSCK_TYPES_SUPPORTED;
1753                         GOTO(out, rc = -ENOTSUPP);
1754                 }
1755
1756                 cfs_list_for_each_entry_safe(com, next,
1757                                              &lfsck->ml_list_scan, lc_link) {
1758                         if (!(com->lc_type & start->ls_active)) {
1759                                 rc = com->lc_ops->lfsck_post(env, com, 0);
1760                                 if (rc != 0)
1761                                         GOTO(out, rc);
1762                         }
1763                 }
1764
1765                 while (start->ls_active != 0) {
1766                         if (type & start->ls_active) {
1767                                 com = __mdd_lfsck_component_find(lfsck, type,
1768                                                         &lfsck->ml_list_idle);
1769                                 if (com != NULL) {
1770                                         /* The component status will be updated
1771                                          * when its prep() is called later by
1772                                          * the LFSCK main engine. */
1773                                         cfs_list_del_init(&com->lc_link);
1774                                         cfs_list_add_tail(&com->lc_link,
1775                                                           &lfsck->ml_list_scan);
1776                                 }
1777                                 start->ls_active &= ~type;
1778                         }
1779                         type <<= 1;
1780                 }
1781         }
1782
1783         cfs_list_for_each_entry(com, &lfsck->ml_list_scan, lc_link) {
1784                 start->ls_active |= com->lc_type;
1785                 if (flags & DOIF_RESET) {
1786                         rc = com->lc_ops->lfsck_reset(env, com, false);
1787                         if (rc != 0)
1788                                 GOTO(out, rc);
1789                 }
1790         }
1791
1792 trigger:
1793         lfsck->ml_args_dir = LUDA_64BITHASH | LUDA_VERIFY;
1794         if (bk->lb_param & LPF_DRYRUN)
1795                 lfsck->ml_args_dir |= LUDA_VERIFY_DRYRUN;
1796
1797         if (bk->lb_param & LPF_FAILOUT) {
1798                 valid |= DOIV_ERROR_HANDLE;
1799                 flags |= DOIF_FAILOUT;
1800         }
1801
1802         if (!cfs_list_empty(&lfsck->ml_list_scan))
1803                 flags |= DOIF_OUTUSED;
1804
1805         lfsck->ml_args_oit = (flags << DT_OTABLE_IT_FLAGS_SHIFT) | valid;
1806         thread_set_flags(thread, 0);
1807         rc = cfs_create_thread(mdd_lfsck_main, lfsck, 0);
1808         if (rc < 0)
1809                 CERROR("%s: cannot start LFSCK thread, rc = %d\n",
1810                        mdd_lfsck2name(lfsck), rc);
1811         else
1812                 l_wait_event(thread->t_ctl_waitq,
1813                              thread_is_running(thread) ||
1814                              thread_is_stopped(thread),
1815                              &lwi);
1816
1817         GOTO(out, rc = 0);
1818
1819 out:
1820         mutex_unlock(&lfsck->ml_mutex);
1821         return (rc < 0 ? rc : 0);
1822 }
1823
1824 int mdd_lfsck_stop(const struct lu_env *env, struct md_lfsck *lfsck,
1825                    bool pause)
1826 {
1827         struct ptlrpc_thread *thread = &lfsck->ml_thread;
1828         struct l_wait_info    lwi    = { 0 };
1829         ENTRY;
1830
1831         mutex_lock(&lfsck->ml_mutex);
1832         spin_lock(&lfsck->ml_lock);
1833         if (thread_is_init(thread) || thread_is_stopped(thread)) {
1834                 spin_unlock(&lfsck->ml_lock);
1835                 mutex_unlock(&lfsck->ml_mutex);
1836                 RETURN(-EALREADY);
1837         }
1838
1839         if (pause)
1840                 lfsck->ml_paused = 1;
1841         thread_set_flags(thread, SVC_STOPPING);
1842         /* The LFSCK thread may be sleeping on low layer wait queue,
1843          * wake it up. */
1844         if (likely(lfsck->ml_di_oit != NULL))
1845                 lfsck->ml_obj_oit->do_index_ops->dio_it.put(env,
1846                                                             lfsck->ml_di_oit);
1847         spin_unlock(&lfsck->ml_lock);
1848
1849         cfs_waitq_broadcast(&thread->t_ctl_waitq);
1850         l_wait_event(thread->t_ctl_waitq,
1851                      thread_is_stopped(thread),
1852                      &lwi);
1853         mutex_unlock(&lfsck->ml_mutex);
1854
1855         RETURN(0);
1856 }
1857
1858 static const struct lu_fid lfsck_it_fid = { .f_seq = FID_SEQ_LOCAL_FILE,
1859                                             .f_oid = OTABLE_IT_OID,
1860                                             .f_ver = 0 };
1861
1862 int mdd_lfsck_setup(const struct lu_env *env, struct mdd_device *mdd)
1863 {
1864         struct md_lfsck  *lfsck = &mdd->mdd_lfsck;
1865         struct dt_object *obj;
1866         int               rc;
1867         ENTRY;
1868
1869         LASSERT(!lfsck->ml_initialized);
1870
1871         lfsck->ml_initialized = 1;
1872         mutex_init(&lfsck->ml_mutex);
1873         spin_lock_init(&lfsck->ml_lock);
1874         CFS_INIT_LIST_HEAD(&lfsck->ml_list_scan);
1875         CFS_INIT_LIST_HEAD(&lfsck->ml_list_dir);
1876         CFS_INIT_LIST_HEAD(&lfsck->ml_list_double_scan);
1877         CFS_INIT_LIST_HEAD(&lfsck->ml_list_idle);
1878         cfs_waitq_init(&lfsck->ml_thread.t_ctl_waitq);
1879
1880         obj = dt_locate(env, mdd->mdd_bottom, &lfsck_it_fid);
1881         if (IS_ERR(obj))
1882                 RETURN(PTR_ERR(obj));
1883
1884         lfsck->ml_obj_oit = obj;
1885         rc = obj->do_ops->do_index_try(env, obj, &dt_otable_features);
1886         if (rc != 0) {
1887                 if (rc == -ENOTSUPP)
1888                         rc = 0;
1889
1890                 RETURN(rc);
1891         }
1892
1893         obj = dt_store_open(env, mdd->mdd_bottom, "", lfsck_bookmark_name,
1894                             &mdd_env_info(env)->mti_fid);
1895         if (IS_ERR(obj))
1896                 RETURN(PTR_ERR(obj));
1897
1898         lfsck->ml_bookmark_obj = obj;
1899         rc = mdd_lfsck_bookmark_load(env, lfsck);
1900         if (rc == -ENODATA)
1901                 rc = mdd_lfsck_bookmark_init(env, lfsck);
1902         if (rc != 0)
1903                 RETURN(rc);
1904
1905         rc = mdd_lfsck_namespace_setup(env, lfsck);
1906         /* XXX: LFSCK components initialization to be added here. */
1907
1908         RETURN(rc);
1909 }
1910
1911 void mdd_lfsck_cleanup(const struct lu_env *env, struct mdd_device *mdd)
1912 {
1913         struct md_lfsck         *lfsck  = &mdd->mdd_lfsck;
1914         struct ptlrpc_thread    *thread = &lfsck->ml_thread;
1915         struct lfsck_component  *com;
1916
1917         if (!lfsck->ml_initialized)
1918                 return;
1919
1920         LASSERT(thread_is_init(thread) || thread_is_stopped(thread));
1921
1922         if (lfsck->ml_obj_oit != NULL) {
1923                 lu_object_put(env, &lfsck->ml_obj_oit->do_lu);
1924                 lfsck->ml_obj_oit = NULL;
1925         }
1926
1927         LASSERT(lfsck->ml_obj_dir == NULL);
1928
1929         if (lfsck->ml_bookmark_obj != NULL) {
1930                 lu_object_put(env, &lfsck->ml_bookmark_obj->do_lu);
1931                 lfsck->ml_bookmark_obj = NULL;
1932         }
1933
1934         while (!cfs_list_empty(&lfsck->ml_list_scan)) {
1935                 com = cfs_list_entry(lfsck->ml_list_scan.next,
1936                                      struct lfsck_component,
1937                                      lc_link);
1938                 mdd_lfsck_component_cleanup(env, com);
1939         }
1940
1941         LASSERT(cfs_list_empty(&lfsck->ml_list_dir));
1942
1943         while (!cfs_list_empty(&lfsck->ml_list_double_scan)) {
1944                 com = cfs_list_entry(lfsck->ml_list_double_scan.next,
1945                                      struct lfsck_component,
1946                                      lc_link);
1947                 mdd_lfsck_component_cleanup(env, com);
1948         }
1949
1950         while (!cfs_list_empty(&lfsck->ml_list_idle)) {
1951                 com = cfs_list_entry(lfsck->ml_list_idle.next,
1952                                      struct lfsck_component,
1953                                      lc_link);
1954                 mdd_lfsck_component_cleanup(env, com);
1955         }
1956 }