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