lustre/lfsck/lfsck_namespace.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, 2013, Intel Corporation.
  24  */
  25 /*
  26  * lustre/lfsck/lfsck_namespace.c
  27  *
  28  * Author: Fan, Yong <fan.yong@intel.com>
  29  */
  30
  31 #define DEBUG_SUBSYSTEM S_LFSCK
  32
  33 #include <lustre/lustre_idl.h>
  34 #include <lu_object.h>
  35 #include <dt_object.h>
  36 #include <md_object.h>
  37 #include <lustre_linkea.h>
  38 #include <lustre_fid.h>
  39 #include <lustre_lib.h>
  40 #include <lustre_net.h>
  41 #include <lustre/lustre_user.h>
  42
  43 #include "lfsck_internal.h"
  44
  45 #define LFSCK_NAMESPACE_MAGIC   0xA0629D03
  46
  47 static const char lfsck_namespace_name[] = "lfsck_namespace";
  48
  49 static void lfsck_namespace_le_to_cpu(struct lfsck_namespace *des,
  50                                       struct lfsck_namespace *src)
  51 {
  52         des->ln_magic = le32_to_cpu(src->ln_magic);
  53         des->ln_status = le32_to_cpu(src->ln_status);
  54         des->ln_flags = le32_to_cpu(src->ln_flags);
  55         des->ln_success_count = le32_to_cpu(src->ln_success_count);
  56         des->ln_run_time_phase1 = le32_to_cpu(src->ln_run_time_phase1);
  57         des->ln_run_time_phase2 = le32_to_cpu(src->ln_run_time_phase2);
  58         des->ln_time_last_complete = le64_to_cpu(src->ln_time_last_complete);
  59         des->ln_time_latest_start = le64_to_cpu(src->ln_time_latest_start);
  60         des->ln_time_last_checkpoint =
  61                                 le64_to_cpu(src->ln_time_last_checkpoint);
  62         lfsck_position_le_to_cpu(&des->ln_pos_latest_start,
  63                                  &src->ln_pos_latest_start);
  64         lfsck_position_le_to_cpu(&des->ln_pos_last_checkpoint,
  65                                  &src->ln_pos_last_checkpoint);
  66         lfsck_position_le_to_cpu(&des->ln_pos_first_inconsistent,
  67                                  &src->ln_pos_first_inconsistent);
  68         des->ln_items_checked = le64_to_cpu(src->ln_items_checked);
  69         des->ln_items_repaired = le64_to_cpu(src->ln_items_repaired);
  70         des->ln_items_failed = le64_to_cpu(src->ln_items_failed);
  71         des->ln_dirs_checked = le64_to_cpu(src->ln_dirs_checked);
  72         des->ln_mlinked_checked = le64_to_cpu(src->ln_mlinked_checked);
  73         des->ln_objs_checked_phase2 = le64_to_cpu(src->ln_objs_checked_phase2);
  74         des->ln_objs_repaired_phase2 =
  75                                 le64_to_cpu(src->ln_objs_repaired_phase2);
  76         des->ln_objs_failed_phase2 = le64_to_cpu(src->ln_objs_failed_phase2);
  77         des->ln_objs_nlink_repaired = le64_to_cpu(src->ln_objs_nlink_repaired);
  78         des->ln_objs_lost_found = le64_to_cpu(src->ln_objs_lost_found);
  79         fid_le_to_cpu(&des->ln_fid_latest_scanned_phase2,
  80                       &src->ln_fid_latest_scanned_phase2);
  81 }
  82
  83 static void lfsck_namespace_cpu_to_le(struct lfsck_namespace *des,
  84                                       struct lfsck_namespace *src)
  85 {
  86         des->ln_magic = cpu_to_le32(src->ln_magic);
  87         des->ln_status = cpu_to_le32(src->ln_status);
  88         des->ln_flags = cpu_to_le32(src->ln_flags);
  89         des->ln_success_count = cpu_to_le32(src->ln_success_count);
  90         des->ln_run_time_phase1 = cpu_to_le32(src->ln_run_time_phase1);
  91         des->ln_run_time_phase2 = cpu_to_le32(src->ln_run_time_phase2);
  92         des->ln_time_last_complete = cpu_to_le64(src->ln_time_last_complete);
  93         des->ln_time_latest_start = cpu_to_le64(src->ln_time_latest_start);
  94         des->ln_time_last_checkpoint =
  95                                 cpu_to_le64(src->ln_time_last_checkpoint);
  96         lfsck_position_cpu_to_le(&des->ln_pos_latest_start,
  97                                  &src->ln_pos_latest_start);
  98         lfsck_position_cpu_to_le(&des->ln_pos_last_checkpoint,
  99                                  &src->ln_pos_last_checkpoint);
 100         lfsck_position_cpu_to_le(&des->ln_pos_first_inconsistent,
 101                                  &src->ln_pos_first_inconsistent);
 102         des->ln_items_checked = cpu_to_le64(src->ln_items_checked);
 103         des->ln_items_repaired = cpu_to_le64(src->ln_items_repaired);
 104         des->ln_items_failed = cpu_to_le64(src->ln_items_failed);
 105         des->ln_dirs_checked = cpu_to_le64(src->ln_dirs_checked);
 106         des->ln_mlinked_checked = cpu_to_le64(src->ln_mlinked_checked);
 107         des->ln_objs_checked_phase2 = cpu_to_le64(src->ln_objs_checked_phase2);
 108         des->ln_objs_repaired_phase2 =
 109                                 cpu_to_le64(src->ln_objs_repaired_phase2);
 110         des->ln_objs_failed_phase2 = cpu_to_le64(src->ln_objs_failed_phase2);
 111         des->ln_objs_nlink_repaired = cpu_to_le64(src->ln_objs_nlink_repaired);
 112         des->ln_objs_lost_found = cpu_to_le64(src->ln_objs_lost_found);
 113         fid_cpu_to_le(&des->ln_fid_latest_scanned_phase2,
 114                       &src->ln_fid_latest_scanned_phase2);
 115 }
 116
 117 /**
 118  * Load namespace LFSCK statistics information from the trace file.
 119  *
 120  * \param[in] env       pointer to the thread context
 121  * \param[in] com       pointer to the lfsck component
 122  *
 123  * \retval              0 for success
 124  * \retval              negative error number on failure or absence the
 125  *                      namespace LFSCK trace file
 126  */
 127 static int lfsck_namespace_load(const struct lu_env *env,
 128                                 struct lfsck_component *com)
 129 {
 130         int len = com->lc_file_size;
 131         int rc;
 132
 133         rc = dt_xattr_get(env, com->lc_obj,
 134                           lfsck_buf_get(env, com->lc_file_disk, len),
 135                           XATTR_NAME_LFSCK_NAMESPACE, BYPASS_CAPA);
 136         if (rc == len) {
 137                 struct lfsck_namespace *ns = com->lc_file_ram;
 138
 139                 lfsck_namespace_le_to_cpu(ns,
 140                                 (struct lfsck_namespace *)com->lc_file_disk);
 141                 if (ns->ln_magic != LFSCK_NAMESPACE_MAGIC) {
 142                         CWARN("%.16s: invalid lfsck_namespace magic "
 143                               "0x%x != 0x%x\n",
 144                               lfsck_lfsck2name(com->lc_lfsck),
 145                               ns->ln_magic, LFSCK_NAMESPACE_MAGIC);
 146                         rc = -ESTALE;
 147                 } else {
 148                         rc = 0;
 149                 }
 150         } else {
 151                 if (rc != -ENODATA)
 152                         CERROR("%.16s: fail to load lfsck_namespace, "
 153                                "expected = %d, rc = %d\n",
 154                                lfsck_lfsck2name(com->lc_lfsck), len, rc);
 155                 else if (rc > 0)
 156                         rc = -ESTALE;
 157         }
 158
 159         return rc;
 160 }
 161
 162 static int lfsck_namespace_store(const struct lu_env *env,
 163                                  struct lfsck_component *com)
 164 {
 165         struct dt_object        *obj    = com->lc_obj;
 166         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 167         struct thandle          *handle;
 168         int                      len    = com->lc_file_size;
 169         int                      rc;
 170         ENTRY;
 171
 172         lfsck_namespace_cpu_to_le((struct lfsck_namespace *)com->lc_file_disk,
 173                                   (struct lfsck_namespace *)com->lc_file_ram);
 174         handle = dt_trans_create(env, lfsck->li_bottom);
 175         if (IS_ERR(handle)) {
 176                 rc = PTR_ERR(handle);
 177                 CERROR("%.16s: fail to create trans for storing "
 178                        "lfsck_namespace: %d\n,", lfsck_lfsck2name(lfsck), rc);
 179                 RETURN(rc);
 180         }
 181
 182         rc = dt_declare_xattr_set(env, obj,
 183                                   lfsck_buf_get(env, com->lc_file_disk, len),
 184                                   XATTR_NAME_LFSCK_NAMESPACE, 0, handle);
 185         if (rc != 0) {
 186                 CERROR("%.16s: fail to declare trans for storing "
 187                        "lfsck_namespace: %d\n,", lfsck_lfsck2name(lfsck), rc);
 188                 GOTO(out, rc);
 189         }
 190
 191         rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
 192         if (rc != 0) {
 193                 CERROR("%.16s: fail to start trans for storing "
 194                        "lfsck_namespace: %d\n,", lfsck_lfsck2name(lfsck), rc);
 195                 GOTO(out, rc);
 196         }
 197
 198         rc = dt_xattr_set(env, obj,
 199                           lfsck_buf_get(env, com->lc_file_disk, len),
 200                           XATTR_NAME_LFSCK_NAMESPACE, 0,
 201                           handle, BYPASS_CAPA);
 202         if (rc != 0)
 203                 CERROR("%.16s: fail to store lfsck_namespace, len = %d, "
 204                        "rc = %d\n", lfsck_lfsck2name(lfsck), len, rc);
 205
 206         GOTO(out, rc);
 207
 208 out:
 209         dt_trans_stop(env, lfsck->li_bottom, handle);
 210         return rc;
 211 }
 212
 213 static int lfsck_namespace_init(const struct lu_env *env,
 214                                 struct lfsck_component *com)
 215 {
 216         struct lfsck_namespace *ns = (struct lfsck_namespace *)com->lc_file_ram;
 217         int rc;
 218
 219         memset(ns, 0, sizeof(*ns));
 220         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
 221         ns->ln_status = LS_INIT;
 222         down_write(&com->lc_sem);
 223         rc = lfsck_namespace_store(env, com);
 224         up_write(&com->lc_sem);
 225         return rc;
 226 }
 227
 228 static int lfsck_namespace_lookup(const struct lu_env *env,
 229                                   struct lfsck_component *com,
 230                                   const struct lu_fid *fid, __u8 *flags)
 231 {
 232         struct lu_fid *key = &lfsck_env_info(env)->lti_fid;
 233         int            rc;
 234
 235         fid_cpu_to_be(key, fid);
 236         rc = dt_lookup(env, com->lc_obj, (struct dt_rec *)flags,
 237                        (const struct dt_key *)key, BYPASS_CAPA);
 238         return rc;
 239 }
 240
 241 static int lfsck_namespace_delete(const struct lu_env *env,
 242                                   struct lfsck_component *com,
 243                                   const struct lu_fid *fid)
 244 {
 245         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 246         struct lu_fid           *key    = &lfsck_env_info(env)->lti_fid;
 247         struct thandle          *handle;
 248         struct dt_object        *obj    = com->lc_obj;
 249         int                      rc;
 250         ENTRY;
 251
 252         handle = dt_trans_create(env, lfsck->li_bottom);
 253         if (IS_ERR(handle))
 254                 RETURN(PTR_ERR(handle));
 255
 256         rc = dt_declare_delete(env, obj, (const struct dt_key *)fid, handle);
 257         if (rc != 0)
 258                 GOTO(out, rc);
 259
 260         rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
 261         if (rc != 0)
 262                 GOTO(out, rc);
 263
 264         fid_cpu_to_be(key, fid);
 265         rc = dt_delete(env, obj, (const struct dt_key *)key, handle,
 266                        BYPASS_CAPA);
 267
 268         GOTO(out, rc);
 269
 270 out:
 271         dt_trans_stop(env, lfsck->li_bottom, handle);
 272         return rc;
 273 }
 274
 275 static int lfsck_namespace_update(const struct lu_env *env,
 276                                   struct lfsck_component *com,
 277                                   const struct lu_fid *fid,
 278                                   __u8 flags, bool force)
 279 {
 280         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 281         struct lu_fid           *key    = &lfsck_env_info(env)->lti_fid;
 282         struct thandle          *handle;
 283         struct dt_object        *obj    = com->lc_obj;
 284         int                      rc;
 285         bool                     exist  = false;
 286         __u8                     tf;
 287         ENTRY;
 288
 289         rc = lfsck_namespace_lookup(env, com, fid, &tf);
 290         if (rc != 0 && rc != -ENOENT)
 291                 RETURN(rc);
 292
 293         if (rc == 0) {
 294                 if (!force || flags == tf)
 295                         RETURN(0);
 296
 297                 exist = true;
 298                 handle = dt_trans_create(env, lfsck->li_bottom);
 299                 if (IS_ERR(handle))
 300                         RETURN(PTR_ERR(handle));
 301
 302                 rc = dt_declare_delete(env, obj, (const struct dt_key *)fid,
 303                                        handle);
 304                 if (rc != 0)
 305                         GOTO(out, rc);
 306         } else {
 307                 handle = dt_trans_create(env, lfsck->li_bottom);
 308                 if (IS_ERR(handle))
 309                         RETURN(PTR_ERR(handle));
 310         }
 311
 312         rc = dt_declare_insert(env, obj, (const struct dt_rec *)&flags,
 313                                (const struct dt_key *)fid, handle);
 314         if (rc != 0)
 315                 GOTO(out, rc);
 316
 317         rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
 318         if (rc != 0)
 319                 GOTO(out, rc);
 320
 321         fid_cpu_to_be(key, fid);
 322         if (exist) {
 323                 rc = dt_delete(env, obj, (const struct dt_key *)key, handle,
 324                                BYPASS_CAPA);
 325                 if (rc != 0) {
 326                         CERROR("%s: fail to insert "DFID", rc = %d\n",
 327                                lfsck_lfsck2name(com->lc_lfsck), PFID(fid), rc);
 328                         GOTO(out, rc);
 329                 }
 330         }
 331
 332         rc = dt_insert(env, obj, (const struct dt_rec *)&flags,
 333                        (const struct dt_key *)key, handle, BYPASS_CAPA, 1);
 334
 335         GOTO(out, rc);
 336
 337 out:
 338         dt_trans_stop(env, lfsck->li_bottom, handle);
 339         return rc;
 340 }
 341
 342 static int lfsck_namespace_check_exist(const struct lu_env *env,
 343                                        struct lfsck_instance *lfsck,
 344                                        struct dt_object *obj, const char *name)
 345 {
 346         struct dt_object *dir = lfsck->li_obj_dir;
 347         struct lu_fid    *fid = &lfsck_env_info(env)->lti_fid;
 348         int               rc;
 349         ENTRY;
 350
 351         if (unlikely(lfsck_is_dead_obj(obj)))
 352                 RETURN(LFSCK_NAMEENTRY_DEAD);
 353
 354         rc = dt_lookup(env, dir, (struct dt_rec *)fid,
 355                        (const struct dt_key *)name, BYPASS_CAPA);
 356         if (rc == -ENOENT)
 357                 RETURN(LFSCK_NAMEENTRY_REMOVED);
 358
 359         if (rc < 0)
 360                 RETURN(rc);
 361
 362         if (!lu_fid_eq(fid, lfsck_dto2fid(obj)))
 363                 RETURN(LFSCK_NAMEENTRY_RECREATED);
 364
 365         RETURN(0);
 366 }
 367
 368 static int lfsck_declare_namespace_exec_dir(const struct lu_env *env,
 369                                             struct dt_object *obj,
 370                                             struct thandle *handle)
 371 {
 372         int rc;
 373
 374         /* For destroying all invalid linkEA entries. */
 375         rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, handle);
 376         if (rc != 0)
 377                 return rc;
 378
 379         /* For insert new linkEA entry. */
 380         rc = dt_declare_xattr_set(env, obj,
 381                         lfsck_buf_get_const(env, NULL, DEFAULT_LINKEA_SIZE),
 382                         XATTR_NAME_LINK, 0, handle);
 383         return rc;
 384 }
 385
 386 static int lfsck_links_read(const struct lu_env *env, struct dt_object *obj,
 387                             struct linkea_data *ldata)
 388 {
 389         int rc;
 390
 391         ldata->ld_buf =
 392                 lu_buf_check_and_alloc(&lfsck_env_info(env)->lti_linkea_buf,
 393                                        PAGE_CACHE_SIZE);
 394         if (ldata->ld_buf->lb_buf == NULL)
 395                 return -ENOMEM;
 396
 397         if (!dt_object_exists(obj))
 398                 return -ENODATA;
 399
 400         rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK, BYPASS_CAPA);
 401         if (rc == -ERANGE) {
 402                 /* Buf was too small, figure out what we need. */
 403                 lu_buf_free(ldata->ld_buf);
 404                 rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK,
 405                                   BYPASS_CAPA);
 406                 if (rc < 0)
 407                         return rc;
 408
 409                 ldata->ld_buf = lu_buf_check_and_alloc(ldata->ld_buf, rc);
 410                 if (ldata->ld_buf->lb_buf == NULL)
 411                         return -ENOMEM;
 412
 413                 rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK,
 414                                   BYPASS_CAPA);
 415         }
 416         if (rc < 0)
 417                 return rc;
 418
 419         linkea_init(ldata);
 420
 421         return 0;
 422 }
 423
 424 static int lfsck_links_write(const struct lu_env *env, struct dt_object *obj,
 425                              struct linkea_data *ldata, struct thandle *handle)
 426 {
 427         const struct lu_buf *buf = lfsck_buf_get_const(env,
 428                                                        ldata->ld_buf->lb_buf,
 429                                                        ldata->ld_leh->leh_len);
 430
 431         return dt_xattr_set(env, obj, buf, XATTR_NAME_LINK, 0, handle,
 432                             BYPASS_CAPA);
 433 }
 434
 435 /**
 436  * \retval ve: removed entries
 437  */
 438 static int lfsck_linkea_entry_unpack(struct lfsck_instance *lfsck,
 439                                      struct linkea_data *ldata,
 440                                      struct lu_name *cname,
 441                                      struct lu_fid *pfid)
 442 {
 443         struct link_ea_entry    *oldlee;
 444         int                      oldlen;
 445         int                      removed = 0;
 446
 447         linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, cname, pfid);
 448         oldlee = ldata->ld_lee;
 449         oldlen = ldata->ld_reclen;
 450         linkea_next_entry(ldata);
 451         while (ldata->ld_lee != NULL) {
 452                 ldata->ld_reclen = (ldata->ld_lee->lee_reclen[0] << 8) |
 453                                    ldata->ld_lee->lee_reclen[1];
 454                 if (unlikely(ldata->ld_reclen == oldlen &&
 455                              memcmp(ldata->ld_lee, oldlee, oldlen) == 0)) {
 456                         linkea_del_buf(ldata, cname);
 457                         removed++;
 458                 } else {
 459                         linkea_next_entry(ldata);
 460                 }
 461         }
 462         ldata->ld_lee = oldlee;
 463         ldata->ld_reclen = oldlen;
 464         return removed;
 465 }
 466
 467 /**
 468  * \retval +ve  repaired
 469  * \retval 0    no need to repair
 470  * \retval -ve  error cases
 471  */
 472 static int lfsck_namespace_double_scan_one(const struct lu_env *env,
 473                                            struct lfsck_component *com,
 474                                            struct dt_object *child, __u8 flags)
 475 {
 476         struct lfsck_thread_info *info    = lfsck_env_info(env);
 477         struct lu_attr           *la      = &info->lti_la;
 478         struct lu_name           *cname   = &info->lti_name;
 479         struct lu_fid            *pfid    = &info->lti_fid;
 480         struct lu_fid            *cfid    = &info->lti_fid2;
 481         struct lfsck_instance   *lfsck    = com->lc_lfsck;
 482         struct lfsck_bookmark   *bk       = &lfsck->li_bookmark_ram;
 483         struct lfsck_namespace  *ns       =
 484                                 (struct lfsck_namespace *)com->lc_file_ram;
 485         struct linkea_data       ldata    = { 0 };
 486         struct thandle          *handle   = NULL;
 487         bool                     locked   = false;
 488         bool                     update   = false;
 489         int                      rc;
 490         ENTRY;
 491
 492         if (com->lc_journal) {
 493
 494 again:
 495                 LASSERT(!locked);
 496
 497                 update = false;
 498                 com->lc_journal = 1;
 499                 handle = dt_trans_create(env, lfsck->li_next);
 500                 if (IS_ERR(handle))
 501                         RETURN(rc = PTR_ERR(handle));
 502
 503                 rc = dt_declare_xattr_set(env, child,
 504                         lfsck_buf_get_const(env, NULL, DEFAULT_LINKEA_SIZE),
 505                         XATTR_NAME_LINK, 0, handle);
 506                 if (rc != 0)
 507                         GOTO(stop, rc);
 508
 509                 rc = dt_trans_start(env, lfsck->li_next, handle);
 510                 if (rc != 0)
 511                         GOTO(stop, rc);
 512
 513                 dt_write_lock(env, child, MOR_TGT_CHILD);
 514                 locked = true;
 515         }
 516
 517         if (unlikely(lfsck_is_dead_obj(child)))
 518                 GOTO(stop, rc = 0);
 519
 520         rc = dt_attr_get(env, child, la, BYPASS_CAPA);
 521         if (rc == 0)
 522                 rc = lfsck_links_read(env, child, &ldata);
 523         if (rc != 0) {
 524                 if ((bk->lb_param & LPF_DRYRUN) &&
 525                     (rc == -EINVAL || rc == -ENODATA))
 526                         rc = 1;
 527
 528                 GOTO(stop, rc);
 529         }
 530
 531         linkea_first_entry(&ldata);
 532         while (ldata.ld_lee != NULL) {
 533                 struct dt_object *parent = NULL;
 534
 535                 rc = lfsck_linkea_entry_unpack(lfsck, &ldata, cname, pfid);
 536                 if (rc > 0)
 537                         update = true;
 538
 539                 if (!fid_is_sane(pfid))
 540                         goto shrink;
 541
 542                 parent = lfsck_object_find(env, lfsck, pfid);
 543                 if (parent == NULL)
 544                         goto shrink;
 545                 else if (IS_ERR(parent))
 546                         GOTO(stop, rc = PTR_ERR(parent));
 547
 548                 if (!dt_object_exists(parent))
 549                         goto shrink;
 550
 551                 /* XXX: Currently, skip remote object, the consistency for
 552                  *      remote object will be processed in LFSCK phase III. */
 553                 if (dt_object_remote(parent)) {
 554                         lfsck_object_put(env, parent);
 555                         linkea_next_entry(&ldata);
 556                         continue;
 557                 }
 558
 559                 if (unlikely(!dt_try_as_dir(env, parent)))
 560                         goto shrink;
 561
 562                 /* To guarantee the 'name' is terminated with '0'. */
 563                 memcpy(info->lti_key, cname->ln_name, cname->ln_namelen);
 564                 info->lti_key[cname->ln_namelen] = 0;
 565                 cname->ln_name = info->lti_key;
 566                 rc = dt_lookup(env, parent, (struct dt_rec *)cfid,
 567                                (const struct dt_key *)cname->ln_name,
 568                                BYPASS_CAPA);
 569                 if (rc != 0 && rc != -ENOENT) {
 570                         lfsck_object_put(env, parent);
 571                         GOTO(stop, rc);
 572                 }
 573
 574                 if (rc == 0) {
 575                         if (lu_fid_eq(cfid, lfsck_dto2fid(child))) {
 576                                 lfsck_object_put(env, parent);
 577                                 linkea_next_entry(&ldata);
 578                                 continue;
 579                         }
 580
 581                         goto shrink;
 582                 }
 583
 584                 /* If there is no name entry in the parent dir and the object
 585                  * link count is less than the linkea entries count, then the
 586                  * linkea entry should be removed. */
 587                 if (ldata.ld_leh->leh_reccount > la->la_nlink)
 588                         goto shrink;
 589
 590                 /* XXX: For the case of there is a linkea entry, but without
 591                  *      name entry pointing to the object and its hard links
 592                  *      count is not less than the object name entries count,
 593                  *      then seems we should add the 'missed' name entry back
 594                  *      to namespace, but before LFSCK phase III finished, we
 595                  *      do not know whether the object has some inconsistency
 596                  *      on other MDTs. So now, do NOT add the name entry back
 597                  *      to the namespace, but keep the linkEA entry. LU-2914 */
 598                 lfsck_object_put(env, parent);
 599                 linkea_next_entry(&ldata);
 600                 continue;
 601
 602 shrink:
 603                 if (parent != NULL)
 604                         lfsck_object_put(env, parent);
 605                 if (bk->lb_param & LPF_DRYRUN)
 606                         RETURN(1);
 607
 608                 CDEBUG(D_LFSCK, "Remove linkEA: "DFID"[%.*s], "DFID"\n",
 609                        PFID(lfsck_dto2fid(child)), cname->ln_namelen, cname->ln_name,
 610                        PFID(pfid));
 611                 linkea_del_buf(&ldata, cname);
 612                 update = true;
 613         }
 614
 615         if (update) {
 616                 if (!com->lc_journal) {
 617                         com->lc_journal = 1;
 618                         goto again;
 619                 }
 620
 621                 rc = lfsck_links_write(env, child, &ldata, handle);
 622         }
 623
 624         GOTO(stop, rc);
 625
 626 stop:
 627         if (locked) {
 628         /* XXX: For the case linkea entries count does not match the object hard
 629          *      links count, we cannot update the later one simply. Before LFSCK
 630          *      phase III finished, we cannot know whether there are some remote
 631          *      name entries to be repaired or not. LU-2914 */
 632                 if (rc == 0 && !lfsck_is_dead_obj(child) &&
 633                     ldata.ld_leh != NULL &&
 634                     ldata.ld_leh->leh_reccount != la->la_nlink)
 635                         CWARN("%.16s: the object "DFID" linkEA entry count %u "
 636                               "may not match its hardlink count %u\n",
 637                               lfsck_lfsck2name(lfsck), PFID(cfid),
 638                               ldata.ld_leh->leh_reccount, la->la_nlink);
 639
 640                 dt_write_unlock(env, child);
 641         }
 642
 643         if (handle != NULL)
 644                 dt_trans_stop(env, lfsck->li_next, handle);
 645
 646         if (rc == 0 && update) {
 647                 ns->ln_objs_nlink_repaired++;
 648                 rc = 1;
 649         }
 650
 651         return rc;
 652 }
 653
 654 /* namespace APIs */
 655
 656 static int lfsck_namespace_reset(const struct lu_env *env,
 657                                  struct lfsck_component *com, bool init)
 658 {
 659         struct lfsck_instance   *lfsck = com->lc_lfsck;
 660         struct lfsck_namespace  *ns    =
 661                                 (struct lfsck_namespace *)com->lc_file_ram;
 662         struct dt_object        *root;
 663         struct dt_object        *dto;
 664         int                      rc;
 665         ENTRY;
 666
 667         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
 668         if (IS_ERR(root))
 669                 RETURN(PTR_ERR(root));
 670
 671         dt_try_as_dir(env, root);
 672
 673         down_write(&com->lc_sem);
 674         if (init) {
 675                 memset(ns, 0, sizeof(*ns));
 676         } else {
 677                 __u32 count = ns->ln_success_count;
 678                 __u64 last_time = ns->ln_time_last_complete;
 679
 680                 memset(ns, 0, sizeof(*ns));
 681                 ns->ln_success_count = count;
 682                 ns->ln_time_last_complete = last_time;
 683         }
 684         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
 685         ns->ln_status = LS_INIT;
 686
 687         rc = local_object_unlink(env, lfsck->li_bottom, root,
 688                                  lfsck_namespace_name);
 689         if (rc != 0)
 690                 GOTO(out, rc);
 691
 692         lfsck_object_put(env, com->lc_obj);
 693         com->lc_obj = NULL;
 694         dto = local_index_find_or_create(env, lfsck->li_los, root,
 695                                          lfsck_namespace_name,
 696                                          S_IFREG | S_IRUGO | S_IWUSR,
 697                                          &dt_lfsck_features);
 698         if (IS_ERR(dto))
 699                 GOTO(out, rc = PTR_ERR(dto));
 700
 701         com->lc_obj = dto;
 702         rc = dto->do_ops->do_index_try(env, dto, &dt_lfsck_features);
 703         if (rc != 0)
 704                 GOTO(out, rc);
 705
 706         rc = lfsck_namespace_store(env, com);
 707
 708         GOTO(out, rc);
 709
 710 out:
 711         up_write(&com->lc_sem);
 712         lu_object_put(env, &root->do_lu);
 713         return rc;
 714 }
 715
 716 static void
 717 lfsck_namespace_fail(const struct lu_env *env, struct lfsck_component *com,
 718                      bool new_checked)
 719 {
 720         struct lfsck_namespace *ns = (struct lfsck_namespace *)com->lc_file_ram;
 721
 722         down_write(&com->lc_sem);
 723         if (new_checked)
 724                 com->lc_new_checked++;
 725         ns->ln_items_failed++;
 726         if (lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent))
 727                 lfsck_pos_fill(env, com->lc_lfsck,
 728                                &ns->ln_pos_first_inconsistent, false);
 729         up_write(&com->lc_sem);
 730 }
 731
 732 static int lfsck_namespace_checkpoint(const struct lu_env *env,
 733                                       struct lfsck_component *com, bool init)
 734 {
 735         struct lfsck_instance   *lfsck = com->lc_lfsck;
 736         struct lfsck_namespace  *ns    =
 737                                 (struct lfsck_namespace *)com->lc_file_ram;
 738         int                      rc;
 739
 740         if (com->lc_new_checked == 0 && !init)
 741                 return 0;
 742
 743         down_write(&com->lc_sem);
 744
 745         if (init) {
 746                 ns->ln_pos_latest_start = lfsck->li_pos_current;
 747         } else {
 748                 ns->ln_pos_last_checkpoint = lfsck->li_pos_current;
 749                 ns->ln_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
 750                                 HALF_SEC - lfsck->li_time_last_checkpoint);
 751                 ns->ln_time_last_checkpoint = cfs_time_current_sec();
 752                 ns->ln_items_checked += com->lc_new_checked;
 753                 com->lc_new_checked = 0;
 754         }
 755
 756         rc = lfsck_namespace_store(env, com);
 757
 758         up_write(&com->lc_sem);
 759         return rc;
 760 }
 761
 762 static int lfsck_namespace_prep(const struct lu_env *env,
 763                                 struct lfsck_component *com)
 764 {
 765         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 766         struct lfsck_namespace  *ns     =
 767                                 (struct lfsck_namespace *)com->lc_file_ram;
 768         struct lfsck_position   *pos    = &com->lc_pos_start;
 769
 770         if (ns->ln_status == LS_COMPLETED) {
 771                 int rc;
 772
 773                 rc = lfsck_namespace_reset(env, com, false);
 774                 if (rc != 0)
 775                         return rc;
 776         }
 777
 778         down_write(&com->lc_sem);
 779
 780         ns->ln_time_latest_start = cfs_time_current_sec();
 781
 782         spin_lock(&lfsck->li_lock);
 783         if (ns->ln_flags & LF_SCANNED_ONCE) {
 784                 if (!lfsck->li_drop_dryrun ||
 785                     lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent)) {
 786                         ns->ln_status = LS_SCANNING_PHASE2;
 787                         cfs_list_del_init(&com->lc_link);
 788                         cfs_list_add_tail(&com->lc_link,
 789                                           &lfsck->li_list_double_scan);
 790                         if (!cfs_list_empty(&com->lc_link_dir))
 791                                 cfs_list_del_init(&com->lc_link_dir);
 792                         lfsck_pos_set_zero(pos);
 793                 } else {
 794                         ns->ln_status = LS_SCANNING_PHASE1;
 795                         ns->ln_run_time_phase1 = 0;
 796                         ns->ln_run_time_phase2 = 0;
 797                         ns->ln_items_checked = 0;
 798                         ns->ln_items_repaired = 0;
 799                         ns->ln_items_failed = 0;
 800                         ns->ln_dirs_checked = 0;
 801                         ns->ln_mlinked_checked = 0;
 802                         ns->ln_objs_checked_phase2 = 0;
 803                         ns->ln_objs_repaired_phase2 = 0;
 804                         ns->ln_objs_failed_phase2 = 0;
 805                         ns->ln_objs_nlink_repaired = 0;
 806                         ns->ln_objs_lost_found = 0;
 807                         fid_zero(&ns->ln_fid_latest_scanned_phase2);
 808                         if (cfs_list_empty(&com->lc_link_dir))
 809                                 cfs_list_add_tail(&com->lc_link_dir,
 810                                                   &lfsck->li_list_dir);
 811                         *pos = ns->ln_pos_first_inconsistent;
 812                 }
 813         } else {
 814                 ns->ln_status = LS_SCANNING_PHASE1;
 815                 if (cfs_list_empty(&com->lc_link_dir))
 816                         cfs_list_add_tail(&com->lc_link_dir,
 817                                           &lfsck->li_list_dir);
 818                 if (!lfsck->li_drop_dryrun ||
 819                     lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent)) {
 820                         *pos = ns->ln_pos_last_checkpoint;
 821                         pos->lp_oit_cookie++;
 822                 } else {
 823                         *pos = ns->ln_pos_first_inconsistent;
 824                 }
 825         }
 826         spin_unlock(&lfsck->li_lock);
 827
 828         up_write(&com->lc_sem);
 829         return 0;
 830 }
 831
 832 static int lfsck_namespace_exec_oit(const struct lu_env *env,
 833                                     struct lfsck_component *com,
 834                                     struct dt_object *obj)
 835 {
 836         down_write(&com->lc_sem);
 837         com->lc_new_checked++;
 838         if (S_ISDIR(lfsck_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 static int lfsck_namespace_exec_dir(const struct lu_env *env,
 845                                     struct lfsck_component *com,
 846                                     struct dt_object *obj,
 847                                     struct lu_dirent *ent)
 848 {
 849         struct lfsck_thread_info   *info     = lfsck_env_info(env);
 850         struct lu_attr             *la       = &info->lti_la;
 851         struct lfsck_instance      *lfsck    = com->lc_lfsck;
 852         struct lfsck_bookmark      *bk       = &lfsck->li_bookmark_ram;
 853         struct lfsck_namespace     *ns       =
 854                                 (struct lfsck_namespace *)com->lc_file_ram;
 855         struct linkea_data          ldata    = { 0 };
 856         const struct lu_fid        *pfid     =
 857                                 lu_object_fid(&lfsck->li_obj_dir->do_lu);
 858         const struct lu_fid        *cfid     = lfsck_dto2fid(obj);
 859         const struct lu_name       *cname;
 860         struct thandle             *handle   = NULL;
 861         bool                        repaired = false;
 862         bool                        locked   = false;
 863         bool                        remove;
 864         bool                        newdata;
 865         int                         count    = 0;
 866         int                         rc;
 867         ENTRY;
 868
 869         cname = lfsck_name_get_const(env, ent->lde_name, ent->lde_namelen);
 870         down_write(&com->lc_sem);
 871         com->lc_new_checked++;
 872
 873         if (ent->lde_attrs & LUDA_UPGRADE) {
 874                 ns->ln_flags |= LF_UPGRADE;
 875                 repaired = true;
 876         } else if (ent->lde_attrs & LUDA_REPAIR) {
 877                 ns->ln_flags |= LF_INCONSISTENT;
 878                 repaired = true;
 879         }
 880
 881         if (ent->lde_name[0] == '.' &&
 882             (ent->lde_namelen == 1 ||
 883              (ent->lde_namelen == 2 && ent->lde_name[1] == '.') ||
 884              fid_is_dot_lustre(&ent->lde_fid)))
 885                 GOTO(out, rc = 0);
 886
 887         if (!(bk->lb_param & LPF_DRYRUN) &&
 888             (com->lc_journal || repaired)) {
 889
 890 again:
 891                 LASSERT(!locked);
 892
 893                 com->lc_journal = 1;
 894                 handle = dt_trans_create(env, lfsck->li_next);
 895                 if (IS_ERR(handle))
 896                         GOTO(out, rc = PTR_ERR(handle));
 897
 898                 rc = lfsck_declare_namespace_exec_dir(env, obj, handle);
 899                 if (rc != 0)
 900                         GOTO(stop, rc);
 901
 902                 rc = dt_trans_start(env, lfsck->li_next, handle);
 903                 if (rc != 0)
 904                         GOTO(stop, rc);
 905
 906                 dt_write_lock(env, obj, MOR_TGT_CHILD);
 907                 locked = true;
 908         }
 909
 910         rc = lfsck_namespace_check_exist(env, lfsck, obj, ent->lde_name);
 911         if (rc != 0)
 912                 GOTO(stop, rc);
 913
 914         rc = lfsck_links_read(env, obj, &ldata);
 915         if (rc == 0) {
 916                 count = ldata.ld_leh->leh_reccount;
 917                 rc = linkea_links_find(&ldata, cname, pfid);
 918                 if ((rc == 0) &&
 919                     (count == 1 || !S_ISDIR(lfsck_object_type(obj))))
 920                         goto record;
 921
 922                 ns->ln_flags |= LF_INCONSISTENT;
 923                 /* For dir, if there are more than one linkea entries, or the
 924                  * linkea entry does not match the name entry, then remove all
 925                  * and add the correct one. */
 926                 if (S_ISDIR(lfsck_object_type(obj))) {
 927                         remove = true;
 928                         newdata = true;
 929                 } else {
 930                         remove = false;
 931                         newdata = false;
 932                 }
 933                 goto nodata;
 934         } else if (unlikely(rc == -EINVAL)) {
 935                 count = 1;
 936                 ns->ln_flags |= LF_INCONSISTENT;
 937                 /* The magic crashed, we are not sure whether there are more
 938                  * corrupt data in the linkea, so remove all linkea entries. */
 939                 remove = true;
 940                 newdata = true;
 941                 goto nodata;
 942         } else if (rc == -ENODATA) {
 943                 count = 1;
 944                 ns->ln_flags |= LF_UPGRADE;
 945                 remove = false;
 946                 newdata = true;
 947
 948 nodata:
 949                 if (bk->lb_param & LPF_DRYRUN) {
 950                         repaired = true;
 951                         goto record;
 952                 }
 953
 954                 if (!com->lc_journal)
 955                         goto again;
 956
 957                 if (remove) {
 958                         LASSERT(newdata);
 959
 960                         rc = dt_xattr_del(env, obj, XATTR_NAME_LINK, handle,
 961                                           BYPASS_CAPA);
 962                         if (rc != 0)
 963                                 GOTO(stop, rc);
 964                 }
 965
 966                 if (newdata) {
 967                         rc = linkea_data_new(&ldata,
 968                                         &lfsck_env_info(env)->lti_linkea_buf);
 969                         if (rc != 0)
 970                                 GOTO(stop, rc);
 971                 }
 972
 973                 rc = linkea_add_buf(&ldata, cname, pfid);
 974                 if (rc != 0)
 975                         GOTO(stop, rc);
 976
 977                 rc = lfsck_links_write(env, obj, &ldata, handle);
 978                 if (rc != 0)
 979                         GOTO(stop, rc);
 980
 981                 count = ldata.ld_leh->leh_reccount;
 982                 repaired = true;
 983         } else {
 984                 GOTO(stop, rc);
 985         }
 986
 987 record:
 988         LASSERT(count > 0);
 989
 990         rc = dt_attr_get(env, obj, la, BYPASS_CAPA);
 991         if (rc != 0)
 992                 GOTO(stop, rc);
 993
 994         if ((count == 1) &&
 995             (la->la_nlink == 1 || S_ISDIR(lfsck_object_type(obj))))
 996                 /* Usually, it is for single linked object or dir, do nothing.*/
 997                 GOTO(stop, rc);
 998
 999         /* Following modification will be in another transaction.  */
1000         if (handle != NULL) {
1001                 LASSERT(dt_write_locked(env, obj));
1002
1003                 dt_write_unlock(env, obj);
1004                 locked = false;
1005
1006                 dt_trans_stop(env, lfsck->li_next, handle);
1007                 handle = NULL;
1008         }
1009
1010         ns->ln_mlinked_checked++;
1011         rc = lfsck_namespace_update(env, com, cfid,
1012                         count != la->la_nlink ? LLF_UNMATCH_NLINKS : 0, false);
1013
1014         GOTO(out, rc);
1015
1016 stop:
1017         if (locked)
1018                 dt_write_unlock(env, obj);
1019
1020         if (handle != NULL)
1021                 dt_trans_stop(env, lfsck->li_next, handle);
1022
1023 out:
1024         if (rc < 0) {
1025                 ns->ln_items_failed++;
1026                 if (lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent))
1027                         lfsck_pos_fill(env, lfsck,
1028                                        &ns->ln_pos_first_inconsistent, false);
1029                 if (!(bk->lb_param & LPF_FAILOUT))
1030                         rc = 0;
1031         } else {
1032                 if (repaired) {
1033                         ns->ln_items_repaired++;
1034                         if (bk->lb_param & LPF_DRYRUN &&
1035                             lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent))
1036                                 lfsck_pos_fill(env, lfsck,
1037                                                &ns->ln_pos_first_inconsistent,
1038                                                false);
1039                 } else {
1040                         com->lc_journal = 0;
1041                 }
1042                 rc = 0;
1043         }
1044         up_write(&com->lc_sem);
1045         return rc;
1046 }
1047
1048 static int lfsck_namespace_post(const struct lu_env *env,
1049                                 struct lfsck_component *com,
1050                                 int result, bool init)
1051 {
1052         struct lfsck_instance   *lfsck = com->lc_lfsck;
1053         struct lfsck_namespace  *ns    =
1054                                 (struct lfsck_namespace *)com->lc_file_ram;
1055         int                      rc;
1056
1057         down_write(&com->lc_sem);
1058
1059         spin_lock(&lfsck->li_lock);
1060         if (!init)
1061                 ns->ln_pos_last_checkpoint = lfsck->li_pos_current;
1062         if (result > 0) {
1063                 ns->ln_status = LS_SCANNING_PHASE2;
1064                 ns->ln_flags |= LF_SCANNED_ONCE;
1065                 ns->ln_flags &= ~LF_UPGRADE;
1066                 cfs_list_del_init(&com->lc_link);
1067                 cfs_list_del_init(&com->lc_link_dir);
1068                 cfs_list_add_tail(&com->lc_link, &lfsck->li_list_double_scan);
1069         } else if (result == 0) {
1070                 if (lfsck->li_paused) {
1071                         ns->ln_status = LS_PAUSED;
1072                 } else {
1073                         ns->ln_status = LS_STOPPED;
1074                         cfs_list_del_init(&com->lc_link);
1075                         cfs_list_del_init(&com->lc_link_dir);
1076                         cfs_list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1077                 }
1078         } else {
1079                 ns->ln_status = LS_FAILED;
1080                 cfs_list_del_init(&com->lc_link);
1081                 cfs_list_del_init(&com->lc_link_dir);
1082                 cfs_list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1083         }
1084         spin_unlock(&lfsck->li_lock);
1085
1086         if (!init) {
1087                 ns->ln_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
1088                                 HALF_SEC - lfsck->li_time_last_checkpoint);
1089                 ns->ln_time_last_checkpoint = cfs_time_current_sec();
1090                 ns->ln_items_checked += com->lc_new_checked;
1091                 com->lc_new_checked = 0;
1092         }
1093
1094         rc = lfsck_namespace_store(env, com);
1095
1096         up_write(&com->lc_sem);
1097         return rc;
1098 }
1099
1100 static int
1101 lfsck_namespace_dump(const struct lu_env *env, struct lfsck_component *com,
1102                      char *buf, int len)
1103 {
1104         struct lfsck_instance   *lfsck = com->lc_lfsck;
1105         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
1106         struct lfsck_namespace  *ns    =
1107                                 (struct lfsck_namespace *)com->lc_file_ram;
1108         int                      save  = len;
1109         int                      ret   = -ENOSPC;
1110         int                      rc;
1111
1112         down_read(&com->lc_sem);
1113         rc = snprintf(buf, len,
1114                       "name: lfsck_namespace\n"
1115                       "magic: 0x%x\n"
1116                       "version: %d\n"
1117                       "status: %s\n",
1118                       ns->ln_magic,
1119                       bk->lb_version,
1120                       lfsck_status_names[ns->ln_status]);
1121         if (rc <= 0)
1122                 goto out;
1123
1124         buf += rc;
1125         len -= rc;
1126         rc = lfsck_bits_dump(&buf, &len, ns->ln_flags, lfsck_flags_names,
1127                              "flags");
1128         if (rc < 0)
1129                 goto out;
1130
1131         rc = lfsck_bits_dump(&buf, &len, bk->lb_param, lfsck_param_names,
1132                              "param");
1133         if (rc < 0)
1134                 goto out;
1135
1136         rc = lfsck_time_dump(&buf, &len, ns->ln_time_last_complete,
1137                              "time_since_last_completed");
1138         if (rc < 0)
1139                 goto out;
1140
1141         rc = lfsck_time_dump(&buf, &len, ns->ln_time_latest_start,
1142                              "time_since_latest_start");
1143         if (rc < 0)
1144                 goto out;
1145
1146         rc = lfsck_time_dump(&buf, &len, ns->ln_time_last_checkpoint,
1147                              "time_since_last_checkpoint");
1148         if (rc < 0)
1149                 goto out;
1150
1151         rc = lfsck_pos_dump(&buf, &len, &ns->ln_pos_latest_start,
1152                             "latest_start_position");
1153         if (rc < 0)
1154                 goto out;
1155
1156         rc = lfsck_pos_dump(&buf, &len, &ns->ln_pos_last_checkpoint,
1157                             "last_checkpoint_position");
1158         if (rc < 0)
1159                 goto out;
1160
1161         rc = lfsck_pos_dump(&buf, &len, &ns->ln_pos_first_inconsistent,
1162                             "first_failure_position");
1163         if (rc < 0)
1164                 goto out;
1165
1166         if (ns->ln_status == LS_SCANNING_PHASE1) {
1167                 struct lfsck_position pos;
1168                 const struct dt_it_ops *iops;
1169                 cfs_duration_t duration = cfs_time_current() -
1170                                           lfsck->li_time_last_checkpoint;
1171                 __u64 checked = ns->ln_items_checked + com->lc_new_checked;
1172                 __u64 speed = checked;
1173                 __u64 new_checked = com->lc_new_checked * HZ;
1174                 __u32 rtime = ns->ln_run_time_phase1 +
1175                               cfs_duration_sec(duration + HALF_SEC);
1176
1177                 if (duration != 0)
1178                         do_div(new_checked, duration);
1179                 if (rtime != 0)
1180                         do_div(speed, rtime);
1181                 rc = snprintf(buf, len,
1182                               "checked_phase1: "LPU64"\n"
1183                               "checked_phase2: "LPU64"\n"
1184                               "updated_phase1: "LPU64"\n"
1185                               "updated_phase2: "LPU64"\n"
1186                               "failed_phase1: "LPU64"\n"
1187                               "failed_phase2: "LPU64"\n"
1188                               "dirs: "LPU64"\n"
1189                               "M-linked: "LPU64"\n"
1190                               "nlinks_repaired: "LPU64"\n"
1191                               "lost_found: "LPU64"\n"
1192                               "success_count: %u\n"
1193                               "run_time_phase1: %u seconds\n"
1194                               "run_time_phase2: %u seconds\n"
1195                               "average_speed_phase1: "LPU64" items/sec\n"
1196                               "average_speed_phase2: N/A\n"
1197                               "real-time_speed_phase1: "LPU64" items/sec\n"
1198                               "real-time_speed_phase2: N/A\n",
1199                               checked,
1200                               ns->ln_objs_checked_phase2,
1201                               ns->ln_items_repaired,
1202                               ns->ln_objs_repaired_phase2,
1203                               ns->ln_items_failed,
1204                               ns->ln_objs_failed_phase2,
1205                               ns->ln_dirs_checked,
1206                               ns->ln_mlinked_checked,
1207                               ns->ln_objs_nlink_repaired,
1208                               ns->ln_objs_lost_found,
1209                               ns->ln_success_count,
1210                               rtime,
1211                               ns->ln_run_time_phase2,
1212                               speed,
1213                               new_checked);
1214                 if (rc <= 0)
1215                         goto out;
1216
1217                 buf += rc;
1218                 len -= rc;
1219
1220                 LASSERT(lfsck->li_di_oit != NULL);
1221
1222                 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
1223
1224                 /* The low layer otable-based iteration position may NOT
1225                  * exactly match the namespace-based directory traversal
1226                  * cookie. Generally, it is not a serious issue. But the
1227                  * caller should NOT make assumption on that. */
1228                 pos.lp_oit_cookie = iops->store(env, lfsck->li_di_oit);
1229                 if (!lfsck->li_current_oit_processed)
1230                         pos.lp_oit_cookie--;
1231
1232                 spin_lock(&lfsck->li_lock);
1233                 if (lfsck->li_di_dir != NULL) {
1234                         pos.lp_dir_cookie = lfsck->li_cookie_dir;
1235                         if (pos.lp_dir_cookie >= MDS_DIR_END_OFF) {
1236                                 fid_zero(&pos.lp_dir_parent);
1237                                 pos.lp_dir_cookie = 0;
1238                         } else {
1239                                 pos.lp_dir_parent =
1240                                 *lu_object_fid(&lfsck->li_obj_dir->do_lu);
1241                         }
1242                 } else {
1243                         fid_zero(&pos.lp_dir_parent);
1244                         pos.lp_dir_cookie = 0;
1245                 }
1246                 spin_unlock(&lfsck->li_lock);
1247                 rc = lfsck_pos_dump(&buf, &len, &pos, "current_position");
1248                 if (rc <= 0)
1249                         goto out;
1250         } else if (ns->ln_status == LS_SCANNING_PHASE2) {
1251                 cfs_duration_t duration = cfs_time_current() -
1252                                           lfsck->li_time_last_checkpoint;
1253                 __u64 checked = ns->ln_objs_checked_phase2 +
1254                                 com->lc_new_checked;
1255                 __u64 speed1 = ns->ln_items_checked;
1256                 __u64 speed2 = checked;
1257                 __u64 new_checked = com->lc_new_checked * HZ;
1258                 __u32 rtime = ns->ln_run_time_phase2 +
1259                               cfs_duration_sec(duration + HALF_SEC);
1260
1261                 if (duration != 0)
1262                         do_div(new_checked, duration);
1263                 if (ns->ln_run_time_phase1 != 0)
1264                         do_div(speed1, ns->ln_run_time_phase1);
1265                 if (rtime != 0)
1266                         do_div(speed2, rtime);
1267                 rc = snprintf(buf, len,
1268                               "checked_phase1: "LPU64"\n"
1269                               "checked_phase2: "LPU64"\n"
1270                               "updated_phase1: "LPU64"\n"
1271                               "updated_phase2: "LPU64"\n"
1272                               "failed_phase1: "LPU64"\n"
1273                               "failed_phase2: "LPU64"\n"
1274                               "dirs: "LPU64"\n"
1275                               "M-linked: "LPU64"\n"
1276                               "nlinks_repaired: "LPU64"\n"
1277                               "lost_found: "LPU64"\n"
1278                               "success_count: %u\n"
1279                               "run_time_phase1: %u seconds\n"
1280                               "run_time_phase2: %u seconds\n"
1281                               "average_speed_phase1: "LPU64" items/sec\n"
1282                               "average_speed_phase2: "LPU64" objs/sec\n"
1283                               "real-time_speed_phase1: N/A\n"
1284                               "real-time_speed_phase2: "LPU64" objs/sec\n"
1285                               "current_position: "DFID"\n",
1286                               ns->ln_items_checked,
1287                               checked,
1288                               ns->ln_items_repaired,
1289                               ns->ln_objs_repaired_phase2,
1290                               ns->ln_items_failed,
1291                               ns->ln_objs_failed_phase2,
1292                               ns->ln_dirs_checked,
1293                               ns->ln_mlinked_checked,
1294                               ns->ln_objs_nlink_repaired,
1295                               ns->ln_objs_lost_found,
1296                               ns->ln_success_count,
1297                               ns->ln_run_time_phase1,
1298                               rtime,
1299                               speed1,
1300                               speed2,
1301                               new_checked,
1302                               PFID(&ns->ln_fid_latest_scanned_phase2));
1303                 if (rc <= 0)
1304                         goto out;
1305
1306                 buf += rc;
1307                 len -= rc;
1308         } else {
1309                 __u64 speed1 = ns->ln_items_checked;
1310                 __u64 speed2 = ns->ln_objs_checked_phase2;
1311
1312                 if (ns->ln_run_time_phase1 != 0)
1313                         do_div(speed1, ns->ln_run_time_phase1);
1314                 if (ns->ln_run_time_phase2 != 0)
1315                         do_div(speed2, ns->ln_run_time_phase2);
1316                 rc = snprintf(buf, len,
1317                               "checked_phase1: "LPU64"\n"
1318                               "checked_phase2: "LPU64"\n"
1319                               "updated_phase1: "LPU64"\n"
1320                               "updated_phase2: "LPU64"\n"
1321                               "failed_phase1: "LPU64"\n"
1322                               "failed_phase2: "LPU64"\n"
1323                               "dirs: "LPU64"\n"
1324                               "M-linked: "LPU64"\n"
1325                               "nlinks_repaired: "LPU64"\n"
1326                               "lost_found: "LPU64"\n"
1327                               "success_count: %u\n"
1328                               "run_time_phase1: %u seconds\n"
1329                               "run_time_phase2: %u seconds\n"
1330                               "average_speed_phase1: "LPU64" items/sec\n"
1331                               "average_speed_phase2: "LPU64" objs/sec\n"
1332                               "real-time_speed_phase1: N/A\n"
1333                               "real-time_speed_phase2: N/A\n"
1334                               "current_position: N/A\n",
1335                               ns->ln_items_checked,
1336                               ns->ln_objs_checked_phase2,
1337                               ns->ln_items_repaired,
1338                               ns->ln_objs_repaired_phase2,
1339                               ns->ln_items_failed,
1340                               ns->ln_objs_failed_phase2,
1341                               ns->ln_dirs_checked,
1342                               ns->ln_mlinked_checked,
1343                               ns->ln_objs_nlink_repaired,
1344                               ns->ln_objs_lost_found,
1345                               ns->ln_success_count,
1346                               ns->ln_run_time_phase1,
1347                               ns->ln_run_time_phase2,
1348                               speed1,
1349                               speed2);
1350                 if (rc <= 0)
1351                         goto out;
1352
1353                 buf += rc;
1354                 len -= rc;
1355         }
1356         ret = save - len;
1357
1358 out:
1359         up_read(&com->lc_sem);
1360         return ret;
1361 }
1362
1363 static int lfsck_namespace_double_scan(const struct lu_env *env,
1364                                        struct lfsck_component *com)
1365 {
1366         struct lfsck_instance   *lfsck  = com->lc_lfsck;
1367         struct ptlrpc_thread    *thread = &lfsck->li_thread;
1368         struct lfsck_bookmark   *bk     = &lfsck->li_bookmark_ram;
1369         struct lfsck_namespace  *ns     =
1370                                 (struct lfsck_namespace *)com->lc_file_ram;
1371         struct dt_object        *obj    = com->lc_obj;
1372         const struct dt_it_ops  *iops   = &obj->do_index_ops->dio_it;
1373         struct dt_object        *target;
1374         struct dt_it            *di;
1375         struct dt_key           *key;
1376         struct lu_fid            fid;
1377         int                      rc;
1378         __u8                     flags = 0;
1379         ENTRY;
1380
1381         lfsck->li_new_scanned = 0;
1382         lfsck->li_time_last_checkpoint = cfs_time_current();
1383         lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
1384                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
1385
1386         di = iops->init(env, obj, 0, BYPASS_CAPA);
1387         if (IS_ERR(di))
1388                 RETURN(PTR_ERR(di));
1389
1390         fid_cpu_to_be(&fid, &ns->ln_fid_latest_scanned_phase2);
1391         rc = iops->get(env, di, (const struct dt_key *)&fid);
1392         if (rc < 0)
1393                 GOTO(fini, rc);
1394
1395         /* Skip the start one, which either has been processed or non-exist. */
1396         rc = iops->next(env, di);
1397         if (rc != 0)
1398                 GOTO(put, rc);
1399
1400         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_DOUBLESCAN))
1401                 GOTO(put, rc = 0);
1402
1403         do {
1404                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY3) &&
1405                     cfs_fail_val > 0) {
1406                         struct l_wait_info lwi;
1407
1408                         lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val),
1409                                           NULL, NULL);
1410                         l_wait_event(thread->t_ctl_waitq,
1411                                      !thread_is_running(thread),
1412                                      &lwi);
1413                 }
1414
1415                 key = iops->key(env, di);
1416                 fid_be_to_cpu(&fid, (const struct lu_fid *)key);
1417                 target = lfsck_object_find(env, lfsck, &fid);
1418                 down_write(&com->lc_sem);
1419                 if (target == NULL) {
1420                         rc = 0;
1421                         goto checkpoint;
1422                 } else if (IS_ERR(target)) {
1423                         rc = PTR_ERR(target);
1424                         goto checkpoint;
1425                 }
1426
1427                 /* XXX: Currently, skip remote object, the consistency for
1428                  *      remote object will be processed in LFSCK phase III. */
1429                 if (dt_object_exists(target) && !dt_object_remote(target)) {
1430                         rc = iops->rec(env, di, (struct dt_rec *)&flags, 0);
1431                         if (rc == 0)
1432                                 rc = lfsck_namespace_double_scan_one(env, com,
1433                                                                 target, flags);
1434                 }
1435
1436                 lfsck_object_put(env, target);
1437
1438 checkpoint:
1439                 lfsck->li_new_scanned++;
1440                 com->lc_new_checked++;
1441                 ns->ln_fid_latest_scanned_phase2 = fid;
1442                 if (rc > 0)
1443                         ns->ln_objs_repaired_phase2++;
1444                 else if (rc < 0)
1445                         ns->ln_objs_failed_phase2++;
1446                 up_write(&com->lc_sem);
1447
1448                 if ((rc == 0) || ((rc > 0) && !(bk->lb_param & LPF_DRYRUN))) {
1449                         lfsck_namespace_delete(env, com, &fid);
1450                 } else if (rc < 0) {
1451                         flags |= LLF_REPAIR_FAILED;
1452                         lfsck_namespace_update(env, com, &fid, flags, true);
1453                 }
1454
1455                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
1456                         GOTO(put, rc);
1457
1458                 if (unlikely(cfs_time_beforeq(lfsck->li_time_next_checkpoint,
1459                                               cfs_time_current())) &&
1460                     com->lc_new_checked != 0) {
1461                         down_write(&com->lc_sem);
1462                         ns->ln_run_time_phase2 +=
1463                                 cfs_duration_sec(cfs_time_current() +
1464                                 HALF_SEC - lfsck->li_time_last_checkpoint);
1465                         ns->ln_time_last_checkpoint = cfs_time_current_sec();
1466                         ns->ln_objs_checked_phase2 += com->lc_new_checked;
1467                         com->lc_new_checked = 0;
1468                         rc = lfsck_namespace_store(env, com);
1469                         up_write(&com->lc_sem);
1470                         if (rc != 0)
1471                                 GOTO(put, rc);
1472
1473                         lfsck->li_time_last_checkpoint = cfs_time_current();
1474                         lfsck->li_time_next_checkpoint =
1475                                 lfsck->li_time_last_checkpoint +
1476                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
1477                 }
1478
1479                 lfsck_control_speed(lfsck);
1480                 if (unlikely(!thread_is_running(thread)))
1481                         GOTO(put, rc = 0);
1482
1483                 rc = iops->next(env, di);
1484         } while (rc == 0);
1485
1486         GOTO(put, rc);
1487
1488 put:
1489         iops->put(env, di);
1490
1491 fini:
1492         iops->fini(env, di);
1493         down_write(&com->lc_sem);
1494
1495         ns->ln_run_time_phase2 += cfs_duration_sec(cfs_time_current() +
1496                                 HALF_SEC - lfsck->li_time_last_checkpoint);
1497         ns->ln_time_last_checkpoint = cfs_time_current_sec();
1498         ns->ln_objs_checked_phase2 += com->lc_new_checked;
1499         com->lc_new_checked = 0;
1500
1501         if (rc > 0) {
1502                 com->lc_journal = 0;
1503                 ns->ln_status = LS_COMPLETED;
1504                 if (!(bk->lb_param & LPF_DRYRUN))
1505                         ns->ln_flags &=
1506                         ~(LF_SCANNED_ONCE | LF_INCONSISTENT | LF_UPGRADE);
1507                 ns->ln_time_last_complete = ns->ln_time_last_checkpoint;
1508                 ns->ln_success_count++;
1509         } else if (rc == 0) {
1510                 if (lfsck->li_paused)
1511                         ns->ln_status = LS_PAUSED;
1512                 else
1513                         ns->ln_status = LS_STOPPED;
1514         } else {
1515                 ns->ln_status = LS_FAILED;
1516         }
1517
1518         if (ns->ln_status != LS_PAUSED) {
1519                 spin_lock(&lfsck->li_lock);
1520                 cfs_list_del_init(&com->lc_link);
1521                 cfs_list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1522                 spin_unlock(&lfsck->li_lock);
1523         }
1524
1525         rc = lfsck_namespace_store(env, com);
1526
1527         up_write(&com->lc_sem);
1528         return rc;
1529 }
1530
1531 static struct lfsck_operations lfsck_namespace_ops = {
1532         .lfsck_reset            = lfsck_namespace_reset,
1533         .lfsck_fail             = lfsck_namespace_fail,
1534         .lfsck_checkpoint       = lfsck_namespace_checkpoint,
1535         .lfsck_prep             = lfsck_namespace_prep,
1536         .lfsck_exec_oit         = lfsck_namespace_exec_oit,
1537         .lfsck_exec_dir         = lfsck_namespace_exec_dir,
1538         .lfsck_post             = lfsck_namespace_post,
1539         .lfsck_dump             = lfsck_namespace_dump,
1540         .lfsck_double_scan      = lfsck_namespace_double_scan,
1541 };
1542
1543 int lfsck_namespace_setup(const struct lu_env *env,
1544                           struct lfsck_instance *lfsck)
1545 {
1546         struct lfsck_component  *com;
1547         struct lfsck_namespace  *ns;
1548         struct dt_object        *root = NULL;
1549         struct dt_object        *obj;
1550         int                      rc;
1551         ENTRY;
1552
1553         LASSERT(lfsck->li_master);
1554
1555         OBD_ALLOC_PTR(com);
1556         if (com == NULL)
1557                 RETURN(-ENOMEM);
1558
1559         CFS_INIT_LIST_HEAD(&com->lc_link);
1560         CFS_INIT_LIST_HEAD(&com->lc_link_dir);
1561         init_rwsem(&com->lc_sem);
1562         atomic_set(&com->lc_ref, 1);
1563         com->lc_lfsck = lfsck;
1564         com->lc_type = LT_NAMESPACE;
1565         com->lc_ops = &lfsck_namespace_ops;
1566         com->lc_file_size = sizeof(struct lfsck_namespace);
1567         OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
1568         if (com->lc_file_ram == NULL)
1569                 GOTO(out, rc = -ENOMEM);
1570
1571         OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
1572         if (com->lc_file_disk == NULL)
1573                 GOTO(out, rc = -ENOMEM);
1574
1575         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
1576         if (IS_ERR(root))
1577                 GOTO(out, rc = PTR_ERR(root));
1578
1579         dt_try_as_dir(env, root);
1580         obj = local_index_find_or_create(env, lfsck->li_los, root,
1581                                          lfsck_namespace_name,
1582                                          S_IFREG | S_IRUGO | S_IWUSR,
1583                                          &dt_lfsck_features);
1584         if (IS_ERR(obj))
1585                 GOTO(out, rc = PTR_ERR(obj));
1586
1587         com->lc_obj = obj;
1588         rc = obj->do_ops->do_index_try(env, obj, &dt_lfsck_features);
1589         if (rc != 0)
1590                 GOTO(out, rc);
1591
1592         rc = lfsck_namespace_load(env, com);
1593         if (rc == -ENODATA)
1594                 rc = lfsck_namespace_init(env, com);
1595         else if (rc < 0)
1596                 rc = lfsck_namespace_reset(env, com, true);
1597         if (rc != 0)
1598                 GOTO(out, rc);
1599
1600         ns = (struct lfsck_namespace *)com->lc_file_ram;
1601         switch (ns->ln_status) {
1602         case LS_INIT:
1603         case LS_COMPLETED:
1604         case LS_FAILED:
1605         case LS_STOPPED:
1606                 cfs_list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1607                 break;
1608         default:
1609                 CERROR("%s: unknown lfsck_namespace status: %u\n",
1610                        lfsck_lfsck2name(lfsck), ns->ln_status);
1611                 /* fall through */
1612         case LS_SCANNING_PHASE1:
1613         case LS_SCANNING_PHASE2:
1614                 /* No need to store the status to disk right now.
1615                  * If the system crashed before the status stored,
1616                  * it will be loaded back when next time. */
1617                 ns->ln_status = LS_CRASHED;
1618                 /* fall through */
1619         case LS_PAUSED:
1620         case LS_CRASHED:
1621                 cfs_list_add_tail(&com->lc_link, &lfsck->li_list_scan);
1622                 cfs_list_add_tail(&com->lc_link_dir, &lfsck->li_list_dir);
1623                 break;
1624         }
1625
1626         GOTO(out, rc = 0);
1627
1628 out:
1629         if (root != NULL && !IS_ERR(root))
1630                 lu_object_put(env, &root->do_lu);
1631         if (rc != 0)
1632                 lfsck_component_cleanup(env, com);
1633         return rc;
1634 }