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