lustre/lfsck/lfsck_namespace.c

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