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) 2013, 2016, 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_fid.h>
  38 #include <lustre_lib.h>
  39 #include <lustre_net.h>
  40 #include <lustre/lustre_user.h>
  41
  42 #include "lfsck_internal.h"
  43
  44 #define LFSCK_NAMESPACE_MAGIC_V1        0xA0629D03
  45 #define LFSCK_NAMESPACE_MAGIC_V2        0xA0621A0B
  46 #define LFSCK_NAMESPACE_MAGIC_V3        0xA06249FF
  47
  48 /* For Lustre-2.x (x <= 6), the namespace LFSCK used LFSCK_NAMESPACE_MAGIC_V1
  49  * as the trace file magic. When downgrade to such old release, the old LFSCK
  50  * will not recognize the new LFSCK_NAMESPACE_MAGIC_V2 in the new trace file,
  51  * then it will reset the whole LFSCK, and will not cause start failure. The
  52  * similar case will happen when upgrade from such old release. */
  53 #define LFSCK_NAMESPACE_MAGIC           LFSCK_NAMESPACE_MAGIC_V3
  54
  55 enum lfsck_nameentry_check {
  56         LFSCK_NAMEENTRY_DEAD            = 1, /* The object has been unlinked. */
  57         LFSCK_NAMEENTRY_REMOVED         = 2, /* The entry has been removed. */
  58         LFSCK_NAMEENTRY_RECREATED       = 3, /* The entry has been recreated. */
  59 };
  60
  61 static struct lfsck_namespace_req *
  62 lfsck_namespace_assistant_req_init(struct lfsck_instance *lfsck,
  63                                    struct lfsck_assistant_object *lso,
  64                                    struct lu_dirent *ent, __u16 type)
  65 {
  66         struct lfsck_namespace_req *lnr;
  67         int                         size;
  68
  69         size = sizeof(*lnr) + (ent->lde_namelen & ~3) + 4;
  70         OBD_ALLOC(lnr, size);
  71         if (lnr == NULL)
  72                 return ERR_PTR(-ENOMEM);
  73
  74         INIT_LIST_HEAD(&lnr->lnr_lar.lar_list);
  75         lnr->lnr_lar.lar_parent = lfsck_assistant_object_get(lso);
  76         lnr->lnr_lmv = lfsck_lmv_get(lfsck->li_lmv);
  77         lnr->lnr_fid = ent->lde_fid;
  78         lnr->lnr_dir_cookie = ent->lde_hash;
  79         lnr->lnr_attr = ent->lde_attrs;
  80         lnr->lnr_size = size;
  81         lnr->lnr_type = type;
  82         lnr->lnr_namelen = ent->lde_namelen;
  83         memcpy(lnr->lnr_name, ent->lde_name, ent->lde_namelen);
  84
  85         return lnr;
  86 }
  87
  88 static void lfsck_namespace_assistant_req_fini(const struct lu_env *env,
  89                                                struct lfsck_assistant_req *lar)
  90 {
  91         struct lfsck_namespace_req *lnr =
  92                         container_of0(lar, struct lfsck_namespace_req, lnr_lar);
  93
  94         if (lnr->lnr_lmv != NULL)
  95                 lfsck_lmv_put(env, lnr->lnr_lmv);
  96
  97         lfsck_assistant_object_put(env, lar->lar_parent);
  98         OBD_FREE(lnr, lnr->lnr_size);
  99 }
 100
 101 static void lfsck_namespace_le_to_cpu(struct lfsck_namespace *dst,
 102                                       struct lfsck_namespace *src)
 103 {
 104         dst->ln_magic = le32_to_cpu(src->ln_magic);
 105         dst->ln_status = le32_to_cpu(src->ln_status);
 106         dst->ln_flags = le32_to_cpu(src->ln_flags);
 107         dst->ln_success_count = le32_to_cpu(src->ln_success_count);
 108         dst->ln_run_time_phase1 = le64_to_cpu(src->ln_run_time_phase1);
 109         dst->ln_run_time_phase2 = le64_to_cpu(src->ln_run_time_phase2);
 110         dst->ln_time_last_complete = le64_to_cpu(src->ln_time_last_complete);
 111         dst->ln_time_latest_start = le64_to_cpu(src->ln_time_latest_start);
 112         dst->ln_time_last_checkpoint =
 113                                 le64_to_cpu(src->ln_time_last_checkpoint);
 114         lfsck_position_le_to_cpu(&dst->ln_pos_latest_start,
 115                                  &src->ln_pos_latest_start);
 116         lfsck_position_le_to_cpu(&dst->ln_pos_last_checkpoint,
 117                                  &src->ln_pos_last_checkpoint);
 118         lfsck_position_le_to_cpu(&dst->ln_pos_first_inconsistent,
 119                                  &src->ln_pos_first_inconsistent);
 120         dst->ln_items_checked = le64_to_cpu(src->ln_items_checked);
 121         dst->ln_items_repaired = le64_to_cpu(src->ln_items_repaired);
 122         dst->ln_items_failed = le64_to_cpu(src->ln_items_failed);
 123         dst->ln_dirs_checked = le64_to_cpu(src->ln_dirs_checked);
 124         dst->ln_objs_checked_phase2 = le64_to_cpu(src->ln_objs_checked_phase2);
 125         dst->ln_objs_repaired_phase2 =
 126                                 le64_to_cpu(src->ln_objs_repaired_phase2);
 127         dst->ln_objs_failed_phase2 = le64_to_cpu(src->ln_objs_failed_phase2);
 128         dst->ln_objs_nlink_repaired = le64_to_cpu(src->ln_objs_nlink_repaired);
 129         fid_le_to_cpu(&dst->ln_fid_latest_scanned_phase2,
 130                       &src->ln_fid_latest_scanned_phase2);
 131         dst->ln_dirent_repaired = le64_to_cpu(src->ln_dirent_repaired);
 132         dst->ln_linkea_repaired = le64_to_cpu(src->ln_linkea_repaired);
 133         dst->ln_mul_linked_checked = le64_to_cpu(src->ln_mul_linked_checked);
 134         dst->ln_mul_linked_repaired = le64_to_cpu(src->ln_mul_linked_repaired);
 135         dst->ln_unknown_inconsistency =
 136                                 le64_to_cpu(src->ln_unknown_inconsistency);
 137         dst->ln_unmatched_pairs_repaired =
 138                                 le64_to_cpu(src->ln_unmatched_pairs_repaired);
 139         dst->ln_dangling_repaired = le64_to_cpu(src->ln_dangling_repaired);
 140         dst->ln_mul_ref_repaired = le64_to_cpu(src->ln_mul_ref_repaired);
 141         dst->ln_bad_type_repaired = le64_to_cpu(src->ln_bad_type_repaired);
 142         dst->ln_lost_dirent_repaired =
 143                                 le64_to_cpu(src->ln_lost_dirent_repaired);
 144         dst->ln_striped_dirs_scanned =
 145                                 le64_to_cpu(src->ln_striped_dirs_scanned);
 146         dst->ln_striped_dirs_repaired =
 147                                 le64_to_cpu(src->ln_striped_dirs_repaired);
 148         dst->ln_striped_dirs_failed =
 149                                 le64_to_cpu(src->ln_striped_dirs_failed);
 150         dst->ln_striped_dirs_disabled =
 151                                 le64_to_cpu(src->ln_striped_dirs_disabled);
 152         dst->ln_striped_dirs_skipped =
 153                                 le64_to_cpu(src->ln_striped_dirs_skipped);
 154         dst->ln_striped_shards_scanned =
 155                                 le64_to_cpu(src->ln_striped_shards_scanned);
 156         dst->ln_striped_shards_repaired =
 157                                 le64_to_cpu(src->ln_striped_shards_repaired);
 158         dst->ln_striped_shards_failed =
 159                                 le64_to_cpu(src->ln_striped_shards_failed);
 160         dst->ln_striped_shards_skipped =
 161                                 le64_to_cpu(src->ln_striped_shards_skipped);
 162         dst->ln_name_hash_repaired = le64_to_cpu(src->ln_name_hash_repaired);
 163         dst->ln_local_lpf_scanned = le64_to_cpu(src->ln_local_lpf_scanned);
 164         dst->ln_local_lpf_moved = le64_to_cpu(src->ln_local_lpf_moved);
 165         dst->ln_local_lpf_skipped = le64_to_cpu(src->ln_local_lpf_skipped);
 166         dst->ln_local_lpf_failed = le64_to_cpu(src->ln_local_lpf_failed);
 167         dst->ln_bitmap_size = le32_to_cpu(src->ln_bitmap_size);
 168         dst->ln_time_latest_reset = le64_to_cpu(src->ln_time_latest_reset);
 169         dst->ln_linkea_overflow_cleared =
 170                                 le64_to_cpu(src->ln_linkea_overflow_cleared);
 171 }
 172
 173 static void lfsck_namespace_cpu_to_le(struct lfsck_namespace *dst,
 174                                       struct lfsck_namespace *src)
 175 {
 176         dst->ln_magic = cpu_to_le32(src->ln_magic);
 177         dst->ln_status = cpu_to_le32(src->ln_status);
 178         dst->ln_flags = cpu_to_le32(src->ln_flags);
 179         dst->ln_success_count = cpu_to_le32(src->ln_success_count);
 180         dst->ln_run_time_phase1 = cpu_to_le64(src->ln_run_time_phase1);
 181         dst->ln_run_time_phase2 = cpu_to_le64(src->ln_run_time_phase2);
 182         dst->ln_time_last_complete = cpu_to_le64(src->ln_time_last_complete);
 183         dst->ln_time_latest_start = cpu_to_le64(src->ln_time_latest_start);
 184         dst->ln_time_last_checkpoint =
 185                                 cpu_to_le64(src->ln_time_last_checkpoint);
 186         lfsck_position_cpu_to_le(&dst->ln_pos_latest_start,
 187                                  &src->ln_pos_latest_start);
 188         lfsck_position_cpu_to_le(&dst->ln_pos_last_checkpoint,
 189                                  &src->ln_pos_last_checkpoint);
 190         lfsck_position_cpu_to_le(&dst->ln_pos_first_inconsistent,
 191                                  &src->ln_pos_first_inconsistent);
 192         dst->ln_items_checked = cpu_to_le64(src->ln_items_checked);
 193         dst->ln_items_repaired = cpu_to_le64(src->ln_items_repaired);
 194         dst->ln_items_failed = cpu_to_le64(src->ln_items_failed);
 195         dst->ln_dirs_checked = cpu_to_le64(src->ln_dirs_checked);
 196         dst->ln_objs_checked_phase2 = cpu_to_le64(src->ln_objs_checked_phase2);
 197         dst->ln_objs_repaired_phase2 =
 198                                 cpu_to_le64(src->ln_objs_repaired_phase2);
 199         dst->ln_objs_failed_phase2 = cpu_to_le64(src->ln_objs_failed_phase2);
 200         dst->ln_objs_nlink_repaired = cpu_to_le64(src->ln_objs_nlink_repaired);
 201         fid_cpu_to_le(&dst->ln_fid_latest_scanned_phase2,
 202                       &src->ln_fid_latest_scanned_phase2);
 203         dst->ln_dirent_repaired = cpu_to_le64(src->ln_dirent_repaired);
 204         dst->ln_linkea_repaired = cpu_to_le64(src->ln_linkea_repaired);
 205         dst->ln_mul_linked_checked = cpu_to_le64(src->ln_mul_linked_checked);
 206         dst->ln_mul_linked_repaired = cpu_to_le64(src->ln_mul_linked_repaired);
 207         dst->ln_unknown_inconsistency =
 208                                 cpu_to_le64(src->ln_unknown_inconsistency);
 209         dst->ln_unmatched_pairs_repaired =
 210                                 cpu_to_le64(src->ln_unmatched_pairs_repaired);
 211         dst->ln_dangling_repaired = cpu_to_le64(src->ln_dangling_repaired);
 212         dst->ln_mul_ref_repaired = cpu_to_le64(src->ln_mul_ref_repaired);
 213         dst->ln_bad_type_repaired = cpu_to_le64(src->ln_bad_type_repaired);
 214         dst->ln_lost_dirent_repaired =
 215                                 cpu_to_le64(src->ln_lost_dirent_repaired);
 216         dst->ln_striped_dirs_scanned =
 217                                 cpu_to_le64(src->ln_striped_dirs_scanned);
 218         dst->ln_striped_dirs_repaired =
 219                                 cpu_to_le64(src->ln_striped_dirs_repaired);
 220         dst->ln_striped_dirs_failed =
 221                                 cpu_to_le64(src->ln_striped_dirs_failed);
 222         dst->ln_striped_dirs_disabled =
 223                                 cpu_to_le64(src->ln_striped_dirs_disabled);
 224         dst->ln_striped_dirs_skipped =
 225                                 cpu_to_le64(src->ln_striped_dirs_skipped);
 226         dst->ln_striped_shards_scanned =
 227                                 cpu_to_le64(src->ln_striped_shards_scanned);
 228         dst->ln_striped_shards_repaired =
 229                                 cpu_to_le64(src->ln_striped_shards_repaired);
 230         dst->ln_striped_shards_failed =
 231                                 cpu_to_le64(src->ln_striped_shards_failed);
 232         dst->ln_striped_shards_skipped =
 233                                 cpu_to_le64(src->ln_striped_shards_skipped);
 234         dst->ln_name_hash_repaired = cpu_to_le64(src->ln_name_hash_repaired);
 235         dst->ln_local_lpf_scanned = cpu_to_le64(src->ln_local_lpf_scanned);
 236         dst->ln_local_lpf_moved = cpu_to_le64(src->ln_local_lpf_moved);
 237         dst->ln_local_lpf_skipped = cpu_to_le64(src->ln_local_lpf_skipped);
 238         dst->ln_local_lpf_failed = cpu_to_le64(src->ln_local_lpf_failed);
 239         dst->ln_bitmap_size = cpu_to_le32(src->ln_bitmap_size);
 240         dst->ln_time_latest_reset = cpu_to_le64(src->ln_time_latest_reset);
 241         dst->ln_linkea_overflow_cleared =
 242                                 cpu_to_le64(src->ln_linkea_overflow_cleared);
 243 }
 244
 245 static void lfsck_namespace_record_failure(const struct lu_env *env,
 246                                            struct lfsck_instance *lfsck,
 247                                            struct lfsck_namespace *ns)
 248 {
 249         struct lfsck_position pos;
 250
 251         ns->ln_items_failed++;
 252         lfsck_pos_fill(env, lfsck, &pos, false);
 253         if (lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent) ||
 254             lfsck_pos_is_eq(&pos, &ns->ln_pos_first_inconsistent) < 0) {
 255                 ns->ln_pos_first_inconsistent = pos;
 256
 257                 CDEBUG(D_LFSCK, "%s: namespace LFSCK hit first non-repaired "
 258                        "inconsistency at the pos [%llu, "DFID", %#llx]\n",
 259                        lfsck_lfsck2name(lfsck),
 260                        ns->ln_pos_first_inconsistent.lp_oit_cookie,
 261                        PFID(&ns->ln_pos_first_inconsistent.lp_dir_parent),
 262                        ns->ln_pos_first_inconsistent.lp_dir_cookie);
 263         }
 264 }
 265
 266 /**
 267  * Load the MDT bitmap from the lfsck_namespace trace file.
 268  *
 269  * \param[in] env       pointer to the thread context
 270  * \param[in] com       pointer to the lfsck component
 271  *
 272  * \retval              0 for success
 273  * \retval              negative error number on failure or data corruption
 274  */
 275 static int lfsck_namespace_load_bitmap(const struct lu_env *env,
 276                                        struct lfsck_component *com)
 277 {
 278         struct dt_object                *obj    = com->lc_obj;
 279         struct lfsck_assistant_data     *lad    = com->lc_data;
 280         struct lfsck_namespace          *ns     = com->lc_file_ram;
 281         struct cfs_bitmap                       *bitmap = lad->lad_bitmap;
 282         ssize_t                          size;
 283         __u32                            nbits;
 284         int                              rc;
 285         ENTRY;
 286
 287         if (com->lc_lfsck->li_mdt_descs.ltd_tgts_bitmap->size >
 288             ns->ln_bitmap_size)
 289                 nbits = com->lc_lfsck->li_mdt_descs.ltd_tgts_bitmap->size;
 290         else
 291                 nbits = ns->ln_bitmap_size;
 292
 293         if (unlikely(nbits < BITS_PER_LONG))
 294                 nbits = BITS_PER_LONG;
 295
 296         if (nbits > bitmap->size) {
 297                 __u32 new_bits = bitmap->size;
 298                 struct cfs_bitmap *new_bitmap;
 299
 300                 while (new_bits < nbits)
 301                         new_bits <<= 1;
 302
 303                 new_bitmap = CFS_ALLOCATE_BITMAP(new_bits);
 304                 if (new_bitmap == NULL)
 305                         RETURN(-ENOMEM);
 306
 307                 lad->lad_bitmap = new_bitmap;
 308                 CFS_FREE_BITMAP(bitmap);
 309                 bitmap = new_bitmap;
 310         }
 311
 312         if (ns->ln_bitmap_size == 0) {
 313                 lad->lad_incomplete = 0;
 314                 CFS_RESET_BITMAP(bitmap);
 315
 316                 RETURN(0);
 317         }
 318
 319         size = (ns->ln_bitmap_size + 7) >> 3;
 320         rc = dt_xattr_get(env, obj,
 321                           lfsck_buf_get(env, bitmap->data, size),
 322                           XATTR_NAME_LFSCK_BITMAP);
 323         if (rc != size)
 324                 RETURN(rc >= 0 ? -EINVAL : rc);
 325
 326         if (cfs_bitmap_check_empty(bitmap))
 327                 lad->lad_incomplete = 0;
 328         else
 329                 lad->lad_incomplete = 1;
 330
 331         RETURN(0);
 332 }
 333
 334 /**
 335  * Load namespace LFSCK statistics information from the trace file.
 336  *
 337  * \param[in] env       pointer to the thread context
 338  * \param[in] com       pointer to the lfsck component
 339  *
 340  * \retval              0 for success
 341  * \retval              negative error number on failure
 342  */
 343 static int lfsck_namespace_load(const struct lu_env *env,
 344                                 struct lfsck_component *com)
 345 {
 346         int len = com->lc_file_size;
 347         int rc;
 348
 349         rc = dt_xattr_get(env, com->lc_obj,
 350                           lfsck_buf_get(env, com->lc_file_disk, len),
 351                           XATTR_NAME_LFSCK_NAMESPACE);
 352         if (rc == len) {
 353                 struct lfsck_namespace *ns = com->lc_file_ram;
 354
 355                 lfsck_namespace_le_to_cpu(ns,
 356                                 (struct lfsck_namespace *)com->lc_file_disk);
 357                 if (ns->ln_magic != LFSCK_NAMESPACE_MAGIC) {
 358                         CDEBUG(D_LFSCK, "%s: invalid lfsck_namespace magic "
 359                                "%#x != %#x\n", lfsck_lfsck2name(com->lc_lfsck),
 360                                ns->ln_magic, LFSCK_NAMESPACE_MAGIC);
 361                         rc = -ESTALE;
 362                 } else {
 363                         rc = 0;
 364                 }
 365         } else if (rc != -ENODATA) {
 366                 CDEBUG(D_LFSCK, "%s: fail to load lfsck_namespace, "
 367                        "expected = %d: rc = %d\n",
 368                        lfsck_lfsck2name(com->lc_lfsck), len, rc);
 369                 if (rc >= 0)
 370                         rc = -ESTALE;
 371         }
 372
 373         return rc;
 374 }
 375
 376 static int lfsck_namespace_store(const struct lu_env *env,
 377                                  struct lfsck_component *com)
 378 {
 379         struct dt_object                *obj    = com->lc_obj;
 380         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 381         struct lfsck_namespace          *ns     = com->lc_file_ram;
 382         struct lfsck_assistant_data     *lad    = com->lc_data;
 383         struct dt_device                *dev    = lfsck_obj2dev(obj);
 384         struct cfs_bitmap               *bitmap = NULL;
 385         struct thandle                  *handle;
 386         __u32                            nbits  = 0;
 387         int                              len    = com->lc_file_size;
 388         int                              rc;
 389         ENTRY;
 390
 391         if (lad != NULL) {
 392                 bitmap = lad->lad_bitmap;
 393                 nbits = bitmap->size;
 394
 395                 LASSERT(nbits > 0);
 396                 LASSERTF((nbits & 7) == 0, "Invalid nbits %u\n", nbits);
 397         }
 398
 399         ns->ln_bitmap_size = nbits;
 400         lfsck_namespace_cpu_to_le((struct lfsck_namespace *)com->lc_file_disk,
 401                                   ns);
 402         handle = dt_trans_create(env, dev);
 403         if (IS_ERR(handle))
 404                 GOTO(log, rc = PTR_ERR(handle));
 405
 406         rc = dt_declare_xattr_set(env, obj,
 407                                   lfsck_buf_get(env, com->lc_file_disk, len),
 408                                   XATTR_NAME_LFSCK_NAMESPACE, 0, handle);
 409         if (rc != 0)
 410                 GOTO(out, rc);
 411
 412         if (bitmap != NULL) {
 413                 rc = dt_declare_xattr_set(env, obj,
 414                                 lfsck_buf_get(env, bitmap->data, nbits >> 3),
 415                                 XATTR_NAME_LFSCK_BITMAP, 0, handle);
 416                 if (rc != 0)
 417                         GOTO(out, rc);
 418         }
 419
 420         rc = dt_trans_start_local(env, dev, handle);
 421         if (rc != 0)
 422                 GOTO(out, rc);
 423
 424         rc = dt_xattr_set(env, obj,
 425                           lfsck_buf_get(env, com->lc_file_disk, len),
 426                           XATTR_NAME_LFSCK_NAMESPACE, 0, handle);
 427         if (rc == 0 && bitmap != NULL)
 428                 rc = dt_xattr_set(env, obj,
 429                                   lfsck_buf_get(env, bitmap->data, nbits >> 3),
 430                                   XATTR_NAME_LFSCK_BITMAP, 0, handle);
 431
 432         GOTO(out, rc);
 433
 434 out:
 435         dt_trans_stop(env, dev, handle);
 436
 437 log:
 438         if (rc != 0)
 439                 CDEBUG(D_LFSCK, "%s: fail to store lfsck_namespace: rc = %d\n",
 440                        lfsck_lfsck2name(lfsck), rc);
 441         return rc;
 442 }
 443
 444 static int lfsck_namespace_init(const struct lu_env *env,
 445                                 struct lfsck_component *com)
 446 {
 447         struct lfsck_namespace *ns = com->lc_file_ram;
 448         int rc;
 449
 450         memset(ns, 0, sizeof(*ns));
 451         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
 452         ns->ln_status = LS_INIT;
 453         ns->ln_time_latest_reset = ktime_get_real_seconds();
 454         down_write(&com->lc_sem);
 455         rc = lfsck_namespace_store(env, com);
 456         if (rc == 0)
 457                 rc = lfsck_load_sub_trace_files(env, com,
 458                         &dt_lfsck_namespace_features, LFSCK_NAMESPACE, true);
 459         up_write(&com->lc_sem);
 460
 461         return rc;
 462 }
 463
 464 /**
 465  * Update the namespace LFSCK trace file for the given @fid
 466  *
 467  * \param[in] env       pointer to the thread context
 468  * \param[in] com       pointer to the lfsck component
 469  * \param[in] fid       the fid which flags to be updated in the lfsck
 470  *                      trace file
 471  * \param[in] add       true if add new flags, otherwise remove flags
 472  *
 473  * \retval              0 for success or nothing to be done
 474  * \retval              negative error number on failure
 475  */
 476 int lfsck_namespace_trace_update(const struct lu_env *env,
 477                                  struct lfsck_component *com,
 478                                  const struct lu_fid *fid,
 479                                  const __u8 flags, bool add)
 480 {
 481         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 482         struct dt_object        *obj;
 483         struct lu_fid           *key    = &lfsck_env_info(env)->lti_fid3;
 484         struct dt_device        *dev;
 485         struct thandle          *th     = NULL;
 486         int                      idx;
 487         int                      rc     = 0;
 488         __u8                     old    = 0;
 489         __u8                     new    = 0;
 490         ENTRY;
 491
 492         LASSERT(flags != 0);
 493
 494         if (unlikely(!fid_is_sane(fid)))
 495                 RETURN(0);
 496
 497         idx = lfsck_sub_trace_file_fid2idx(fid);
 498         mutex_lock(&com->lc_sub_trace_objs[idx].lsto_mutex);
 499         obj = com->lc_sub_trace_objs[idx].lsto_obj;
 500         if (unlikely(obj == NULL)) {
 501                 mutex_unlock(&com->lc_sub_trace_objs[idx].lsto_mutex);
 502                 RETURN(0);
 503         }
 504
 505         lfsck_object_get(obj);
 506         dev = lfsck_obj2dev(obj);
 507         fid_cpu_to_be(key, fid);
 508         rc = dt_lookup(env, obj, (struct dt_rec *)&old,
 509                        (const struct dt_key *)key);
 510         if (rc == -ENOENT) {
 511                 if (!add)
 512                         GOTO(unlock, rc = 0);
 513
 514                 old = 0;
 515                 new = flags;
 516         } else if (rc == 0) {
 517                 if (add) {
 518                         if ((old & flags) == flags)
 519                                 GOTO(unlock, rc = 0);
 520
 521                         new = old | flags;
 522                 } else {
 523                         if ((old & flags) == 0)
 524                                 GOTO(unlock, rc = 0);
 525
 526                         new = old & ~flags;
 527                 }
 528         } else {
 529                 GOTO(log, rc);
 530         }
 531
 532         th = dt_trans_create(env, dev);
 533         if (IS_ERR(th))
 534                 GOTO(log, rc = PTR_ERR(th));
 535
 536         if (old != 0) {
 537                 rc = dt_declare_delete(env, obj,
 538                                        (const struct dt_key *)key, th);
 539                 if (rc != 0)
 540                         GOTO(log, rc);
 541         }
 542
 543         if (new != 0) {
 544                 rc = dt_declare_insert(env, obj,
 545                                        (const struct dt_rec *)&new,
 546                                        (const struct dt_key *)key, th);
 547                 if (rc != 0)
 548                         GOTO(log, rc);
 549         }
 550
 551         rc = dt_trans_start_local(env, dev, th);
 552         if (rc != 0)
 553                 GOTO(log, rc);
 554
 555         if (old != 0) {
 556                 rc = dt_delete(env, obj, (const struct dt_key *)key, th);
 557                 if (rc != 0)
 558                         GOTO(log, rc);
 559         }
 560
 561         if (new != 0) {
 562                 rc = dt_insert(env, obj, (const struct dt_rec *)&new,
 563                                (const struct dt_key *)key, th, 1);
 564                 if (rc != 0)
 565                         GOTO(log, rc);
 566         }
 567
 568         GOTO(log, rc);
 569
 570 log:
 571         if (th != NULL && !IS_ERR(th))
 572                 dt_trans_stop(env, dev, th);
 573
 574         CDEBUG(D_LFSCK, "%s: namespace LFSCK %s flags for "DFID" in the "
 575                "trace file, flags %x, old %x, new %x: rc = %d\n",
 576                lfsck_lfsck2name(lfsck), add ? "add" : "del", PFID(fid),
 577                (__u32)flags, (__u32)old, (__u32)new, rc);
 578
 579 unlock:
 580         mutex_unlock(&com->lc_sub_trace_objs[idx].lsto_mutex);
 581         lfsck_object_put(env, obj);
 582
 583         return rc;
 584 }
 585
 586 int lfsck_namespace_check_exist(const struct lu_env *env,
 587                                 struct dt_object *dir,
 588                                 struct dt_object *obj, const char *name)
 589 {
 590         struct lu_fid    *fid = &lfsck_env_info(env)->lti_fid;
 591         int               rc;
 592         ENTRY;
 593
 594         if (unlikely(lfsck_is_dead_obj(obj)))
 595                 RETURN(LFSCK_NAMEENTRY_DEAD);
 596
 597         rc = dt_lookup(env, dir, (struct dt_rec *)fid,
 598                        (const struct dt_key *)name);
 599         if (rc == -ENOENT)
 600                 RETURN(LFSCK_NAMEENTRY_REMOVED);
 601
 602         if (rc < 0)
 603                 RETURN(rc);
 604
 605         if (!lu_fid_eq(fid, lfsck_dto2fid(obj)))
 606                 RETURN(LFSCK_NAMEENTRY_RECREATED);
 607
 608         RETURN(0);
 609 }
 610
 611 static int lfsck_declare_namespace_exec_dir(const struct lu_env *env,
 612                                             struct dt_object *obj,
 613                                             struct thandle *handle)
 614 {
 615         int rc;
 616
 617         /* For remote updating LINKEA, there may be further LFSCK action
 618          * on remote MDT after the updating, so update the LINKEA ASAP. */
 619         if (dt_object_remote(obj))
 620                 handle->th_sync = 1;
 621
 622         /* For destroying all invalid linkEA entries. */
 623         rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, handle);
 624         if (rc == 0)
 625                 /* For insert new linkEA entry. */
 626                 rc = dt_declare_xattr_set(env, obj,
 627                         lfsck_buf_get_const(env, NULL, MAX_LINKEA_SIZE),
 628                         XATTR_NAME_LINK, 0, handle);
 629         return rc;
 630 }
 631
 632 int __lfsck_links_read(const struct lu_env *env, struct dt_object *obj,
 633                        struct linkea_data *ldata, bool with_rec)
 634 {
 635         int rc;
 636
 637         if (ldata->ld_buf->lb_buf == NULL)
 638                 return -ENOMEM;
 639
 640         if (!dt_object_exists(obj))
 641                 return -ENOENT;
 642
 643         rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK);
 644         if (rc == -ERANGE) {
 645                 /* Buf was too small, figure out what we need. */
 646                 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LINK);
 647                 if (unlikely(rc == 0))
 648                         return -ENODATA;
 649
 650                 if (rc < 0)
 651                         return rc;
 652
 653                 lu_buf_realloc(ldata->ld_buf, rc);
 654                 if (ldata->ld_buf->lb_buf == NULL)
 655                         return -ENOMEM;
 656
 657                 rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK);
 658         }
 659
 660         if (unlikely(rc == 0))
 661                 return -ENODATA;
 662
 663         if (rc > 0) {
 664                 if (with_rec)
 665                         rc = linkea_init_with_rec(ldata);
 666                 else
 667                         rc = linkea_init(ldata);
 668         }
 669
 670         return rc;
 671 }
 672
 673 /**
 674  * Remove linkEA for the given object.
 675  *
 676  * The caller should take the ldlm lock before the calling.
 677  *
 678  * \param[in] env       pointer to the thread context
 679  * \param[in] com       pointer to the lfsck component
 680  * \param[in] obj       pointer to the dt_object to be handled
 681  *
 682  * \retval              0 for repaired cases
 683  * \retval              negative error number on failure
 684  */
 685 static int lfsck_namespace_links_remove(const struct lu_env *env,
 686                                         struct lfsck_component *com,
 687                                         struct dt_object *obj)
 688 {
 689         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 690         struct dt_device                *dev    = lfsck_obj2dev(obj);
 691         struct thandle                  *th     = NULL;
 692         int                              rc     = 0;
 693         ENTRY;
 694
 695         LASSERT(dt_object_remote(obj) == 0);
 696
 697         th = dt_trans_create(env, dev);
 698         if (IS_ERR(th))
 699                 GOTO(log, rc = PTR_ERR(th));
 700
 701         rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, th);
 702         if (rc != 0)
 703                 GOTO(stop, rc);
 704
 705         rc = dt_trans_start_local(env, dev, th);
 706         if (rc != 0)
 707                 GOTO(stop, rc);
 708
 709         dt_write_lock(env, obj, 0);
 710         if (unlikely(lfsck_is_dead_obj(obj)))
 711                 GOTO(unlock, rc = -ENOENT);
 712
 713         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
 714                 GOTO(unlock, rc = 0);
 715
 716         rc = dt_xattr_del(env, obj, XATTR_NAME_LINK, th);
 717
 718         GOTO(unlock, rc);
 719
 720 unlock:
 721         dt_write_unlock(env, obj);
 722
 723 stop:
 724         dt_trans_stop(env, dev, th);
 725
 726 log:
 727         CDEBUG(D_LFSCK, "%s: namespace LFSCK remove invalid linkEA "
 728                "for the object "DFID": rc = %d\n",
 729                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
 730
 731         if (rc == 0) {
 732                 struct lfsck_namespace *ns = com->lc_file_ram;
 733
 734                 ns->ln_flags |= LF_INCONSISTENT;
 735         }
 736
 737         return rc;
 738 }
 739
 740 static int lfsck_links_write(const struct lu_env *env, struct dt_object *obj,
 741                              struct linkea_data *ldata, struct thandle *handle)
 742 {
 743         struct lu_buf buf;
 744         int rc;
 745
 746         lfsck_buf_init(&buf, ldata->ld_buf->lb_buf, ldata->ld_leh->leh_len);
 747
 748 again:
 749         rc = dt_xattr_set(env, obj, &buf, XATTR_NAME_LINK, 0, handle);
 750         if (unlikely(rc == -ENOSPC)) {
 751                 rc = linkea_overflow_shrink(ldata);
 752                 if (likely(rc > 0)) {
 753                         buf.lb_len = rc;
 754                         goto again;
 755                 }
 756         }
 757
 758         return rc;
 759 }
 760
 761 static int lfsck_namespace_unpack_linkea_entry(struct linkea_data *ldata,
 762                                                struct lu_name *cname,
 763                                                struct lu_fid *pfid,
 764                                                char *buf, const int buflen)
 765 {
 766         linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, cname, pfid);
 767         if (unlikely(ldata->ld_reclen <= 0 ||
 768                      ldata->ld_reclen + sizeof(struct link_ea_header) >
 769                         ldata->ld_leh->leh_len ||
 770                      cname->ln_namelen <= 0 ||
 771                      cname->ln_namelen > NAME_MAX ||
 772                      cname->ln_namelen >= buflen ||
 773                      !fid_is_sane(pfid)))
 774                 return -EINVAL;
 775
 776         /* To guarantee the 'name' is terminated with '0'. */
 777         memcpy(buf, cname->ln_name, cname->ln_namelen);
 778         buf[cname->ln_namelen] = 0;
 779         cname->ln_name = buf;
 780
 781         return 0;
 782 }
 783
 784 static void lfsck_linkea_del_buf(struct linkea_data *ldata,
 785                                  const struct lu_name *lname)
 786 {
 787         LASSERT(ldata->ld_leh != NULL && ldata->ld_lee != NULL);
 788
 789         /* If current record is corrupted, all the subsequent
 790          * records will be dropped. */
 791         if (unlikely(ldata->ld_reclen <= 0 ||
 792                      ldata->ld_reclen + sizeof(struct link_ea_header) >
 793                         ldata->ld_leh->leh_len)) {
 794                 void *ptr = ldata->ld_lee;
 795
 796                 ldata->ld_leh->leh_len = sizeof(struct link_ea_header);
 797                 ldata->ld_leh->leh_reccount = 0;
 798                 linkea_first_entry(ldata);
 799                 while (ldata->ld_lee != NULL &&
 800                        (char *)ldata->ld_lee < (char *)ptr) {
 801                         int reclen = (ldata->ld_lee->lee_reclen[0] << 8) |
 802                                      ldata->ld_lee->lee_reclen[1];
 803
 804                         ldata->ld_leh->leh_len += reclen;
 805                         ldata->ld_leh->leh_reccount++;
 806                         ldata->ld_lee = (struct link_ea_entry *)
 807                                         ((char *)ldata->ld_lee + reclen);
 808                 }
 809
 810                 ldata->ld_lee = NULL;
 811         } else {
 812                 linkea_del_buf(ldata, lname);
 813         }
 814 }
 815
 816 static int lfsck_namespace_filter_linkea_entry(struct linkea_data *ldata,
 817                                                struct lu_name *cname,
 818                                                struct lu_fid *pfid,
 819                                                bool remove)
 820 {
 821         struct link_ea_entry    *oldlee;
 822         int                      oldlen;
 823         int                      repeated = 0;
 824
 825         oldlee = ldata->ld_lee;
 826         oldlen = ldata->ld_reclen;
 827         linkea_next_entry(ldata);
 828         while (ldata->ld_lee != NULL) {
 829                 ldata->ld_reclen = (ldata->ld_lee->lee_reclen[0] << 8) |
 830                                    ldata->ld_lee->lee_reclen[1];
 831                 if (unlikely(ldata->ld_reclen == oldlen &&
 832                              memcmp(ldata->ld_lee, oldlee, oldlen) == 0)) {
 833                         repeated++;
 834                         if (!remove)
 835                                 break;
 836
 837                         lfsck_linkea_del_buf(ldata, cname);
 838                 } else {
 839                         linkea_next_entry(ldata);
 840                 }
 841         }
 842         ldata->ld_lee = oldlee;
 843         ldata->ld_reclen = oldlen;
 844
 845         return repeated;
 846 }
 847
 848 /**
 849  * Insert orphan into .lustre/lost+found/MDTxxxx/ locally.
 850  *
 851  * Add the specified orphan MDT-object to the .lustre/lost+found/MDTxxxx/
 852  * with the given type to generate the name, the detailed rules for name
 853  * have been described as following.
 854  *
 855  * The function also generates the linkEA corresponding to the name entry
 856  * under the .lustre/lost+found/MDTxxxx/ for the orphan MDT-object.
 857  *
 858  * \param[in] env       pointer to the thread context
 859  * \param[in] com       pointer to the lfsck component
 860  * \param[in] orphan    pointer to the orphan MDT-object
 861  * \param[in] infix     additional information for the orphan name, such as
 862  *                      the FID for original
 863  * \param[in] type      the type for describing why the orphan MDT-object is
 864  *                      created. The rules are as following:
 865  *
 866  *  type "D":           The MDT-object is a directory, it may knows its parent
 867  *                      but because there is no valid linkEA, the LFSCK cannot
 868  *                      know where to put it back to the namespace.
 869  *  type "O":           The MDT-object has no linkEA, and there is no name
 870  *                      entry that references the MDT-object.
 871  *
 872  *  type "S":           The orphan MDT-object is a shard of a striped directory
 873  *
 874  * \see lfsck_layout_recreate_parent() for more types.
 875  *
 876  * The orphan name will be like:
 877  * ${FID}-${infix}-${type}-${conflict_version}
 878  *
 879  * \param[out] count    if some others inserted some linkEA entries by race,
 880  *                      then return the linkEA entries count.
 881  *
 882  * \retval              positive number for repaired cases
 883  * \retval              0 if needs to repair nothing
 884  * \retval              negative error number on failure
 885  */
 886 static int lfsck_namespace_insert_orphan(const struct lu_env *env,
 887                                          struct lfsck_component *com,
 888                                          struct dt_object *orphan,
 889                                          const char *infix, const char *type,
 890                                          int *count)
 891 {
 892         struct lfsck_thread_info        *info   = lfsck_env_info(env);
 893         struct lu_name                  *cname  = &info->lti_name;
 894         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
 895         struct lu_attr                  *la     = &info->lti_la2;
 896         const struct lu_fid             *cfid   = lfsck_dto2fid(orphan);
 897         const struct lu_fid             *pfid;
 898         struct lu_fid                    tfid;
 899         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 900         struct dt_device                *dev    = lfsck_obj2dev(orphan);
 901         struct dt_object                *parent;
 902         struct thandle                  *th     = NULL;
 903         struct lfsck_lock_handle        *pllh   = &info->lti_llh;
 904         struct lustre_handle             clh    = { 0 };
 905         struct linkea_data               ldata2 = { NULL };
 906         struct lu_buf                    linkea_buf;
 907         int                              namelen;
 908         int                              idx    = 0;
 909         int                              rc     = 0;
 910         bool                             exist  = false;
 911         ENTRY;
 912
 913         cname->ln_name = NULL;
 914         if (unlikely(lfsck->li_lpf_obj == NULL))
 915                 GOTO(log, rc = -ENXIO);
 916
 917         parent = lfsck->li_lpf_obj;
 918         pfid = lfsck_dto2fid(parent);
 919
 920 again:
 921         do {
 922                 namelen = snprintf(info->lti_key, NAME_MAX, DFID"%s-%s-%d",
 923                                    PFID(cfid), infix, type, idx++);
 924                 rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
 925                                (const struct dt_key *)info->lti_key);
 926                 if (rc != 0 && rc != -ENOENT)
 927                         GOTO(log, rc);
 928
 929                 if (unlikely(rc == 0 && lu_fid_eq(cfid, &tfid)))
 930                         exist = true;
 931         } while (rc == 0 && !exist);
 932
 933         rc = lfsck_lock(env, lfsck, parent, info->lti_key, pllh,
 934                         MDS_INODELOCK_UPDATE, LCK_PW);
 935         if (rc != 0)
 936                 GOTO(log, rc);
 937
 938         /* Re-check whether the name conflict with othrs after taken
 939          * the ldlm lock. */
 940         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
 941                        (const struct dt_key *)info->lti_key);
 942         if (rc == 0) {
 943                 if (!lu_fid_eq(cfid, &tfid)) {
 944                         exist = false;
 945                         lfsck_unlock(pllh);
 946                         goto again;
 947                 }
 948
 949                 exist = true;
 950         } else if (rc != -ENOENT) {
 951                 GOTO(log, rc);
 952         } else {
 953                 exist = false;
 954         }
 955
 956         cname->ln_name = info->lti_key;
 957         cname->ln_namelen = namelen;
 958         rc = linkea_links_new(&ldata2, &info->lti_linkea_buf2,
 959                               cname, pfid);
 960         if (rc != 0)
 961                 GOTO(log, rc);
 962
 963         rc = lfsck_ibits_lock(env, lfsck, orphan, &clh,
 964                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_LOOKUP |
 965                               MDS_INODELOCK_XATTR, LCK_EX);
 966         if (rc != 0)
 967                 GOTO(log, rc);
 968
 969         lfsck_buf_init(&linkea_buf, ldata2.ld_buf->lb_buf,
 970                        ldata2.ld_leh->leh_len);
 971         th = dt_trans_create(env, dev);
 972         if (IS_ERR(th))
 973                 GOTO(log, rc = PTR_ERR(th));
 974
 975         if (S_ISDIR(lfsck_object_type(orphan))) {
 976                 rc = dt_declare_delete(env, orphan,
 977                                        (const struct dt_key *)dotdot, th);
 978                 if (rc != 0)
 979                         GOTO(stop, rc);
 980
 981                 rec->rec_type = S_IFDIR;
 982                 rec->rec_fid = pfid;
 983                 rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,
 984                                        (const struct dt_key *)dotdot, th);
 985                 if (rc != 0)
 986                         GOTO(stop, rc);
 987         }
 988
 989         rc = dt_declare_xattr_set(env, orphan, &linkea_buf,
 990                                   XATTR_NAME_LINK, 0, th);
 991         if (rc != 0)
 992                 GOTO(stop, rc);
 993
 994         if (!exist) {
 995                 rec->rec_type = lfsck_object_type(orphan) & S_IFMT;
 996                 rec->rec_fid = cfid;
 997                 rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
 998                                        (const struct dt_key *)cname->ln_name,
 999                                        th);
1000                 if (rc != 0)
1001                         GOTO(stop, rc);
1002
1003                 if (S_ISDIR(rec->rec_type)) {
1004                         rc = dt_declare_ref_add(env, parent, th);
1005                         if (rc != 0)
1006                                 GOTO(stop, rc);
1007                 }
1008         }
1009
1010         memset(la, 0, sizeof(*la));
1011         la->la_ctime = ktime_get_real_seconds();
1012         la->la_valid = LA_CTIME;
1013         rc = dt_declare_attr_set(env, orphan, la, th);
1014         if (rc != 0)
1015                 GOTO(stop, rc);
1016
1017         rc = dt_trans_start_local(env, dev, th);
1018         if (rc != 0)
1019                 GOTO(stop, rc);
1020
1021         dt_write_lock(env, orphan, 0);
1022         rc = lfsck_links_read2_with_rec(env, orphan, &ldata2);
1023         if (likely(rc == -ENODATA || rc == -EINVAL)) {
1024                 if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1025                         GOTO(unlock, rc = 1);
1026
1027                 if (S_ISDIR(lfsck_object_type(orphan))) {
1028                         rc = dt_delete(env, orphan,
1029                                        (const struct dt_key *)dotdot, th);
1030                         if (rc != 0)
1031                                 GOTO(unlock, rc);
1032
1033                         rec->rec_type = S_IFDIR;
1034                         rec->rec_fid = pfid;
1035                         rc = dt_insert(env, orphan, (const struct dt_rec *)rec,
1036                                        (const struct dt_key *)dotdot, th, 1);
1037                         if (rc != 0)
1038                                 GOTO(unlock, rc);
1039                 }
1040
1041                 rc = dt_xattr_set(env, orphan, &linkea_buf, XATTR_NAME_LINK, 0,
1042                                   th);
1043         } else {
1044                 if (rc == 0 && count != NULL)
1045                         *count = ldata2.ld_leh->leh_reccount;
1046
1047                 GOTO(unlock, rc);
1048         }
1049         dt_write_unlock(env, orphan);
1050
1051         if (rc == 0 && !exist) {
1052                 rec->rec_type = lfsck_object_type(orphan) & S_IFMT;
1053                 rec->rec_fid = cfid;
1054                 rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1055                                (const struct dt_key *)cname->ln_name, th, 1);
1056                 if (rc == 0 && S_ISDIR(rec->rec_type)) {
1057                         dt_write_lock(env, parent, 0);
1058                         rc = dt_ref_add(env, parent, th);
1059                         dt_write_unlock(env, parent);
1060                 }
1061         }
1062
1063         if (rc == 0)
1064                 rc = dt_attr_set(env, orphan, la, th);
1065
1066         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1067
1068 unlock:
1069         dt_write_unlock(env, orphan);
1070
1071 stop:
1072         dt_trans_stop(env, dev, th);
1073
1074 log:
1075         lfsck_ibits_unlock(&clh, LCK_EX);
1076         lfsck_unlock(pllh);
1077         CDEBUG(D_LFSCK, "%s: namespace LFSCK insert orphan for the "
1078                "object "DFID", name = %s: rc = %d\n",
1079                lfsck_lfsck2name(lfsck), PFID(cfid),
1080                cname->ln_name != NULL ? cname->ln_name : "<NULL>", rc);
1081
1082         if (rc != 0) {
1083                 struct lfsck_namespace *ns = com->lc_file_ram;
1084
1085                 ns->ln_flags |= LF_INCONSISTENT;
1086         }
1087
1088         return rc;
1089 }
1090
1091 /**
1092  * Add the specified name entry back to namespace.
1093  *
1094  * If there is a linkEA entry that back references a name entry under
1095  * some parent directory, but such parent directory does not have the
1096  * claimed name entry. On the other hand, the linkEA entries count is
1097  * not larger than the MDT-object's hard link count. Under such case,
1098  * it is quite possible that the name entry is lost. Then the LFSCK
1099  * should add the name entry back to the namespace.
1100  *
1101  * \param[in] env       pointer to the thread context
1102  * \param[in] com       pointer to the lfsck component
1103  * \param[in] parent    pointer to the directory under which the name entry
1104  *                      will be inserted into
1105  * \param[in] child     pointer to the object referenced by the name entry
1106  *                      that to be inserted into the parent
1107  * \param[in] name      the name for the child in the parent directory
1108  *
1109  * \retval              positive number for repaired cases
1110  * \retval              0 if nothing to be repaired
1111  * \retval              negative error number on failure
1112  */
1113 static int lfsck_namespace_insert_normal(const struct lu_env *env,
1114                                          struct lfsck_component *com,
1115                                          struct dt_object *parent,
1116                                          struct dt_object *child,
1117                                          const char *name)
1118 {
1119         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1120         struct lu_attr                  *la     = &info->lti_la;
1121         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
1122         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1123         /* The child and its name may be on different MDTs. */
1124         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
1125         const struct lu_fid             *cfid   = lfsck_dto2fid(child);
1126         struct dt_device                *dev    = lfsck->li_next;
1127         struct thandle                  *th     = NULL;
1128         struct lfsck_lock_handle        *llh    = &info->lti_llh;
1129         int                              rc     = 0;
1130         ENTRY;
1131
1132         /* @parent/@child may be based on lfsck->li_bottom,
1133          * but here we need the object based on the lfsck->li_next. */
1134
1135         parent = lfsck_object_locate(dev, parent);
1136         if (IS_ERR(parent))
1137                 GOTO(log, rc = PTR_ERR(parent));
1138
1139         if (unlikely(!dt_try_as_dir(env, parent)))
1140                 GOTO(log, rc = -ENOTDIR);
1141
1142         child = lfsck_object_locate(dev, child);
1143         if (IS_ERR(child))
1144                 GOTO(log, rc = PTR_ERR(child));
1145
1146         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1147                 GOTO(log, rc = 1);
1148
1149         rc = lfsck_lock(env, lfsck, parent, name, llh,
1150                         MDS_INODELOCK_UPDATE, LCK_PW);
1151         if (rc != 0)
1152                 GOTO(log, rc);
1153
1154         th = dt_trans_create(env, dev);
1155         if (IS_ERR(th))
1156                 GOTO(unlock, rc = PTR_ERR(th));
1157
1158         rec->rec_type = lfsck_object_type(child) & S_IFMT;
1159         rec->rec_fid = cfid;
1160         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
1161                                (const struct dt_key *)name, th);
1162         if (rc != 0)
1163                 GOTO(stop, rc);
1164
1165         if (S_ISDIR(rec->rec_type)) {
1166                 rc = dt_declare_ref_add(env, parent, th);
1167                 if (rc != 0)
1168                         GOTO(stop, rc);
1169         }
1170
1171         memset(la, 0, sizeof(*la));
1172         la->la_ctime = ktime_get_real_seconds();
1173         la->la_valid = LA_CTIME;
1174         rc = dt_declare_attr_set(env, parent, la, th);
1175         if (rc != 0)
1176                 GOTO(stop, rc);
1177
1178         rc = dt_declare_attr_set(env, child, la, th);
1179         if (rc != 0)
1180                 GOTO(stop, rc);
1181
1182         rc = dt_trans_start_local(env, dev, th);
1183         if (rc != 0)
1184                 GOTO(stop, rc);
1185
1186         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1187                        (const struct dt_key *)name, th, 1);
1188         if (rc != 0)
1189                 GOTO(stop, rc);
1190
1191         if (S_ISDIR(rec->rec_type)) {
1192                 dt_write_lock(env, parent, 0);
1193                 rc = dt_ref_add(env, parent, th);
1194                 dt_write_unlock(env, parent);
1195                 if (rc != 0)
1196                         GOTO(stop, rc);
1197         }
1198
1199         la->la_ctime = ktime_get_real_seconds();
1200         rc = dt_attr_set(env, parent, la, th);
1201         if (rc != 0)
1202                 GOTO(stop, rc);
1203
1204         rc = dt_attr_set(env, child, la, th);
1205
1206         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1207
1208 stop:
1209         dt_trans_stop(env, dev, th);
1210
1211 unlock:
1212         lfsck_unlock(llh);
1213
1214 log:
1215         CDEBUG(D_LFSCK, "%s: namespace LFSCK insert object "DFID" with "
1216                "the name %s and type %o to the parent "DFID": rc = %d\n",
1217                lfsck_lfsck2name(lfsck), PFID(cfid), name,
1218                lfsck_object_type(child) & S_IFMT, PFID(pfid), rc);
1219
1220         if (rc != 0) {
1221                 struct lfsck_namespace *ns = com->lc_file_ram;
1222
1223                 ns->ln_flags |= LF_INCONSISTENT;
1224                 if (rc > 0)
1225                         ns->ln_lost_dirent_repaired++;
1226         }
1227
1228         return rc;
1229 }
1230
1231 /**
1232  * Create the specified orphan directory.
1233  *
1234  * For the case that the parent MDT-object stored in some MDT-object's
1235  * linkEA entry is lost, the LFSCK will re-create the parent object as
1236  * an orphan and insert it into .lustre/lost+found/MDTxxxx/ directory
1237  * with the name ${FID}-P-${conflict_version}.
1238  *
1239  * \param[in] env       pointer to the thread context
1240  * \param[in] com       pointer to the lfsck component
1241  * \param[in] orphan    pointer to the orphan MDT-object to be created
1242  * \param[in] lmv       pointer to master LMV EA that will be set to the orphan
1243  *
1244  * \retval              positive number for repaired cases
1245  * \retval              negative error number on failure
1246  */
1247 static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
1248                                              struct lfsck_component *com,
1249                                              struct dt_object *orphan,
1250                                              struct lmv_mds_md_v1 *lmv)
1251 {
1252         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1253         struct lu_attr                  *la     = &info->lti_la;
1254         struct dt_allocation_hint       *hint   = &info->lti_hint;
1255         struct dt_object_format         *dof    = &info->lti_dof;
1256         struct lu_name                  *cname  = &info->lti_name2;
1257         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
1258         struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
1259         const struct lu_fid             *cfid   = lfsck_dto2fid(orphan);
1260         struct lu_fid                    tfid;
1261         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1262         struct lfsck_namespace          *ns     = com->lc_file_ram;
1263         struct dt_device                *dev    = lfsck_obj2dev(orphan);
1264         struct dt_object                *parent = NULL;
1265         struct thandle                  *th     = NULL;
1266         struct lfsck_lock_handle        *llh    = &info->lti_llh;
1267         struct linkea_data               ldata  = { NULL };
1268         struct lu_buf                    linkea_buf;
1269         struct lu_buf                    lmv_buf;
1270         char                             name[32];
1271         int                              namelen;
1272         int                              idx    = 0;
1273         int                              rc     = 0;
1274         int                              rc1    = 0;
1275         ENTRY;
1276
1277         LASSERT(!dt_object_exists(orphan));
1278
1279         cname->ln_name = NULL;
1280         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1281                 GOTO(log, rc = 1);
1282
1283         if (dt_object_remote(orphan)) {
1284                 LASSERT(lfsck->li_lpf_root_obj != NULL);
1285
1286                 idx = lfsck_find_mdt_idx_by_fid(env, lfsck, cfid);
1287                 if (idx < 0)
1288                         GOTO(log, rc = idx);
1289
1290                 snprintf(name, 8, "MDT%04x", idx);
1291                 rc = dt_lookup(env, lfsck->li_lpf_root_obj,
1292                                (struct dt_rec *)&tfid,
1293                                (const struct dt_key *)name);
1294                 if (rc != 0)
1295                         GOTO(log, rc = (rc == -ENOENT ? -ENXIO : rc));
1296
1297                 parent = lfsck_object_find_bottom(env, lfsck, &tfid);
1298                 if (IS_ERR(parent))
1299                         GOTO(log, rc = PTR_ERR(parent));
1300
1301                 if (unlikely(!dt_try_as_dir(env, parent)))
1302                         GOTO(log, rc = -ENOTDIR);
1303         } else {
1304                 if (unlikely(lfsck->li_lpf_obj == NULL))
1305                         GOTO(log, rc = -ENXIO);
1306
1307                 parent = lfsck->li_lpf_obj;
1308         }
1309
1310         dev = lfsck_find_dev_by_fid(env, lfsck, cfid);
1311         if (IS_ERR(dev))
1312                 GOTO(log, rc = PTR_ERR(dev));
1313
1314         idx = 0;
1315
1316 again:
1317         do {
1318                 namelen = snprintf(name, 31, DFID"-P-%d",
1319                                    PFID(cfid), idx++);
1320                 rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1321                                (const struct dt_key *)name);
1322                 if (rc != 0 && rc != -ENOENT)
1323                         GOTO(log, rc);
1324         } while (rc == 0);
1325
1326         rc = lfsck_lock(env, lfsck, parent, name, llh,
1327                         MDS_INODELOCK_UPDATE, LCK_PW);
1328         if (rc != 0)
1329                 GOTO(log, rc);
1330
1331         /* Re-check whether the name conflict with othrs after taken
1332          * the ldlm lock. */
1333         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1334                        (const struct dt_key *)name);
1335         if (unlikely(rc == 0)) {
1336                 lfsck_unlock(llh);
1337                 goto again;
1338         }
1339
1340         if (rc != -ENOENT)
1341                 GOTO(unlock1, rc);
1342
1343         cname->ln_name = name;
1344         cname->ln_namelen = namelen;
1345
1346         memset(la, 0, sizeof(*la));
1347         la->la_mode = S_IFDIR | 0700;
1348         la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
1349                        LA_ATIME | LA_MTIME | LA_CTIME;
1350
1351         orphan->do_ops->do_ah_init(env, hint, parent, orphan,
1352                                    la->la_mode & S_IFMT);
1353
1354         memset(dof, 0, sizeof(*dof));
1355         dof->dof_type = dt_mode_to_dft(S_IFDIR);
1356
1357         rc = linkea_links_new(&ldata, &info->lti_linkea_buf2,
1358                               cname, lfsck_dto2fid(parent));
1359         if (rc != 0)
1360                 GOTO(unlock1, rc);
1361
1362         th = dt_trans_create(env, dev);
1363         if (IS_ERR(th))
1364                 GOTO(unlock1, rc = PTR_ERR(th));
1365
1366         /* Sync the remote transaction to guarantee that the subsequent
1367          * lock against the @orphan can find the @orphan in time. */
1368         if (dt_object_remote(orphan))
1369                 th->th_sync = 1;
1370
1371         rc = dt_declare_create(env, orphan, la, hint, dof, th);
1372         if (rc != 0)
1373                 GOTO(stop, rc);
1374
1375         if (unlikely(!dt_try_as_dir(env, orphan)))
1376                 GOTO(stop, rc = -ENOTDIR);
1377
1378         rc = dt_declare_ref_add(env, orphan, th);
1379         if (rc != 0)
1380                 GOTO(stop, rc);
1381
1382         rec->rec_type = S_IFDIR;
1383         rec->rec_fid = cfid;
1384         rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,
1385                                (const struct dt_key *)dot, th);
1386         if (rc != 0)
1387                 GOTO(stop, rc);
1388
1389         rec->rec_fid = lfsck_dto2fid(parent);
1390         rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,
1391                                (const struct dt_key *)dotdot, th);
1392         if (rc != 0)
1393                 GOTO(stop, rc);
1394
1395         if (lmv != NULL) {
1396                 lmv->lmv_magic = LMV_MAGIC;
1397                 lmv->lmv_master_mdt_index = lfsck_dev_idx(lfsck);
1398                 lfsck_lmv_header_cpu_to_le(lmv2, lmv);
1399                 lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
1400                 rc = dt_declare_xattr_set(env, orphan, &lmv_buf,
1401                                           XATTR_NAME_LMV, 0, th);
1402                 if (rc != 0)
1403                         GOTO(stop, rc);
1404         }
1405
1406         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
1407                        ldata.ld_leh->leh_len);
1408         rc = dt_declare_xattr_set(env, orphan, &linkea_buf,
1409                                   XATTR_NAME_LINK, 0, th);
1410         if (rc != 0)
1411                 GOTO(stop, rc);
1412
1413         rec->rec_fid = cfid;
1414         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
1415                                (const struct dt_key *)name, th);
1416         if (rc == 0)
1417                 rc = dt_declare_ref_add(env, parent, th);
1418
1419         if (rc != 0)
1420                 GOTO(stop, rc);
1421
1422         rc = dt_trans_start_local(env, dev, th);
1423         if (rc != 0)
1424                 GOTO(stop, rc);
1425
1426         dt_write_lock(env, orphan, 0);
1427         rc = dt_create(env, orphan, la, hint, dof, th);
1428         if (rc != 0)
1429                 GOTO(unlock2, rc);
1430
1431         rc = dt_ref_add(env, orphan, th);
1432         if (rc != 0)
1433                 GOTO(unlock2, rc);
1434
1435         rec->rec_fid = cfid;
1436         rc = dt_insert(env, orphan, (const struct dt_rec *)rec,
1437                        (const struct dt_key *)dot, th, 1);
1438         if (rc != 0)
1439                 GOTO(unlock2, rc);
1440
1441         rec->rec_fid = lfsck_dto2fid(parent);
1442         rc = dt_insert(env, orphan, (const struct dt_rec *)rec,
1443                        (const struct dt_key *)dotdot, th, 1);
1444         if (rc != 0)
1445                 GOTO(unlock2, rc);
1446
1447         if (lmv != NULL) {
1448                 rc = dt_xattr_set(env, orphan, &lmv_buf, XATTR_NAME_LMV, 0, th);
1449                 if (rc != 0)
1450                         GOTO(unlock2, rc);
1451         }
1452
1453         rc = dt_xattr_set(env, orphan, &linkea_buf,
1454                           XATTR_NAME_LINK, 0, th);
1455         dt_write_unlock(env, orphan);
1456         if (rc != 0)
1457                 GOTO(stop, rc);
1458
1459         rec->rec_fid = cfid;
1460         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1461                        (const struct dt_key *)name, th, 1);
1462         if (rc == 0) {
1463                 dt_write_lock(env, parent, 0);
1464                 rc = dt_ref_add(env, parent, th);
1465                 dt_write_unlock(env, parent);
1466         }
1467
1468         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1469
1470 unlock2:
1471         dt_write_unlock(env, orphan);
1472
1473 stop:
1474         rc1 = dt_trans_stop(env, dev, th);
1475         if (rc1 != 0 && rc > 0)
1476                 rc = rc1;
1477
1478 unlock1:
1479         lfsck_unlock(llh);
1480
1481 log:
1482         CDEBUG(D_LFSCK, "%s: namespace LFSCK create orphan dir for "
1483                "the object "DFID", name = %s: rc = %d\n",
1484                lfsck_lfsck2name(lfsck), PFID(cfid),
1485                cname->ln_name != NULL ? cname->ln_name : "<NULL>", rc);
1486
1487         if (parent != NULL && !IS_ERR(parent) && parent != lfsck->li_lpf_obj)
1488                 lfsck_object_put(env, parent);
1489
1490         if (rc != 0)
1491                 ns->ln_flags |= LF_INCONSISTENT;
1492
1493         return rc;
1494 }
1495
1496 /**
1497  * Remove the specified entry from the linkEA.
1498  *
1499  * Locate the linkEA entry with the given @cname and @pfid, then
1500  * remove this entry or the other entries those are repeated with
1501  * this entry.
1502  *
1503  * \param[in] env       pointer to the thread context
1504  * \param[in] com       pointer to the lfsck component
1505  * \param[in] obj       pointer to the dt_object to be handled
1506  * \param[in,out]ldata  pointer to the buffer that holds the linkEA
1507  * \param[in] cname     the name for the child in the parent directory
1508  * \param[in] pfid      the parent directory's FID for the linkEA
1509  * \param[in] next      if true, then remove the first found linkEA
1510  *                      entry, and move the ldata->ld_lee to next entry
1511  *
1512  * \retval              positive number for repaired cases
1513  * \retval              0 if nothing to be repaired
1514  * \retval              negative error number on failure
1515  */
1516 static int lfsck_namespace_shrink_linkea(const struct lu_env *env,
1517                                          struct lfsck_component *com,
1518                                          struct dt_object *obj,
1519                                          struct linkea_data *ldata,
1520                                          struct lu_name *cname,
1521                                          struct lu_fid *pfid,
1522                                          bool next)
1523 {
1524         struct lfsck_instance           *lfsck     = com->lc_lfsck;
1525         struct dt_device                *dev       = lfsck_obj2dev(obj);
1526         struct lfsck_bookmark           *bk        = &lfsck->li_bookmark_ram;
1527         struct thandle                  *th        = NULL;
1528         struct lustre_handle             lh        = { 0 };
1529         struct linkea_data               ldata_new = { NULL };
1530         struct lu_buf                    linkea_buf;
1531         int                              buflen    = 0;
1532         int                              rc        = 0;
1533         ENTRY;
1534
1535         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
1536                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
1537                               LCK_EX);
1538         if (rc != 0)
1539                 GOTO(log, rc);
1540
1541         if (next)
1542                 lfsck_linkea_del_buf(ldata, cname);
1543         else
1544                 lfsck_namespace_filter_linkea_entry(ldata, cname, pfid,
1545                                                     true);
1546         if (ldata->ld_leh->leh_reccount > 0 ||
1547             unlikely(ldata->ld_leh->leh_overflow_time)) {
1548                 lfsck_buf_init(&linkea_buf, ldata->ld_buf->lb_buf,
1549                                ldata->ld_leh->leh_len);
1550                 buflen = linkea_buf.lb_len;
1551         }
1552
1553 again:
1554         th = dt_trans_create(env, dev);
1555         if (IS_ERR(th))
1556                 GOTO(unlock1, rc = PTR_ERR(th));
1557
1558         if (buflen != 0)
1559                 rc = dt_declare_xattr_set(env, obj, &linkea_buf,
1560                                           XATTR_NAME_LINK, 0, th);
1561         else
1562                 rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, th);
1563         if (rc != 0)
1564                 GOTO(stop, rc);
1565
1566         rc = dt_trans_start_local(env, dev, th);
1567         if (rc != 0)
1568                 GOTO(stop, rc);
1569
1570         dt_write_lock(env, obj, 0);
1571         if (unlikely(lfsck_is_dead_obj(obj)))
1572                 GOTO(unlock2, rc = -ENOENT);
1573
1574         rc = lfsck_links_read2_with_rec(env, obj, &ldata_new);
1575         if (rc)
1576                 GOTO(unlock2, rc = (rc == -ENODATA ? 0 : rc));
1577
1578         /* The specified linkEA entry has been removed by race. */
1579         rc = linkea_links_find(&ldata_new, cname, pfid);
1580         if (rc != 0)
1581                 GOTO(unlock2, rc = 0);
1582
1583         if (bk->lb_param & LPF_DRYRUN)
1584                 GOTO(unlock2, rc = 1);
1585
1586         if (next)
1587                 lfsck_linkea_del_buf(&ldata_new, cname);
1588         else
1589                 lfsck_namespace_filter_linkea_entry(&ldata_new, cname, pfid,
1590                                                     true);
1591
1592         if (buflen < ldata_new.ld_leh->leh_len) {
1593                 dt_write_unlock(env, obj);
1594                 dt_trans_stop(env, dev, th);
1595                 lfsck_buf_init(&linkea_buf, ldata_new.ld_buf->lb_buf,
1596                                ldata_new.ld_leh->leh_len);
1597                 goto again;
1598         }
1599
1600         if (ldata_new.ld_leh->leh_reccount > 0 ||
1601             unlikely(ldata->ld_leh->leh_overflow_time))
1602                 rc = lfsck_links_write(env, obj, &ldata_new, th);
1603         else
1604                 rc = dt_xattr_del(env, obj, XATTR_NAME_LINK, th);
1605
1606         GOTO(unlock2, rc = (rc == 0 ? 1 : rc));
1607
1608 unlock2:
1609         dt_write_unlock(env, obj);
1610
1611 stop:
1612         dt_trans_stop(env, dev, th);
1613
1614 unlock1:
1615         lfsck_ibits_unlock(&lh, LCK_EX);
1616
1617 log:
1618         CDEBUG(D_LFSCK, "%s: namespace LFSCK remove %s linkEA entry "
1619                "for the object: "DFID", parent "DFID", name %.*s\n",
1620                lfsck_lfsck2name(lfsck), next ? "invalid" : "redundant",
1621                PFID(lfsck_dto2fid(obj)), PFID(pfid), cname->ln_namelen,
1622                cname->ln_name);
1623
1624         if (rc != 0) {
1625                 struct lfsck_namespace *ns = com->lc_file_ram;
1626
1627                 ns->ln_flags |= LF_INCONSISTENT;
1628         }
1629
1630         return rc;
1631 }
1632
1633 /**
1634  * Conditionally remove the specified entry from the linkEA.
1635  *
1636  * Take the parent lock firstly, then check whether the specified
1637  * name entry exists or not: if yes, do nothing; otherwise, call
1638  * lfsck_namespace_shrink_linkea() to remove the linkea entry.
1639  *
1640  * \param[in] env       pointer to the thread context
1641  * \param[in] com       pointer to the lfsck component
1642  * \param[in] parent    pointer to the parent directory
1643  * \param[in] child     pointer to the child object that holds the linkEA
1644  * \param[in,out]ldata  pointer to the buffer that holds the linkEA
1645  * \param[in] cname     the name for the child in the parent directory
1646  * \param[in] pfid      the parent directory's FID for the linkEA
1647  *
1648  * \retval              positive number for repaired cases
1649  * \retval              0 if nothing to be repaired
1650  * \retval              negative error number on failure
1651  */
1652 static int lfsck_namespace_shrink_linkea_cond(const struct lu_env *env,
1653                                               struct lfsck_component *com,
1654                                               struct dt_object *parent,
1655                                               struct dt_object *child,
1656                                               struct linkea_data *ldata,
1657                                               struct lu_name *cname,
1658                                               struct lu_fid *pfid)
1659 {
1660         struct lfsck_thread_info *info  = lfsck_env_info(env);
1661         struct lu_fid            *cfid  = &info->lti_fid3;
1662         struct lfsck_lock_handle *llh   = &info->lti_llh;
1663         int                       rc;
1664         ENTRY;
1665
1666         rc = lfsck_lock(env, com->lc_lfsck, parent, cname->ln_name, llh,
1667                         MDS_INODELOCK_UPDATE, LCK_PR);
1668         if (rc != 0)
1669                 RETURN(rc);
1670
1671         dt_read_lock(env, parent, 0);
1672         if (unlikely(lfsck_is_dead_obj(parent))) {
1673                 dt_read_unlock(env, parent);
1674                 lfsck_unlock(llh);
1675                 rc = lfsck_namespace_shrink_linkea(env, com, child, ldata,
1676                                                    cname, pfid, true);
1677
1678                 RETURN(rc);
1679         }
1680
1681         rc = dt_lookup(env, parent, (struct dt_rec *)cfid,
1682                        (const struct dt_key *)cname->ln_name);
1683         dt_read_unlock(env, parent);
1684
1685         /* It is safe to release the ldlm lock, because when the logic come
1686          * here, we have got all the needed information above whether the
1687          * linkEA entry is valid or not. It is not important that others
1688          * may add new linkEA entry after the ldlm lock released. If other
1689          * has removed the specified linkEA entry by race, then it is OK,
1690          * because the subsequent lfsck_namespace_shrink_linkea() can handle
1691          * such case. */
1692         lfsck_unlock(llh);
1693         if (rc == -ENOENT) {
1694                 rc = lfsck_namespace_shrink_linkea(env, com, child, ldata,
1695                                                    cname, pfid, true);
1696
1697                 RETURN(rc);
1698         }
1699
1700         if (rc != 0)
1701                 RETURN(rc);
1702
1703         /* The LFSCK just found some internal status of cross-MDTs
1704          * create operation. That is normal. */
1705         if (lu_fid_eq(cfid, lfsck_dto2fid(child))) {
1706                 linkea_next_entry(ldata);
1707
1708                 RETURN(0);
1709         }
1710
1711         rc = lfsck_namespace_shrink_linkea(env, com, child, ldata, cname,
1712                                            pfid, true);
1713
1714         RETURN(rc);
1715 }
1716
1717 /**
1718  * Conditionally replace name entry in the parent.
1719  *
1720  * As required, the LFSCK may re-create the lost MDT-object for dangling
1721  * name entry, but such repairing may be wrong because of bad FID in the
1722  * name entry. As the LFSCK processing, the real MDT-object may be found,
1723  * then the LFSCK should check whether the former re-created MDT-object
1724  * has been modified or not, if not, then destroy it and update the name
1725  * entry in the parent to reference the real MDT-object.
1726  *
1727  * \param[in] env       pointer to the thread context
1728  * \param[in] com       pointer to the lfsck component
1729  * \param[in] parent    pointer to the parent directory
1730  * \param[in] child     pointer to the MDT-object that may be the real
1731  *                      MDT-object corresponding to the name entry in parent
1732  * \param[in] cfid      the current FID in the name entry
1733  * \param[in] cname     contains the name of the child in the parent directory
1734  *
1735  * \retval              positive number for repaired cases
1736  * \retval              0 if nothing to be repaired
1737  * \retval              negative error number on failure
1738  */
1739 static int lfsck_namespace_replace_cond(const struct lu_env *env,
1740                                         struct lfsck_component *com,
1741                                         struct dt_object *parent,
1742                                         struct dt_object *child,
1743                                         const struct lu_fid *cfid,
1744                                         const struct lu_name *cname)
1745 {
1746         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1747         struct lu_attr                  *la     = &info->lti_la;
1748         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
1749         struct lu_fid                    tfid;
1750         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1751         /* The child and its name may be on different MDTs. */
1752         struct dt_device                *dev    = lfsck->li_next;
1753         const char                      *name   = cname->ln_name;
1754         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
1755         struct dt_object                *cobj   = NULL;
1756         struct lfsck_lock_handle        *pllh   = &info->lti_llh;
1757         struct lustre_handle             clh    = { 0 };
1758         struct linkea_data               ldata  = { NULL };
1759         struct thandle                  *th     = NULL;
1760         bool                             exist  = true;
1761         int                              rc     = 0;
1762         ENTRY;
1763
1764         /* @parent/@child may be based on lfsck->li_bottom,
1765          * but here we need the object based on the lfsck->li_next. */
1766
1767         parent = lfsck_object_locate(dev, parent);
1768         if (IS_ERR(parent))
1769                 GOTO(log, rc = PTR_ERR(parent));
1770
1771         if (unlikely(!dt_try_as_dir(env, parent)))
1772                 GOTO(log, rc = -ENOTDIR);
1773
1774         rc = lfsck_lock(env, lfsck, parent, name, pllh,
1775                         MDS_INODELOCK_UPDATE, LCK_PW);
1776         if (rc != 0)
1777                 GOTO(log, rc);
1778
1779         if (!fid_is_sane(cfid)) {
1780                 exist = false;
1781                 goto replace;
1782         }
1783
1784         cobj = lfsck_object_find_by_dev(env, dev, cfid);
1785         if (IS_ERR(cobj)) {
1786                 rc = PTR_ERR(cobj);
1787                 if (rc == -ENOENT) {
1788                         exist = false;
1789                         goto replace;
1790                 }
1791
1792                 GOTO(log, rc);
1793         }
1794
1795         if (!dt_object_exists(cobj)) {
1796                 exist = false;
1797                 goto replace;
1798         }
1799
1800         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1801                        (const struct dt_key *)name);
1802         if (rc == -ENOENT) {
1803                 exist = false;
1804                 goto replace;
1805         }
1806
1807         if (rc != 0)
1808                 GOTO(log, rc);
1809
1810         /* Someone changed the name entry, cannot replace it. */
1811         if (!lu_fid_eq(cfid, &tfid))
1812                 GOTO(log, rc = 0);
1813
1814         /* lock the object to be destroyed. */
1815         rc = lfsck_ibits_lock(env, lfsck, cobj, &clh,
1816                               MDS_INODELOCK_UPDATE |
1817                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
1818                               LCK_EX);
1819         if (rc != 0)
1820                 GOTO(log, rc);
1821
1822         if (unlikely(lfsck_is_dead_obj(cobj))) {
1823                 exist = false;
1824                 goto replace;
1825         }
1826
1827         rc = dt_attr_get(env, cobj, la);
1828         if (rc != 0)
1829                 GOTO(log, rc);
1830
1831         /* The object has been modified by other(s), or it is not created by
1832          * LFSCK, the two cases are indistinguishable. So cannot replace it. */
1833         if (la->la_ctime != 0)
1834                 GOTO(log, rc);
1835
1836         if (S_ISREG(la->la_mode)) {
1837                 rc = dt_xattr_get(env, cobj, &LU_BUF_NULL, XATTR_NAME_LOV);
1838                 /* If someone has created related OST-object(s),
1839                  * then keep it. */
1840                 if ((rc > 0) || (rc < 0 && rc != -ENODATA))
1841                         GOTO(log, rc = (rc > 0 ? 0 : rc));
1842         }
1843
1844 replace:
1845         dt_read_lock(env, child, 0);
1846         rc = lfsck_links_read2_with_rec(env, child, &ldata);
1847         dt_read_unlock(env, child);
1848
1849         /* Someone changed the child, no need to replace. */
1850         if (rc == -ENODATA)
1851                 GOTO(log, rc = 0);
1852
1853         if (rc != 0)
1854                 GOTO(log, rc);
1855
1856         rc = linkea_links_find(&ldata, cname, pfid);
1857         /* Someone moved the child, no need to replace. */
1858         if (rc != 0)
1859                 GOTO(log, rc = 0);
1860
1861         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1862                 GOTO(log, rc = 1);
1863
1864         th = dt_trans_create(env, dev);
1865         if (IS_ERR(th))
1866                 GOTO(log, rc = PTR_ERR(th));
1867
1868         if (exist) {
1869                 rc = dt_declare_destroy(env, cobj, th);
1870                 if (rc != 0)
1871                         GOTO(stop, rc);
1872         }
1873
1874         rc = dt_declare_delete(env, parent, (const struct dt_key *)name, th);
1875         if (rc != 0)
1876                 GOTO(stop, rc);
1877
1878         rec->rec_type = S_IFDIR;
1879         rec->rec_fid = lfsck_dto2fid(child);
1880         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
1881                                (const struct dt_key *)name, th);
1882         if (rc != 0)
1883                 GOTO(stop, rc);
1884
1885         rc = dt_trans_start_local(env, dev, th);
1886         if (rc != 0)
1887                 GOTO(stop, rc);
1888
1889         if (exist) {
1890                 rc = dt_destroy(env, cobj, th);
1891                 if (rc != 0)
1892                         GOTO(stop, rc);
1893         }
1894
1895         /* The old name entry maybe not exist. */
1896         rc = dt_delete(env, parent, (const struct dt_key *)name, th);
1897         if (rc != 0 && rc != -ENOENT)
1898                 GOTO(stop, rc);
1899
1900         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1901                        (const struct dt_key *)name, th, 1);
1902
1903         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1904
1905 stop:
1906         dt_trans_stop(env, dev, th);
1907
1908 log:
1909         lfsck_ibits_unlock(&clh, LCK_EX);
1910         lfsck_unlock(pllh);
1911
1912         if (cobj != NULL && !IS_ERR(cobj))
1913                 lfsck_object_put(env, cobj);
1914
1915         CDEBUG(D_LFSCK, "%s: namespace LFSCK conditionally destroy the "
1916                "object "DFID" because of conflict with the object "DFID
1917                " under the parent "DFID" with name %s: rc = %d\n",
1918                lfsck_lfsck2name(lfsck), PFID(cfid),
1919                PFID(lfsck_dto2fid(child)), PFID(pfid), name, rc);
1920
1921         return rc;
1922 }
1923
1924 /**
1925  * Overwrite the linkEA for the object with the given ldata.
1926  *
1927  * The caller should take the ldlm lock before the calling.
1928  *
1929  * \param[in] env       pointer to the thread context
1930  * \param[in] com       pointer to the lfsck component
1931  * \param[in] obj       pointer to the dt_object to be handled
1932  * \param[in] ldata     pointer to the new linkEA data
1933  *
1934  * \retval              positive number for repaired cases
1935  * \retval              0 if nothing to be repaired
1936  * \retval              negative error number on failure
1937  */
1938 int lfsck_namespace_rebuild_linkea(const struct lu_env *env,
1939                                    struct lfsck_component *com,
1940                                    struct dt_object *obj,
1941                                    struct linkea_data *ldata)
1942 {
1943         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1944         struct dt_device                *dev    = lfsck_obj2dev(obj);
1945         struct thandle                  *th     = NULL;
1946         struct lu_buf                    linkea_buf;
1947         int                              rc     = 0;
1948         ENTRY;
1949
1950         th = dt_trans_create(env, dev);
1951         if (IS_ERR(th))
1952                 GOTO(log, rc = PTR_ERR(th));
1953
1954         lfsck_buf_init(&linkea_buf, ldata->ld_buf->lb_buf,
1955                        ldata->ld_leh->leh_len);
1956         rc = dt_declare_xattr_set(env, obj, &linkea_buf,
1957                                   XATTR_NAME_LINK, 0, th);
1958         if (rc != 0)
1959                 GOTO(stop, rc);
1960
1961         rc = dt_trans_start_local(env, dev, th);
1962         if (rc != 0)
1963                 GOTO(stop, rc);
1964
1965         dt_write_lock(env, obj, 0);
1966         if (unlikely(lfsck_is_dead_obj(obj)))
1967                 GOTO(unlock, rc = 0);
1968
1969         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1970                 GOTO(unlock, rc = 1);
1971
1972         rc = dt_xattr_set(env, obj, &linkea_buf,
1973                           XATTR_NAME_LINK, 0, th);
1974
1975         GOTO(unlock, rc = (rc == 0 ? 1 : rc));
1976
1977 unlock:
1978         dt_write_unlock(env, obj);
1979
1980 stop:
1981         dt_trans_stop(env, dev, th);
1982
1983 log:
1984         CDEBUG(D_LFSCK, "%s: namespace LFSCK rebuild linkEA for the "
1985                "object "DFID": rc = %d\n",
1986                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
1987
1988         if (rc != 0) {
1989                 struct lfsck_namespace *ns = com->lc_file_ram;
1990
1991                 ns->ln_flags |= LF_INCONSISTENT;
1992         }
1993
1994         return rc;
1995 }
1996
1997 /**
1998  * Repair invalid name entry.
1999  *
2000  * If the name entry contains invalid information, such as bad file type
2001  * or (and) corrupted object FID, then either remove the name entry or
2002  * udpate the name entry with the given (right) information.
2003  *
2004  * \param[in] env       pointer to the thread context
2005  * \param[in] com       pointer to the lfsck component
2006  * \param[in] parent    pointer to the parent directory
2007  * \param[in] child     pointer to the object referenced by the name entry
2008  * \param[in] name      the old name of the child under the parent directory
2009  * \param[in] name2     the new name of the child under the parent directory
2010  * \param[in] type      the type claimed by the name entry
2011  * \param[in] update    update the name entry if true; otherwise, remove it
2012  * \param[in] dec       decrease the parent nlink count if true
2013  *
2014  * \retval              positive number for repaired successfully
2015  * \retval              0 if nothing to be repaired
2016  * \retval              negative error number on failure
2017  */
2018 int lfsck_namespace_repair_dirent(const struct lu_env *env,
2019                                   struct lfsck_component *com,
2020                                   struct dt_object *parent,
2021                                   struct dt_object *child,
2022                                   const char *name, const char *name2,
2023                                   __u16 type, bool update, bool dec)
2024 {
2025         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2026         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
2027         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
2028         const struct lu_fid             *cfid   = lfsck_dto2fid(child);
2029         struct lu_fid                    tfid;
2030         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2031         struct dt_device                *dev    = lfsck->li_next;
2032         struct thandle                  *th     = NULL;
2033         struct lfsck_lock_handle        *llh    = &info->lti_llh;
2034         struct lustre_handle             lh     = { 0 };
2035         int                              rc     = 0;
2036         ENTRY;
2037
2038         parent = lfsck_object_locate(dev, parent);
2039         if (IS_ERR(parent))
2040                 GOTO(log, rc = PTR_ERR(parent));
2041
2042         if (unlikely(!dt_try_as_dir(env, parent)))
2043                 GOTO(log, rc = -ENOTDIR);
2044
2045         if (!update || strcmp(name, name2) == 0)
2046                 rc = lfsck_lock(env, lfsck, parent, name, llh,
2047                                 MDS_INODELOCK_UPDATE, LCK_PW);
2048         else
2049                 rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2050                                       MDS_INODELOCK_UPDATE, LCK_PW);
2051         if (rc != 0)
2052                 GOTO(log, rc);
2053
2054         th = dt_trans_create(env, dev);
2055         if (IS_ERR(th))
2056                 GOTO(unlock1, rc = PTR_ERR(th));
2057
2058         rc = dt_declare_delete(env, parent, (const struct dt_key *)name, th);
2059         if (rc != 0)
2060                 GOTO(stop, rc);
2061
2062         if (update) {
2063                 rec->rec_type = lfsck_object_type(child) & S_IFMT;
2064                 rec->rec_fid = cfid;
2065                 rc = dt_declare_insert(env, parent,
2066                                        (const struct dt_rec *)rec,
2067                                        (const struct dt_key *)name2, th);
2068                 if (rc != 0)
2069                         GOTO(stop, rc);
2070         }
2071
2072         if (dec && S_ISDIR(type)) {
2073                 rc = dt_declare_ref_del(env, parent, th);
2074                 if (rc != 0)
2075                         GOTO(stop, rc);
2076         }
2077
2078         rc = dt_trans_start_local(env, dev, th);
2079         if (rc != 0)
2080                 GOTO(stop, rc);
2081
2082
2083         dt_write_lock(env, parent, 0);
2084         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
2085                        (const struct dt_key *)name);
2086         /* Someone has removed the bad name entry by race. */
2087         if (rc == -ENOENT)
2088                 GOTO(unlock2, rc = 0);
2089
2090         if (rc != 0)
2091                 GOTO(unlock2, rc);
2092
2093         /* Someone has removed the bad name entry and reused it for other
2094          * object by race. */
2095         if (!lu_fid_eq(&tfid, cfid))
2096                 GOTO(unlock2, rc = 0);
2097
2098         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
2099                 GOTO(unlock2, rc = 1);
2100
2101         rc = dt_delete(env, parent, (const struct dt_key *)name, th);
2102         if (rc != 0)
2103                 GOTO(unlock2, rc);
2104
2105         if (update) {
2106                 rc = dt_insert(env, parent,
2107                                (const struct dt_rec *)rec,
2108                                (const struct dt_key *)name2, th, 1);
2109                 if (rc != 0)
2110                         GOTO(unlock2, rc);
2111         }
2112
2113         if (dec && S_ISDIR(type)) {
2114                 rc = dt_ref_del(env, parent, th);
2115                 if (rc != 0)
2116                         GOTO(unlock2, rc);
2117         }
2118
2119         GOTO(unlock2, rc = (rc == 0 ? 1 : rc));
2120
2121 unlock2:
2122         dt_write_unlock(env, parent);
2123
2124 stop:
2125         dt_trans_stop(env, dev, th);
2126
2127         /* We are not sure whether the child will become orphan or not.
2128          * Record it in the LFSCK trace file for further checking in
2129          * the second-stage scanning. */
2130         if (!update && !dec && rc == 0)
2131                 lfsck_namespace_trace_update(env, com, cfid,
2132                                              LNTF_CHECK_LINKEA, true);
2133
2134 unlock1:
2135         /* It is harmless even if unlock the unused lock_handle */
2136         lfsck_ibits_unlock(&lh, LCK_PW);
2137         lfsck_unlock(llh);
2138
2139 log:
2140         CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found bad name "
2141                "entry for: parent "DFID", child "DFID", name %s, type "
2142                "in name entry %o, type claimed by child %o. repair it "
2143                "by %s with new name2 %s: rc = %d\n",
2144                lfsck_lfsck2name(lfsck), PFID(pfid), PFID(cfid),
2145                name, type, update ? lfsck_object_type(child) : 0,
2146                update ? "updating" : "removing", name2, rc);
2147
2148         if (rc != 0) {
2149                 struct lfsck_namespace *ns = com->lc_file_ram;
2150
2151                 ns->ln_flags |= LF_INCONSISTENT;
2152         }
2153
2154         return rc;
2155 }
2156
2157 /**
2158  * Update the ".." name entry for the given object.
2159  *
2160  * The object's ".." is corrupted, this function will update the ".." name
2161  * entry with the given pfid, and the linkEA with the given ldata.
2162  *
2163  * The caller should take the ldlm lock before the calling.
2164  *
2165  * \param[in] env       pointer to the thread context
2166  * \param[in] com       pointer to the lfsck component
2167  * \param[in] obj       pointer to the dt_object to be handled
2168  * \param[in] pfid      the new fid for the object's ".." name entry
2169  * \param[in] cname     the name for the @obj in the parent directory
2170  *
2171  * \retval              positive number for repaired cases
2172  * \retval              0 if nothing to be repaired
2173  * \retval              negative error number on failure
2174  */
2175 static int lfsck_namespace_repair_unmatched_pairs(const struct lu_env *env,
2176                                                   struct lfsck_component *com,
2177                                                   struct dt_object *obj,
2178                                                   const struct lu_fid *pfid,
2179                                                   struct lu_name *cname)
2180 {
2181         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2182         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
2183         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2184         struct dt_device                *dev    = lfsck_obj2dev(obj);
2185         struct thandle                  *th     = NULL;
2186         struct linkea_data               ldata  = { NULL };
2187         struct lu_buf                    linkea_buf;
2188         int                              rc     = 0;
2189         ENTRY;
2190
2191         LASSERT(!dt_object_remote(obj));
2192         LASSERT(S_ISDIR(lfsck_object_type(obj)));
2193
2194         rc = linkea_links_new(&ldata, &info->lti_big_buf, cname, pfid);
2195         if (rc != 0)
2196                 GOTO(log, rc);
2197
2198         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
2199                        ldata.ld_leh->leh_len);
2200
2201         th = dt_trans_create(env, dev);
2202         if (IS_ERR(th))
2203                 GOTO(log, rc = PTR_ERR(th));
2204
2205         rc = dt_declare_delete(env, obj, (const struct dt_key *)dotdot, th);
2206         if (rc != 0)
2207                 GOTO(stop, rc);
2208
2209         rec->rec_type = S_IFDIR;
2210         rec->rec_fid = pfid;
2211         rc = dt_declare_insert(env, obj, (const struct dt_rec *)rec,
2212                                (const struct dt_key *)dotdot, th);
2213         if (rc != 0)
2214                 GOTO(stop, rc);
2215
2216         rc = dt_declare_xattr_set(env, obj, &linkea_buf,
2217                                   XATTR_NAME_LINK, 0, th);
2218         if (rc != 0)
2219                 GOTO(stop, rc);
2220
2221         rc = dt_trans_start_local(env, dev, th);
2222         if (rc != 0)
2223                 GOTO(stop, rc);
2224
2225         dt_write_lock(env, obj, 0);
2226         if (unlikely(lfsck_is_dead_obj(obj)))
2227                 GOTO(unlock, rc = 0);
2228
2229         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
2230                 GOTO(unlock, rc = 1);
2231
2232         /* The old ".." name entry maybe not exist. */
2233         dt_delete(env, obj, (const struct dt_key *)dotdot, th);
2234
2235         rc = dt_insert(env, obj, (const struct dt_rec *)rec,
2236                        (const struct dt_key *)dotdot, th, 1);
2237         if (rc != 0)
2238                 GOTO(unlock, rc);
2239
2240         rc = lfsck_links_write(env, obj, &ldata, th);
2241
2242         GOTO(unlock, rc = (rc == 0 ? 1 : rc));
2243
2244 unlock:
2245         dt_write_unlock(env, obj);
2246
2247 stop:
2248         dt_trans_stop(env, dev, th);
2249
2250 log:
2251         CDEBUG(D_LFSCK, "%s: namespace LFSCK rebuild dotdot name entry for "
2252                "the object "DFID", new parent "DFID": rc = %d\n",
2253                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)),
2254                PFID(pfid), rc);
2255
2256         if (rc != 0) {
2257                 struct lfsck_namespace *ns = com->lc_file_ram;
2258
2259                 ns->ln_flags |= LF_INCONSISTENT;
2260         }
2261
2262         return rc;
2263 }
2264
2265 /**
2266  * Handle orphan @obj during Double Scan Directory.
2267  *
2268  * Remove the @obj's current (invalid) linkEA entries, and insert
2269  * it in the directory .lustre/lost+found/MDTxxxx/ with the name:
2270  * ${FID}-${PFID}-D-${conflict_version}
2271  *
2272  * The caller should take the ldlm lock before the calling.
2273  *
2274  * \param[in] env       pointer to the thread context
2275  * \param[in] com       pointer to the lfsck component
2276  * \param[in] obj       pointer to the orphan object to be handled
2277  * \param[in] pfid      the new fid for the object's ".." name entry
2278  * \param[in,out] lh    ldlm lock handler for the given @obj
2279  * \param[out] type     to tell the caller what the inconsistency is
2280  *
2281  * \retval              positive number for repaired cases
2282  * \retval              0 if nothing to be repaired
2283  * \retval              negative error number on failure
2284  */
2285 static int
2286 lfsck_namespace_dsd_orphan(const struct lu_env *env,
2287                            struct lfsck_component *com,
2288                            struct dt_object *obj,
2289                            const struct lu_fid *pfid,
2290                            struct lustre_handle *lh,
2291                            enum lfsck_namespace_inconsistency_type *type)
2292 {
2293         struct lfsck_thread_info *info = lfsck_env_info(env);
2294         struct lfsck_namespace   *ns   = com->lc_file_ram;
2295         int                       rc;
2296         ENTRY;
2297
2298         /* Remove the unrecognized linkEA. */
2299         rc = lfsck_namespace_links_remove(env, com, obj);
2300         lfsck_ibits_unlock(lh, LCK_EX);
2301         if (rc < 0 && rc != -ENODATA)
2302                 RETURN(rc);
2303
2304         *type = LNIT_MUL_REF;
2305
2306         /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT has
2307          * ever tried to verify some remote MDT-object that resides on this
2308          * MDT, but this MDT failed to respond such request. So means there
2309          * may be some remote name entry on other MDT that references this
2310          * object with another name, so we cannot know whether this linkEA
2311          * is valid or not. So keep it there and maybe resolved when next
2312          * LFSCK run. */
2313         if (ns->ln_flags & LF_INCOMPLETE)
2314                 RETURN(0);
2315
2316         /* The unique linkEA is invalid, even if the ".." name entry may be
2317          * valid, we still cannot know via which name entry this directory
2318          * will be referenced. Then handle it as pure orphan. */
2319         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2320                  "-"DFID, PFID(pfid));
2321         rc = lfsck_namespace_insert_orphan(env, com, obj,
2322                                            info->lti_tmpbuf, "D", NULL);
2323
2324         RETURN(rc);
2325 }
2326
2327 /**
2328  * Double Scan Directory object for single linkEA entry case.
2329  *
2330  * The given @child has unique linkEA entry. If the linkEA entry is valid,
2331  * then check whether the name is in the namespace or not, if not, add the
2332  * missing name entry back to namespace. If the linkEA entry is invalid,
2333  * then remove it and insert the @child in the .lustre/lost+found/MDTxxxx/
2334  * as an orphan.
2335  *
2336  * \param[in] env       pointer to the thread context
2337  * \param[in] com       pointer to the lfsck component
2338  * \param[in] child     pointer to the directory to be double scanned
2339  * \param[in] pfid      the FID corresponding to the ".." entry
2340  * \param[in] ldata     pointer to the linkEA data for the given @child
2341  * \param[in,out] lh    ldlm lock handler for the given @child
2342  * \param[out] type     to tell the caller what the inconsistency is
2343  * \param[in] retry     if found inconsistency, but the caller does not hold
2344  *                      ldlm lock on the @child, then set @retry as true
2345  * \param[in] unknown   set if does not know how to repair the inconsistency
2346  *
2347  * \retval              positive number for repaired cases
2348  * \retval              0 if nothing to be repaired
2349  * \retval              negative error number on failure
2350  */
2351 static int
2352 lfsck_namespace_dsd_single(const struct lu_env *env,
2353                            struct lfsck_component *com,
2354                            struct dt_object *child,
2355                            const struct lu_fid *pfid,
2356                            struct linkea_data *ldata,
2357                            struct lustre_handle *lh,
2358                            enum lfsck_namespace_inconsistency_type *type,
2359                            bool *retry, bool *unknown)
2360 {
2361         struct lfsck_thread_info *info          = lfsck_env_info(env);
2362         struct lu_name           *cname         = &info->lti_name;
2363         const struct lu_fid      *cfid          = lfsck_dto2fid(child);
2364         struct lu_fid             tfid;
2365         struct lfsck_namespace   *ns            = com->lc_file_ram;
2366         struct lfsck_instance    *lfsck         = com->lc_lfsck;
2367         struct dt_object         *parent        = NULL;
2368         struct lmv_mds_md_v1     *lmv;
2369         int                       rc            = 0;
2370         ENTRY;
2371
2372         rc = lfsck_namespace_unpack_linkea_entry(ldata, cname, &tfid,
2373                                                  info->lti_key,
2374                                                  sizeof(info->lti_key));
2375         /* The unique linkEA entry with bad parent will be handled as orphan. */
2376         if (rc != 0) {
2377                 if (!lustre_handle_is_used(lh) && retry != NULL)
2378                         *retry = true;
2379                 else
2380                         rc = lfsck_namespace_dsd_orphan(env, com, child,
2381                                                         pfid, lh, type);
2382
2383                 GOTO(out, rc);
2384         }
2385
2386         parent = lfsck_object_find_bottom(env, lfsck, &tfid);
2387         if (IS_ERR(parent))
2388                 GOTO(out, rc = PTR_ERR(parent));
2389
2390         /* We trust the unique linkEA entry in spite of whether it matches the
2391          * ".." name entry or not. Because even if the linkEA entry is wrong
2392          * and the ".." name entry is right, we still cannot know via which
2393          * name entry the child will be referenced, since all known entries
2394          * have been verified during the first-stage scanning. */
2395         if (!dt_object_exists(parent)) {
2396                 /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT
2397                  * has ever tried to verify some remote MDT-object that resides
2398                  * on this MDT, but this MDT failed to respond such request. So
2399                  * means there may be some remote name entry on other MDT that
2400                  * references this object with another name, so we cannot know
2401                  * whether this linkEA is valid or not. So keep it there and
2402                  * maybe resolved when next LFSCK run. */
2403                 if (ns->ln_flags & LF_INCOMPLETE)
2404                         GOTO(out, rc = 0);
2405
2406                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2407                         *retry = true;
2408
2409                         GOTO(out, rc = 0);
2410                 }
2411
2412                 lfsck_ibits_unlock(lh, LCK_EX);
2413
2414 lost_parent:
2415                 lmv = &info->lti_lmv;
2416                 rc = lfsck_read_stripe_lmv(env, child, lmv);
2417                 if (rc != 0 && rc != -ENODATA)
2418                         GOTO(out, rc);
2419
2420                 if (rc == -ENODATA || lmv->lmv_magic != LMV_MAGIC_STRIPE) {
2421                         lmv = NULL;
2422                 } else if (lfsck_shard_name_to_index(env,
2423                                         cname->ln_name, cname->ln_namelen,
2424                                         S_IFDIR, cfid) < 0) {
2425                         /* It is an invalid name entry, we
2426                          * cannot trust the parent also. */
2427                         rc = lfsck_namespace_shrink_linkea(env, com, child,
2428                                                 ldata, cname, &tfid, true);
2429                         if (rc < 0)
2430                                 GOTO(out, rc);
2431
2432                         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2433                                  "-"DFID, PFID(pfid));
2434                         rc = lfsck_namespace_insert_orphan(env, com, child,
2435                                                 info->lti_tmpbuf, "S", NULL);
2436
2437                         GOTO(out, rc);
2438                 }
2439
2440                 /* Create the lost parent as an orphan. */
2441                 rc = lfsck_namespace_create_orphan_dir(env, com, parent, lmv);
2442                 if (rc >= 0) {
2443                         /* Add the missing name entry to the parent. */
2444                         rc = lfsck_namespace_insert_normal(env, com, parent,
2445                                                         child, cname->ln_name);
2446                         if (unlikely(rc == -EEXIST)) {
2447                                 /* Unfortunately, someone reused the name
2448                                  * under the parent by race. So we have
2449                                  * to remove the linkEA entry from
2450                                  * current child object. It means that the
2451                                  * LFSCK cannot recover the system
2452                                  * totally back to its original status,
2453                                  * but it is necessary to make the
2454                                  * current system to be consistent. */
2455                                 rc = lfsck_namespace_shrink_linkea(env,
2456                                                 com, child, ldata,
2457                                                 cname, &tfid, true);
2458                                 if (rc >= 0) {
2459                                         snprintf(info->lti_tmpbuf,
2460                                                  sizeof(info->lti_tmpbuf),
2461                                                  "-"DFID, PFID(pfid));
2462                                         rc = lfsck_namespace_insert_orphan(env,
2463                                                 com, child, info->lti_tmpbuf,
2464                                                 "D", NULL);
2465                                 }
2466                         }
2467                 }
2468
2469                 GOTO(out, rc);
2470         } /* !dt_object_exists(parent) */
2471
2472         /* The unique linkEA entry with bad parent will be handled as orphan. */
2473         if (unlikely(!dt_try_as_dir(env, parent))) {
2474                 if (!lustre_handle_is_used(lh) && retry != NULL)
2475                         *retry = true;
2476                 else
2477                         rc = lfsck_namespace_dsd_orphan(env, com, child,
2478                                                         pfid, lh, type);
2479
2480                 GOTO(out, rc);
2481         }
2482
2483         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
2484                        (const struct dt_key *)cname->ln_name);
2485         if (rc == -ENOENT) {
2486                 /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT
2487                  * has ever tried to verify some remote MDT-object that resides
2488                  * on this MDT, but this MDT failed to respond such request. So
2489                  * means there may be some remote name entry on other MDT that
2490                  * references this object with another name, so we cannot know
2491                  * whether this linkEA is valid or not. So keep it there and
2492                  * maybe resolved when next LFSCK run. */
2493                 if (ns->ln_flags & LF_INCOMPLETE)
2494                         GOTO(out, rc = 0);
2495
2496                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2497                         *retry = true;
2498
2499                         GOTO(out, rc = 0);
2500                 }
2501
2502                 lfsck_ibits_unlock(lh, LCK_EX);
2503                 rc = lfsck_namespace_check_name(env, parent, child, cname);
2504                 if (rc == -ENOENT)
2505                         goto lost_parent;
2506
2507                 if (rc < 0)
2508                         GOTO(out, rc);
2509
2510                 /* It is an invalid name entry, drop it. */
2511                 if (unlikely(rc > 0)) {
2512                         rc = lfsck_namespace_shrink_linkea(env, com, child,
2513                                                 ldata, cname, &tfid, true);
2514                         if (rc >= 0) {
2515                                 snprintf(info->lti_tmpbuf,
2516                                          sizeof(info->lti_tmpbuf),
2517                                          "-"DFID, PFID(pfid));
2518                                 rc = lfsck_namespace_insert_orphan(env, com,
2519                                         child, info->lti_tmpbuf, "D", NULL);
2520                         }
2521
2522                         GOTO(out, rc);
2523                 }
2524
2525                 /* Add the missing name entry back to the namespace. */
2526                 rc = lfsck_namespace_insert_normal(env, com, parent, child,
2527                                                    cname->ln_name);
2528                 if (unlikely(rc == -ESTALE))
2529                         /* It may happen when the remote object has been
2530                          * removed, but the local MDT is not aware of that. */
2531                         goto lost_parent;
2532
2533                 if (unlikely(rc == -EEXIST)) {
2534                         /* Unfortunately, someone reused the name under the
2535                          * parent by race. So we have to remove the linkEA
2536                          * entry from current child object. It means that the
2537                          * LFSCK cannot recover the system totally back to
2538                          * its original status, but it is necessary to make
2539                          * the current system to be consistent.
2540                          *
2541                          * It also may be because of the LFSCK found some
2542                          * internal status of create operation. Under such
2543                          * case, nothing to be done. */
2544                         rc = lfsck_namespace_shrink_linkea_cond(env, com,
2545                                         parent, child, ldata, cname, &tfid);
2546                         if (rc >= 0) {
2547                                 snprintf(info->lti_tmpbuf,
2548                                          sizeof(info->lti_tmpbuf),
2549                                          "-"DFID, PFID(pfid));
2550                                 rc = lfsck_namespace_insert_orphan(env, com,
2551                                         child, info->lti_tmpbuf, "D", NULL);
2552                         }
2553                 }
2554
2555                 GOTO(out, rc);
2556         } /* rc == -ENOENT */
2557
2558         if (rc != 0)
2559                 GOTO(out, rc);
2560
2561         if (!lu_fid_eq(&tfid, cfid)) {
2562                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2563                         *retry = true;
2564
2565                         GOTO(out, rc = 0);
2566                 }
2567
2568                 lfsck_ibits_unlock(lh, LCK_EX);
2569                 /* The name entry references another MDT-object that
2570                  * may be created by the LFSCK for repairing dangling
2571                  * name entry. Try to replace it. */
2572                 rc = lfsck_namespace_replace_cond(env, com, parent, child,
2573                                                   &tfid, cname);
2574                 if (rc == 0)
2575                         rc = lfsck_namespace_dsd_orphan(env, com, child,
2576                                                         pfid, lh, type);
2577
2578                 GOTO(out, rc);
2579         }
2580
2581         /* Zero FID may because the remote directroy object has invalid linkEA,
2582          * or lost linkEA. Under such case, the LFSCK on this MDT does not know
2583          * how to repair the inconsistency, but the namespace LFSCK on the MDT
2584          * where its name entry resides may has more information (name, FID) to
2585          * repair such inconsistency. So here, keep the inconsistency to avoid
2586          * some imporper repairing. */
2587         if (fid_is_zero(pfid)) {
2588                 if (unknown)
2589                         *unknown = true;
2590
2591                 GOTO(out, rc = 0);
2592         }
2593
2594         /* The ".." name entry is wrong, update it. */
2595         if (!lu_fid_eq(pfid, lfsck_dto2fid(parent))) {
2596                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2597                         *retry = true;
2598
2599                         GOTO(out, rc = 0);
2600                 }
2601
2602                 *type = LNIT_UNMATCHED_PAIRS;
2603                 rc = lfsck_namespace_repair_unmatched_pairs(env, com, child,
2604                                                 lfsck_dto2fid(parent), cname);
2605         }
2606
2607         GOTO(out, rc);
2608
2609 out:
2610         if (parent != NULL && !IS_ERR(parent))
2611                 lfsck_object_put(env, parent);
2612
2613         return rc;
2614 }
2615
2616 /**
2617  * Double Scan Directory object for multiple linkEA entries case.
2618  *
2619  * The given @child has multiple linkEA entries. There is at most one linkEA
2620  * entry will be valid, all the others will be removed. Firstly, the function
2621  * will try to find out the linkEA entry for which the name entry exists under
2622  * the given parent (@pfid). If there is no linkEA entry that matches the given
2623  * ".." name entry, then tries to find out the first linkEA entry that both the
2624  * parent and the name entry exist to rebuild a new ".." name entry.
2625  *
2626  * \param[in] env       pointer to the thread context
2627  * \param[in] com       pointer to the lfsck component
2628  * \param[in] child     pointer to the directory to be double scanned
2629  * \param[in] pfid      the FID corresponding to the ".." entry
2630  * \param[in] ldata     pointer to the linkEA data for the given @child
2631  * \param[in,out] lh    ldlm lock handler for the given @child
2632  * \param[out] type     to tell the caller what the inconsistency is
2633  * \param[in] lpf       true if the ".." entry is under lost+found/MDTxxxx/
2634  * \param[in] unknown   set if does not know how to repair the inconsistency
2635  *
2636  * \retval              positive number for repaired cases
2637  * \retval              0 if nothing to be repaired
2638  * \retval              negative error number on failure
2639  */
2640 static int
2641 lfsck_namespace_dsd_multiple(const struct lu_env *env,
2642                              struct lfsck_component *com,
2643                              struct dt_object *child,
2644                              const struct lu_fid *pfid,
2645                              struct linkea_data *ldata,
2646                              struct lustre_handle *lh,
2647                              enum lfsck_namespace_inconsistency_type *type,
2648                              bool lpf, bool *unknown)
2649 {
2650         struct lfsck_thread_info *info          = lfsck_env_info(env);
2651         struct lu_name           *cname         = &info->lti_name;
2652         const struct lu_fid      *cfid          = lfsck_dto2fid(child);
2653         struct lu_fid            *pfid2         = &info->lti_fid3;
2654         struct lu_fid             tfid;
2655         struct lfsck_namespace   *ns            = com->lc_file_ram;
2656         struct lfsck_instance    *lfsck         = com->lc_lfsck;
2657         struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
2658         struct dt_object         *parent        = NULL;
2659         struct linkea_data        ldata_new     = { NULL };
2660         int                       dirent_count  = 0;
2661         int                       rc            = 0;
2662         bool                      once          = true;
2663         ENTRY;
2664
2665 again:
2666         while (ldata->ld_lee != NULL) {
2667                 rc = lfsck_namespace_unpack_linkea_entry(ldata, cname, &tfid,
2668                                                          info->lti_key,
2669                                                          sizeof(info->lti_key));
2670                 /* Drop invalid linkEA entry. */
2671                 if (rc != 0) {
2672                         lfsck_linkea_del_buf(ldata, cname);
2673                         continue;
2674                 }
2675
2676                 /* Drop repeated linkEA entries. */
2677                 lfsck_namespace_filter_linkea_entry(ldata, cname, &tfid, true);
2678
2679                 /* If current dotdot is the .lustre/lost+found/MDTxxxx/,
2680                  * then it is possible that: the directry object has ever
2681                  * been lost, but its name entry was there. In the former
2682                  * LFSCK run, during the first-stage scanning, the LFSCK
2683                  * found the dangling name entry, but it did not recreate
2684                  * the lost object, and when moved to the second-stage
2685                  * scanning, some children objects of the lost directory
2686                  * object were found, then the LFSCK recreated such lost
2687                  * directory object as an orphan.
2688                  *
2689                  * When the LFSCK runs again, if the dangling name is still
2690                  * there, the LFSCK should move the orphan directory object
2691                  * back to the normal namespace. */
2692                 if (!lpf && !fid_is_zero(pfid) &&
2693                     !lu_fid_eq(pfid, &tfid) && once) {
2694                         linkea_next_entry(ldata);
2695                         continue;
2696                 }
2697
2698                 parent = lfsck_object_find_bottom(env, lfsck, &tfid);
2699                 if (IS_ERR(parent))
2700                         RETURN(PTR_ERR(parent));
2701
2702                 if (!dt_object_exists(parent)) {
2703                         lfsck_object_put(env, parent);
2704                         if (ldata->ld_leh->leh_reccount > 1) {
2705                                 /* If it is NOT the last linkEA entry, then
2706                                  * there is still other chance to make the
2707                                  * child to be visible via other parent, then
2708                                  * remove this linkEA entry. */
2709                                 lfsck_linkea_del_buf(ldata, cname);
2710                                 continue;
2711                         }
2712
2713                         break;
2714                 }
2715
2716                 /* The linkEA entry with bad parent will be removed. */
2717                 if (unlikely(!dt_try_as_dir(env, parent))) {
2718                         lfsck_object_put(env, parent);
2719                         lfsck_linkea_del_buf(ldata, cname);
2720                         continue;
2721                 }
2722
2723                 rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
2724                                (const struct dt_key *)cname->ln_name);
2725                 *pfid2 = *lfsck_dto2fid(parent);
2726                 if (rc == -ENOENT) {
2727                         lfsck_object_put(env, parent);
2728                         linkea_next_entry(ldata);
2729                         continue;
2730                 }
2731
2732                 if (rc != 0) {
2733                         lfsck_object_put(env, parent);
2734
2735                         RETURN(rc);
2736                 }
2737
2738                 if (lu_fid_eq(&tfid, cfid)) {
2739                         lfsck_object_put(env, parent);
2740                         /* If the parent (that is declared via linkEA entry)
2741                          * directory contains the specified child, but such
2742                          * parent does not match the dotdot name entry, then
2743                          * trust the linkEA. */
2744                         if (!fid_is_zero(pfid) && !lu_fid_eq(pfid, pfid2)) {
2745                                 *type = LNIT_UNMATCHED_PAIRS;
2746                                 rc = lfsck_namespace_repair_unmatched_pairs(env,
2747                                                 com, child, pfid2, cname);
2748
2749                                 RETURN(rc);
2750                         }
2751
2752 rebuild:
2753                         /* It is the most common case that we find the
2754                          * name entry corresponding to the linkEA entry
2755                          * that matches the ".." name entry. */
2756                         rc = linkea_links_new(&ldata_new, &info->lti_big_buf,
2757                                               cname, pfid2);
2758                         if (rc != 0)
2759                                 RETURN(rc);
2760
2761                         rc = lfsck_namespace_rebuild_linkea(env, com, child,
2762                                                             &ldata_new);
2763                         if (rc < 0)
2764                                 RETURN(rc);
2765
2766                         lfsck_linkea_del_buf(ldata, cname);
2767                         linkea_first_entry(ldata);
2768                         /* There may be some invalid dangling name entries under
2769                          * other parent directories, remove all of them. */
2770                         while (ldata->ld_lee != NULL) {
2771                                 rc = lfsck_namespace_unpack_linkea_entry(ldata,
2772                                                 cname, &tfid, info->lti_key,
2773                                                 sizeof(info->lti_key));
2774                                 if (rc != 0)
2775                                         goto next;
2776
2777                                 parent = lfsck_object_find_bottom(env, lfsck,
2778                                                                   &tfid);
2779                                 if (IS_ERR(parent)) {
2780                                         rc = PTR_ERR(parent);
2781                                         if (rc != -ENOENT &&
2782                                             bk->lb_param & LPF_FAILOUT)
2783                                                 RETURN(rc);
2784
2785                                         goto next;
2786                                 }
2787
2788                                 if (!dt_object_exists(parent)) {
2789                                         lfsck_object_put(env, parent);
2790                                         goto next;
2791                                 }
2792
2793                                 rc = lfsck_namespace_repair_dirent(env, com,
2794                                         parent, child, cname->ln_name,
2795                                         cname->ln_name, S_IFDIR, false, true);
2796                                 lfsck_object_put(env, parent);
2797                                 if (rc < 0) {
2798                                         if (bk->lb_param & LPF_FAILOUT)
2799                                                 RETURN(rc);
2800
2801                                         goto next;
2802                                 }
2803
2804                                 dirent_count += rc;
2805
2806 next:
2807                                 lfsck_linkea_del_buf(ldata, cname);
2808                         }
2809
2810                         ns->ln_dirent_repaired += dirent_count;
2811
2812                         RETURN(rc);
2813                 } /* lu_fid_eq(&tfid, lfsck_dto2fid(child)) */
2814
2815                 lfsck_ibits_unlock(lh, LCK_EX);
2816                 /* The name entry references another MDT-object that may be
2817                  * created by the LFSCK for repairing dangling name entry.
2818                  * Try to replace it. */
2819                 rc = lfsck_namespace_replace_cond(env, com, parent, child,
2820                                                   &tfid, cname);
2821                 lfsck_object_put(env, parent);
2822                 if (rc < 0)
2823                         RETURN(rc);
2824
2825                 if (rc > 0)
2826                         goto rebuild;
2827
2828                 lfsck_linkea_del_buf(ldata, cname);
2829         } /* while (ldata->ld_lee != NULL) */
2830
2831         /* If there is still linkEA overflow, return. */
2832         if (unlikely(ldata->ld_leh->leh_overflow_time))
2833                 RETURN(0);
2834
2835         linkea_first_entry(ldata);
2836         if (ldata->ld_leh->leh_reccount == 1) {
2837                 rc = lfsck_namespace_dsd_single(env, com, child, pfid, ldata,
2838                                                 lh, type, NULL, unknown);
2839
2840                 RETURN(rc);
2841         }
2842
2843         /* All linkEA entries are invalid and removed, then handle the @child
2844          * as an orphan.*/
2845         if (ldata->ld_leh->leh_reccount == 0) {
2846                 rc = lfsck_namespace_dsd_orphan(env, com, child, pfid, lh,
2847                                                 type);
2848
2849                 RETURN(rc);
2850         }
2851
2852         /* If the dangling name entry for the orphan directory object has
2853          * been remvoed, then just check whether the directory object is
2854          * still under the .lustre/lost+found/MDTxxxx/ or not. */
2855         if (lpf) {
2856                 lpf = false;
2857                 goto again;
2858         }
2859
2860         /* There is no linkEA entry that matches the ".." name entry. Find
2861          * the first linkEA entry that both parent and name entry exist to
2862          * rebuild a new ".." name entry. */
2863         if (once) {
2864                 once = false;
2865                 goto again;
2866         }
2867
2868         RETURN(rc);
2869 }
2870
2871 /**
2872  * Repair the object's nlink attribute.
2873  *
2874  * If all the known name entries have been verified, then the object's hard
2875  * link attribute should match the object's linkEA entries count unless the
2876  * object's has too many hard link to be recorded in the linkEA. Such cases
2877  * should have been marked in the LFSCK trace file. Otherwise, trust the
2878  * linkEA to update the object's nlink attribute.
2879  *
2880  * \param[in] env       pointer to the thread context
2881  * \param[in] com       pointer to the lfsck component
2882  * \param[in] obj       pointer to the dt_object to be handled
2883  * \param[in,out] la    pointer to buffer to object's attribute before
2884  *                      and after the repairing
2885  *
2886  * \retval              positive number for repaired cases
2887  * \retval              0 if nothing to be repaired
2888  * \retval              negative error number on failure
2889  */
2890 static int lfsck_namespace_repair_nlink(const struct lu_env *env,
2891                                         struct lfsck_component *com,
2892                                         struct dt_object *obj,
2893                                         struct lu_attr *la)
2894 {
2895         struct lfsck_namespace          *ns     = com->lc_file_ram;
2896         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2897         struct dt_device                *dev    = lfsck_obj2dev(obj);
2898         const struct lu_fid             *cfid   = lfsck_dto2fid(obj);
2899         struct thandle                  *th     = NULL;
2900         struct linkea_data               ldata  = { NULL };
2901         struct lustre_handle             lh     = { 0 };
2902         __u32                            old    = la->la_nlink;
2903         int                              rc     = 0;
2904         ENTRY;
2905
2906         LASSERT(!dt_object_remote(obj));
2907
2908         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
2909                               MDS_INODELOCK_UPDATE, LCK_PW);
2910         if (rc != 0)
2911                 GOTO(log, rc);
2912
2913         th = dt_trans_create(env, dev);
2914         if (IS_ERR(th))
2915                 GOTO(log, rc = PTR_ERR(th));
2916
2917         la->la_valid = LA_NLINK;
2918         rc = dt_declare_attr_set(env, obj, la, th);
2919         if (rc != 0)
2920                 GOTO(stop, rc);
2921
2922         rc = dt_trans_start_local(env, dev, th);
2923         if (rc != 0)
2924                 GOTO(stop, rc);
2925
2926         dt_write_lock(env, obj, 0);
2927         /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT has
2928          * ever tried to verify some remote MDT-object that resides on this
2929          * MDT, but this MDT failed to respond such request. So means there
2930          * may be some remote name entry on other MDT that references this
2931          * object with another name, so we cannot know whether this linkEA
2932          * is valid or not. So keep it there and maybe resolved when next
2933          * LFSCK run. */
2934         if (ns->ln_flags & LF_INCOMPLETE)
2935                 GOTO(unlock, rc = 0);
2936
2937         rc = dt_attr_get(env, obj, la);
2938         if (rc != 0)
2939                 GOTO(unlock, rc = (rc == -ENOENT ? 0 : rc));
2940
2941         rc = lfsck_links_read2_with_rec(env, obj, &ldata);
2942         if (rc)
2943                 GOTO(unlock, rc = (rc == -ENODATA ? 0 : rc));
2944
2945         /* XXX: Currently, we only update the nlink attribute if the known
2946          *      linkEA entries is larger than the nlink attribute. That is
2947          *      safe action. */
2948         if (la->la_nlink >= ldata.ld_leh->leh_reccount ||
2949             unlikely(la->la_nlink == 0 ||
2950                      ldata.ld_leh->leh_overflow_time))
2951                 GOTO(unlock, rc = 0);
2952
2953         la->la_nlink = ldata.ld_leh->leh_reccount;
2954         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
2955                 GOTO(unlock, rc = 1);
2956
2957         rc = dt_attr_set(env, obj, la, th);
2958
2959         GOTO(unlock, rc = (rc == 0 ? 1 : rc));
2960
2961 unlock:
2962         dt_write_unlock(env, obj);
2963
2964 stop:
2965         dt_trans_stop(env, dev, th);
2966
2967 log:
2968         lfsck_ibits_unlock(&lh, LCK_PW);
2969         CDEBUG(D_LFSCK, "%s: namespace LFSCK repaired the object "DFID"'s "
2970                "nlink count from %u to %u: rc = %d\n",
2971                lfsck_lfsck2name(lfsck), PFID(cfid), old, la->la_nlink, rc);
2972
2973         if (rc != 0)
2974                 ns->ln_flags |= LF_INCONSISTENT;
2975
2976         return rc;
2977 }
2978
2979 /**
2980  * Double scan the directory object for namespace LFSCK.
2981  *
2982  * This function will verify the <parent, child> pairs in the namespace tree:
2983  * the parent references the child via some name entry that should be in the
2984  * child's linkEA entry, the child should back references the parent via its
2985  * ".." name entry.
2986  *
2987  * The LFSCK will scan every linkEA entry in turn until find out the first
2988  * matched pairs. If found, then all other linkEA entries will be dropped.
2989  * If all the linkEA entries cannot match the ".." name entry, then there
2990  * are serveral possible cases:
2991  *
2992  * 1) If there is only one linkEA entry, then trust it as long as the PFID
2993  *    in the linkEA entry is valid.
2994  *
2995  * 2) If there are multiple linkEA entries, then try to find the linkEA
2996  *    that matches the ".." name entry. If found, then all other entries
2997  *    are invalid; otherwise, it is quite possible that the ".." name entry
2998  *    is corrupted. Under such case, the LFSCK will rebuild the ".." name
2999  *    entry according to the first valid linkEA entry (both the parent and
3000  *    the name entry should exist).
3001  *
3002  * 3) If the directory object has no (valid) linkEA entry, then the
3003  *    directory object will be handled as pure orphan and inserted
3004  *    in the .lustre/lost+found/MDTxxxx/ with the name:
3005  *    ${self_FID}-${PFID}-D-${conflict_version}
3006  *
3007  * \param[in] env       pointer to the thread context
3008  * \param[in] com       pointer to the lfsck component
3009  * \param[in] child     pointer to the directory object to be handled
3010  * \param[in] flags     to indicate the specical checking on the @child
3011  *
3012  * \retval              positive number for repaired cases
3013  * \retval              0 if nothing to be repaired
3014  * \retval              negative error number on failure
3015  */
3016 static int lfsck_namespace_double_scan_dir(const struct lu_env *env,
3017                                            struct lfsck_component *com,
3018                                            struct dt_object *child, __u8 flags)
3019 {
3020         struct lfsck_thread_info *info          = lfsck_env_info(env);
3021         const struct lu_fid      *cfid          = lfsck_dto2fid(child);
3022         struct lu_fid            *pfid          = &info->lti_fid2;
3023         struct lfsck_namespace   *ns            = com->lc_file_ram;
3024         struct lfsck_instance    *lfsck         = com->lc_lfsck;
3025         struct lustre_handle      lh            = { 0 };
3026         struct linkea_data        ldata         = { NULL };
3027         bool                      unknown       = false;
3028         bool                      lpf           = false;
3029         bool                      retry         = false;
3030         enum lfsck_namespace_inconsistency_type type = LNIT_BAD_LINKEA;
3031         int                       rc            = 0;
3032         ENTRY;
3033
3034         LASSERT(!dt_object_remote(child));
3035
3036         if (flags & LNTF_UNCERTAIN_LMV) {
3037                 if (flags & LNTF_RECHECK_NAME_HASH) {
3038                         rc = lfsck_namespace_scan_shard(env, com, child);
3039                         if (rc < 0)
3040                                 RETURN(rc);
3041
3042                         ns->ln_striped_shards_scanned++;
3043                 } else {
3044                         ns->ln_striped_shards_skipped++;
3045                 }
3046         }
3047
3048         flags &= ~(LNTF_RECHECK_NAME_HASH | LNTF_UNCERTAIN_LMV);
3049         if (flags == 0)
3050                 RETURN(0);
3051
3052         if (flags & (LNTF_CHECK_LINKEA | LNTF_CHECK_PARENT) &&
3053             !(lfsck->li_bookmark_ram.lb_param & LPF_ALL_TGT)) {
3054                 CDEBUG(D_LFSCK, "%s: some MDT(s) maybe NOT take part in the"
3055                        "the namespace LFSCK, then the LFSCK cannot guarantee"
3056                        "all the name entries have been verified in first-stage"
3057                        "scanning. So have to skip orphan related handling for"
3058                        "the directory object "DFID" with remote name entry\n",
3059                        lfsck_lfsck2name(lfsck), PFID(cfid));
3060
3061                 RETURN(0);
3062         }
3063
3064         if (unlikely(!dt_try_as_dir(env, child)))
3065                 GOTO(out, rc = -ENOTDIR);
3066
3067         /* We only take ldlm lock on the @child when required. When the
3068          * logic comes here for the first time, it is always false. */
3069         if (0) {
3070
3071 lock:
3072                 rc = lfsck_ibits_lock(env, lfsck, child, &lh,
3073                                       MDS_INODELOCK_UPDATE |
3074                                       MDS_INODELOCK_XATTR, LCK_EX);
3075                 if (rc != 0)
3076                         GOTO(out, rc);
3077         }
3078
3079         dt_read_lock(env, child, 0);
3080         if (unlikely(lfsck_is_dead_obj(child))) {
3081                 dt_read_unlock(env, child);
3082
3083                 GOTO(out, rc = 0);
3084         }
3085
3086         rc = dt_lookup(env, child, (struct dt_rec *)pfid,
3087                        (const struct dt_key *)dotdot);
3088         if (rc != 0) {
3089                 if (rc != -ENOENT && rc != -ENODATA && rc != -EINVAL) {
3090                         dt_read_unlock(env, child);
3091
3092                         GOTO(out, rc);
3093                 }
3094
3095                 if (!lustre_handle_is_used(&lh)) {
3096                         dt_read_unlock(env, child);
3097                         goto lock;
3098                 }
3099
3100                 fid_zero(pfid);
3101         } else if (lfsck->li_lpf_obj != NULL &&
3102                    lu_fid_eq(pfid, lfsck_dto2fid(lfsck->li_lpf_obj))) {
3103                 lpf = true;
3104         } else if (unlikely(!fid_is_sane(pfid))) {
3105                 fid_zero(pfid);
3106         }
3107
3108         rc = lfsck_links_read(env, child, &ldata);
3109         dt_read_unlock(env, child);
3110         if (rc != 0) {
3111                 if (rc != -ENODATA && rc != -EINVAL)
3112                         GOTO(out, rc);
3113
3114                 if (!lustre_handle_is_used(&lh))
3115                         goto lock;
3116
3117                 if (rc == -EINVAL && !fid_is_zero(pfid)) {
3118                         /* Remove the corrupted linkEA. */
3119                         rc = lfsck_namespace_links_remove(env, com, child);
3120                         if (rc == 0)
3121                                 /* Here, because of the crashed linkEA, we
3122                                  * cannot know whether there is some parent
3123                                  * that references the child directory via
3124                                  * some name entry or not. So keep it there,
3125                                  * when the LFSCK run next time, if there is
3126                                  * some parent that references this object,
3127                                  * then the LFSCK can rebuild the linkEA;
3128                                  * otherwise, this object will be handled
3129                                  * as orphan as above. */
3130                                 unknown = true;
3131                 } else {
3132                         /* 1. If we have neither ".." nor linkEA,
3133                          *    then it is an orphan.
3134                          *
3135                          * 2. If we only have the ".." name entry,
3136                          *    but no parent references this child
3137                          *    directory, then handle it as orphan. */
3138                         lfsck_ibits_unlock(&lh, LCK_EX);
3139                         type = LNIT_MUL_REF;
3140
3141                         /* If the LFSCK is marked as LF_INCOMPLETE,
3142                          * then means some MDT has ever tried to
3143                          * verify some remote MDT-object that resides
3144                          * on this MDT, but this MDT failed to respond
3145                          * such request. So means there may be some
3146                          * remote name entry on other MDT that
3147                          * references this object with another name,
3148                          * so we cannot know whether this linkEA is
3149                          * valid or not. So keep it there and maybe
3150                          * resolved when next LFSCK run. */
3151                         if (ns->ln_flags & LF_INCOMPLETE)
3152                                 GOTO(out, rc = 0);
3153
3154                         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
3155                                  "-"DFID, PFID(pfid));
3156                         rc = lfsck_namespace_insert_orphan(env, com, child,
3157                                                 info->lti_tmpbuf, "D", NULL);
3158                 }
3159
3160                 GOTO(out, rc);
3161         } /* rc != 0 */
3162
3163         linkea_first_entry(&ldata);
3164         /* This is the most common case: the object has unique linkEA entry. */
3165         if (ldata.ld_leh->leh_reccount == 1) {
3166                 rc = lfsck_namespace_dsd_single(env, com, child, pfid, &ldata,
3167                                                 &lh, &type, &retry, &unknown);
3168                 if (retry) {
3169                         LASSERT(!lustre_handle_is_used(&lh));
3170
3171                         retry = false;
3172                         goto lock;
3173                 }
3174
3175                 GOTO(out, rc);
3176         }
3177
3178         if (!lustre_handle_is_used(&lh))
3179                 goto lock;
3180
3181         if (unlikely(ldata.ld_leh->leh_reccount == 0)) {
3182                 rc = lfsck_namespace_dsd_orphan(env, com, child, pfid, &lh,
3183                                                 &type);
3184
3185                 GOTO(out, rc);
3186         }
3187
3188         /* When we come here, the cases usually like that:
3189          * 1) The directory object has a corrupted linkEA entry. During the
3190          *    first-stage scanning, the LFSCK cannot know such corruption,
3191          *    then it appends the right linkEA entry according to the found
3192          *    name entry after the bad one.
3193          *
3194          * 2) The directory object has a right linkEA entry. During the
3195          *    first-stage scanning, the LFSCK finds some bad name entry,
3196          *    but the LFSCK cannot aware that at that time, then it adds
3197          *    the bad linkEA entry for further processing. */
3198         rc = lfsck_namespace_dsd_multiple(env, com, child, pfid, &ldata,
3199                                           &lh, &type, lpf, &unknown);
3200
3201         GOTO(out, rc);
3202
3203 out:
3204         lfsck_ibits_unlock(&lh, LCK_EX);
3205         if (rc > 0) {
3206                 switch (type) {
3207                 case LNIT_BAD_LINKEA:
3208                         ns->ln_linkea_repaired++;
3209                         break;
3210                 case LNIT_UNMATCHED_PAIRS:
3211                         ns->ln_unmatched_pairs_repaired++;
3212                         break;
3213                 case LNIT_MUL_REF:
3214                         ns->ln_mul_ref_repaired++;
3215                         break;
3216                 default:
3217                         break;
3218                 }
3219         }
3220
3221         if (unknown)
3222                 ns->ln_unknown_inconsistency++;
3223
3224         return rc;
3225 }
3226
3227 static inline bool
3228 lfsck_namespace_linkea_stale_overflow(struct linkea_data *ldata,
3229                                       struct lfsck_namespace *ns)
3230 {
3231         /* Both the leh_overflow_time and ln_time_latest_reset are
3232          * local time based, so need NOT to care about clock drift
3233          * among the servers. */
3234         return ldata->ld_leh->leh_overflow_time &&
3235                ldata->ld_leh->leh_overflow_time < ns->ln_time_latest_reset;
3236 }
3237
3238 /**
3239  * Clear the object's linkEA overflow timestamp.
3240  *
3241  * If the MDT-object has too many hard links as to the linkEA cannot hold
3242  * all of them, then overflow timestamp will be set in the linkEA header.
3243  * If some hard links are removed after that, then it is possible to hold
3244  * other missed linkEA entries. If the namespace LFSCK have added all the
3245  * related linkEA entries, then it will remove the overflow timestamp.
3246  *
3247  * \param[in] env       pointer to the thread context
3248  * \param[in] com       pointer to the lfsck component
3249  * \param[in] ldata     pointer to the linkEA data for the given @obj
3250  * \param[in] obj       pointer to the dt_object to be handled
3251  *
3252  * \retval              positive number for repaired cases
3253  * \retval              0 if nothing to be repaired
3254  * \retval              negative error number on failure
3255  */
3256 static int lfsck_namespace_linkea_clear_overflow(const struct lu_env *env,
3257                                                  struct lfsck_component *com,
3258                                                  struct linkea_data *ldata,
3259                                                  struct dt_object *obj)
3260 {
3261         struct lfsck_namespace *ns = com->lc_file_ram;
3262         struct lfsck_instance *lfsck = com->lc_lfsck;
3263         struct dt_device *dev = lfsck_obj2dev(obj);
3264         struct thandle *th = NULL;
3265         struct lustre_handle lh = { 0 };
3266         struct lu_buf linkea_buf;
3267         int rc = 0;
3268         ENTRY;
3269
3270         LASSERT(!dt_object_remote(obj));
3271
3272         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
3273                               MDS_INODELOCK_UPDATE, LCK_PW);
3274         if (rc != 0)
3275                 GOTO(log, rc);
3276
3277         th = dt_trans_create(env, dev);
3278         if (IS_ERR(th))
3279                 GOTO(log, rc = PTR_ERR(th));
3280
3281         rc = dt_declare_xattr_set(env, obj,
3282                         lfsck_buf_get_const(env, NULL, MAX_LINKEA_SIZE),
3283                         XATTR_NAME_LINK, 0, th);
3284         if (rc != 0)
3285                 GOTO(stop, rc);
3286
3287         rc = dt_trans_start_local(env, dev, th);
3288         if (rc != 0)
3289                 GOTO(stop, rc);
3290
3291         dt_write_lock(env, obj, 0);
3292         rc = lfsck_links_read(env, obj, ldata);
3293         if (rc != 0)
3294                 GOTO(unlock, rc);
3295
3296         if (unlikely(!lfsck_namespace_linkea_stale_overflow(ldata, ns)))
3297                 GOTO(unlock, rc = 0);
3298
3299         ldata->ld_leh->leh_overflow_time = 0;
3300         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
3301                 GOTO(unlock, rc = 1);
3302
3303         /* If all known entries are in the linkEA, then the 'leh_reccount'
3304          * should NOT be zero. */
3305         LASSERT(ldata->ld_leh->leh_reccount > 0);
3306
3307         lfsck_buf_init(&linkea_buf, ldata->ld_buf->lb_buf,
3308                        ldata->ld_leh->leh_len);
3309         rc = dt_xattr_set(env, obj, &linkea_buf, XATTR_NAME_LINK, 0, th);
3310         if (unlikely(rc == -ENOSPC))
3311                 rc = 0;
3312         else if (!rc)
3313                 rc = 1;
3314
3315         GOTO(unlock, rc);
3316
3317 unlock:
3318         dt_write_unlock(env, obj);
3319
3320 stop:
3321         dt_trans_stop(env, dev, th);
3322
3323 log:
3324         lfsck_ibits_unlock(&lh, LCK_PW);
3325         CDEBUG(D_LFSCK, "%s: clear linkea overflow timestamp for the object "
3326                DFID": rc = %d\n",
3327                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
3328
3329         return rc;
3330 }
3331
3332 /**
3333  * Double scan the MDT-object for namespace LFSCK.
3334  *
3335  * If the MDT-object contains invalid or repeated linkEA entries, then drop
3336  * those entries from the linkEA; if the linkEA becomes empty or the object
3337  * has no linkEA, then it is an orphan and will be added into the directory
3338  * .lustre/lost+found/MDTxxxx/; if the remote parent is lost, then recreate
3339  * the remote parent; if the name entry corresponding to some linkEA entry
3340  * is lost, then add the name entry back to the namespace.
3341  *
3342  * \param[in] env       pointer to the thread context
3343  * \param[in] com       pointer to the lfsck component
3344  * \param[in] child     pointer to the dt_object to be handled
3345  * \param[in] flags     some hints to indicate how the @child should be handled
3346  *
3347  * \retval              positive number for repaired cases
3348  * \retval              0 if nothing to be repaired
3349  * \retval              negative error number on failure
3350  */
3351 static int lfsck_namespace_double_scan_one(const struct lu_env *env,
3352                                            struct lfsck_component *com,
3353                                            struct dt_object *child, __u8 flags)
3354 {
3355         struct lfsck_thread_info *info     = lfsck_env_info(env);
3356         struct lu_attr           *la       = &info->lti_la;
3357         struct lu_name           *cname    = &info->lti_name;
3358         struct lu_fid            *pfid     = &info->lti_fid;
3359         struct lu_fid            *cfid     = &info->lti_fid2;
3360         struct lfsck_instance    *lfsck    = com->lc_lfsck;
3361         struct lfsck_namespace   *ns       = com->lc_file_ram;
3362         struct dt_object         *parent   = NULL;
3363         struct linkea_data        ldata    = { NULL };
3364         bool                      repaired = false;
3365         int                       count    = 0;
3366         int                       rc;
3367         ENTRY;
3368
3369         dt_read_lock(env, child, 0);
3370         if (unlikely(lfsck_is_dead_obj(child))) {
3371                 dt_read_unlock(env, child);
3372
3373                 RETURN(0);
3374         }
3375
3376         if (S_ISDIR(lfsck_object_type(child))) {
3377                 dt_read_unlock(env, child);
3378                 rc = lfsck_namespace_double_scan_dir(env, com, child, flags);
3379
3380                 RETURN(rc);
3381         }
3382
3383         rc = lfsck_links_read(env, child, &ldata);
3384         dt_read_unlock(env, child);
3385
3386         if (rc == -EINVAL) {
3387                 struct lustre_handle lh = { 0 };
3388
3389                 rc = lfsck_ibits_lock(env, com->lc_lfsck, child, &lh,
3390                                       MDS_INODELOCK_UPDATE |
3391                                       MDS_INODELOCK_XATTR, LCK_EX);
3392                 if (rc == 0) {
3393                         rc = lfsck_namespace_links_remove(env, com, child);
3394                         lfsck_ibits_unlock(&lh, LCK_EX);
3395                 }
3396
3397                 GOTO(out, rc);
3398         }
3399
3400         if (rc != 0)
3401                 GOTO(out, rc);
3402
3403         if (!(ns->ln_flags & LF_INCOMPLETE) &&
3404             unlikely(lfsck_namespace_linkea_stale_overflow(&ldata, ns))) {
3405                 rc = lfsck_namespace_linkea_clear_overflow(env, com, &ldata,
3406                                                            child);
3407                 if (rc < 0)
3408                         GOTO(out, rc);
3409
3410                 if (rc > 0)
3411                         ns->ln_linkea_overflow_cleared++;
3412         }
3413
3414         linkea_first_entry(&ldata);
3415         while (ldata.ld_lee != NULL) {
3416                 rc = lfsck_namespace_unpack_linkea_entry(&ldata, cname, pfid,
3417                                                          info->lti_key,
3418                                                          sizeof(info->lti_key));
3419                 /* Invalid PFID in the linkEA entry. */
3420                 if (rc != 0) {
3421                         rc = lfsck_namespace_shrink_linkea(env, com, child,
3422                                                 &ldata, cname, pfid, true);
3423                         if (rc < 0)
3424                                 GOTO(out, rc);
3425
3426                         if (rc > 0)
3427                                 repaired = true;
3428
3429                         continue;
3430                 }
3431
3432                 rc = lfsck_namespace_filter_linkea_entry(&ldata, cname, pfid,
3433                                                          false);
3434                 /* Found repeated linkEA entries */
3435                 if (rc > 0) {
3436                         rc = lfsck_namespace_shrink_linkea(env, com, child,
3437                                                 &ldata, cname, pfid, false);
3438                         if (rc < 0)
3439                                 GOTO(out, rc);
3440
3441                         if (rc == 0)
3442                                 continue;
3443
3444                         repaired = true;
3445
3446                         /* fall through */
3447                 }
3448
3449                 parent = lfsck_object_find_bottom(env, lfsck, pfid);
3450                 if (IS_ERR(parent))
3451                         GOTO(out, rc = PTR_ERR(parent));
3452
3453                 if (!dt_object_exists(parent)) {
3454
3455 lost_parent:
3456                         if (ldata.ld_leh->leh_reccount > 1) {
3457                                 /* If it is NOT the last linkEA entry, then
3458                                  * there is still other chance to make the
3459                                  * child to be visible via other parent, then
3460                                  * remove this linkEA entry. */
3461                                 rc = lfsck_namespace_shrink_linkea(env, com,
3462                                         child, &ldata, cname, pfid, true);
3463                         } else {
3464                                 /* If the LFSCK is marked as LF_INCOMPLETE,
3465                                  * then means some MDT has ever tried to
3466                                  * verify some remote MDT-object that resides
3467                                  * on this MDT, but this MDT failed to respond
3468                                  * such request. So means there may be some
3469                                  * remote name entry on other MDT that
3470                                  * references this object with another name,
3471                                  * so we cannot know whether this linkEA is
3472                                  * valid or not. So keep it there and maybe
3473                                  * resolved when next LFSCK run. */
3474                                 if (ns->ln_flags & LF_INCOMPLETE) {
3475                                         lfsck_object_put(env, parent);
3476
3477                                         GOTO(out, rc = 0);
3478                                 }
3479
3480                                 /* Create the lost parent as an orphan. */
3481                                 rc = lfsck_namespace_create_orphan_dir(env, com,
3482                                                                 parent, NULL);
3483                                 if (rc < 0) {
3484                                         lfsck_object_put(env, parent);
3485
3486                                         GOTO(out, rc);
3487                                 }
3488
3489                                 if (rc > 0)
3490                                         repaired = true;
3491
3492                                 /* Add the missing name entry to the parent. */
3493                                 rc = lfsck_namespace_insert_normal(env, com,
3494                                                 parent, child, cname->ln_name);
3495                                 if (unlikely(rc == -EEXIST))
3496                                         /* Unfortunately, someone reused the
3497                                          * name under the parent by race. So we
3498                                          * have to remove the linkEA entry from
3499                                          * current child object. It means that
3500                                          * the LFSCK cannot recover the system
3501                                          * totally back to its original status,
3502                                          * but it is necessary to make the
3503                                          * current system to be consistent. */
3504                                         rc = lfsck_namespace_shrink_linkea(env,
3505                                                         com, child, &ldata,
3506                                                         cname, pfid, true);
3507                                 else
3508                                         linkea_next_entry(&ldata);
3509                         }
3510
3511                         lfsck_object_put(env, parent);
3512                         if (rc < 0)
3513                                 GOTO(out, rc);
3514
3515                         if (rc > 0)
3516                                 repaired = true;
3517
3518                         continue;
3519                 } /* !dt_object_exists(parent) */
3520
3521                 /* The linkEA entry with bad parent will be removed. */
3522                 if (unlikely(!dt_try_as_dir(env, parent))) {
3523                         lfsck_object_put(env, parent);
3524                         rc = lfsck_namespace_shrink_linkea(env, com, child,
3525                                                 &ldata, cname, pfid, true);
3526                         if (rc < 0)
3527                                 GOTO(out, rc);
3528
3529                         if (rc > 0)
3530                                 repaired = true;
3531
3532                         continue;
3533                 }
3534
3535                 rc = dt_lookup(env, parent, (struct dt_rec *)cfid,
3536                                (const struct dt_key *)cname->ln_name);
3537                 if (rc != 0 && rc != -ENOENT) {
3538                         lfsck_object_put(env, parent);
3539
3540                         GOTO(out, rc);
3541                 }
3542
3543                 if (rc == 0) {
3544                         if (lu_fid_eq(cfid, lfsck_dto2fid(child))) {
3545                                 /* It is the most common case that we
3546                                  * find the name entry corresponding
3547                                  * to the linkEA entry. */
3548                                 lfsck_object_put(env, parent);
3549                                 linkea_next_entry(&ldata);
3550                         } else {
3551                                 /* The name entry references another
3552                                  * MDT-object that may be created by
3553                                  * the LFSCK for repairing dangling
3554                                  * name entry. Try to replace it. */
3555                                 rc = lfsck_namespace_replace_cond(env, com,
3556                                                 parent, child, cfid, cname);
3557                                 lfsck_object_put(env, parent);
3558                                 if (rc < 0)
3559                                         GOTO(out, rc);
3560
3561                                 if (rc > 0) {
3562                                         repaired = true;
3563                                         linkea_next_entry(&ldata);
3564                                 } else {
3565                                         rc = lfsck_namespace_shrink_linkea(env,
3566                                                         com, child, &ldata,
3567                                                         cname, pfid, true);
3568                                         if (rc < 0)
3569                                                 GOTO(out, rc);
3570
3571                                         if (rc > 0)
3572                                                 repaired = true;
3573                                 }
3574                         }
3575
3576                         continue;
3577                 }
3578
3579                 /* The following handles -ENOENT case */
3580
3581                 rc = dt_attr_get(env, child, la);
3582                 if (rc != 0)
3583                         GOTO(out, rc);
3584
3585                 /* If there is no name entry in the parent dir and the object
3586                  * link count is fewer than the linkea entries count, then the
3587                  * linkea entry should be removed. */
3588                 if (ldata.ld_leh->leh_reccount > la->la_nlink) {
3589                         rc = lfsck_namespace_shrink_linkea_cond(env, com,
3590                                         parent, child, &ldata, cname, pfid);
3591                         lfsck_object_put(env, parent);
3592                         if (rc < 0)
3593                                 GOTO(out, rc);
3594
3595                         if (rc > 0)
3596                                 repaired = true;
3597
3598                         continue;
3599                 }
3600
3601                 /* If the LFSCK is marked as LF_INCOMPLETE, then means some
3602                  * MDT has ever tried to verify some remote MDT-object that
3603                  * resides on this MDT, but this MDT failed to respond such
3604                  * request. So means there may be some remote name entry on
3605                  * other MDT that references this object with another name,
3606                  * so we cannot know whether this linkEA is valid or not.
3607                  * So keep it there and maybe resolved when next LFSCK run. */
3608                 if (ns->ln_flags & LF_INCOMPLETE) {
3609                         lfsck_object_put(env, parent);
3610
3611                         GOTO(out, rc = 0);
3612                 }
3613
3614                 rc = lfsck_namespace_check_name(env, parent, child, cname);
3615                 if (rc == -ENOENT)
3616                         goto lost_parent;
3617
3618                 if (rc < 0) {
3619                         lfsck_object_put(env, parent);
3620
3621                         GOTO(out, rc);
3622                 }
3623
3624                 /* It is an invalid name entry, drop it. */
3625                 if (unlikely(rc > 0)) {
3626                         lfsck_object_put(env, parent);
3627                         rc = lfsck_namespace_shrink_linkea(env, com, child,
3628                                                 &ldata, cname, pfid, true);
3629                         if (rc < 0)
3630                                 GOTO(out, rc);
3631
3632                         if (rc > 0)
3633                                 repaired = true;
3634
3635                         continue;
3636                 }
3637
3638                 /* Add the missing name entry back to the namespace. */
3639                 rc = lfsck_namespace_insert_normal(env, com, parent, child,
3640                                                    cname->ln_name);
3641                 if (unlikely(rc == -ESTALE))
3642                         /* It may happen when the remote object has been
3643                          * removed, but the local MDT is not aware of that. */
3644                         goto lost_parent;
3645
3646                 if (unlikely(rc == -EEXIST))
3647                         /* Unfortunately, someone reused the name under the
3648                          * parent by race. So we have to remove the linkEA
3649                          * entry from current child object. It means that the
3650                          * LFSCK cannot recover the system totally back to
3651                          * its original status, but it is necessary to make
3652                          * the current system to be consistent.
3653                          *
3654                          * It also may be because of the LFSCK found some
3655                          * internal status of create operation. Under such
3656                          * case, nothing to be done. */
3657                         rc = lfsck_namespace_shrink_linkea_cond(env, com,
3658                                         parent, child, &ldata, cname, pfid);
3659                 else
3660                         linkea_next_entry(&ldata);
3661
3662                 lfsck_object_put(env, parent);
3663                 if (rc < 0)
3664                         GOTO(out, rc);
3665
3666                 if (rc > 0)
3667                         repaired = true;
3668         }
3669
3670         GOTO(out, rc = 0);
3671
3672 out:
3673         if (rc < 0 && rc != -ENODATA)
3674                 return rc;
3675
3676         if (rc == 0 && ldata.ld_leh != NULL)
3677                 count = ldata.ld_leh->leh_reccount;
3678
3679         if (count == 0) {
3680                 /* If the LFSCK is marked as LF_INCOMPLETE, then means some
3681                  * MDT has ever tried to verify some remote MDT-object that
3682                  * resides on this MDT, but this MDT failed to respond such
3683                  * request. So means there may be some remote name entry on
3684                  * other MDT that references this object with another name,
3685                  * so we cannot know whether this linkEA is valid or not.
3686                  * So keep it there and maybe resolved when next LFSCK run. */
3687                 if (!(ns->ln_flags & LF_INCOMPLETE) &&
3688                     (ldata.ld_leh == NULL ||
3689                      !ldata.ld_leh->leh_overflow_time)) {
3690                         /* If the child becomes orphan, then insert it into
3691                          * the global .lustre/lost+found/MDTxxxx directory. */
3692                         rc = lfsck_namespace_insert_orphan(env, com, child,
3693                                                            "", "O", &count);
3694                         if (rc < 0)
3695                                 return rc;
3696
3697                         if (rc > 0) {
3698                                 ns->ln_mul_ref_repaired++;
3699                                 repaired = true;
3700                         }
3701                 }
3702         } else {
3703                 rc = dt_attr_get(env, child, la);
3704                 if (rc != 0)
3705                         return rc;
3706
3707                 if (la->la_nlink != 0 && la->la_nlink != count) {
3708                         if (unlikely(!S_ISREG(lfsck_object_type(child)) &&
3709                                      !S_ISLNK(lfsck_object_type(child)))) {
3710                                 CDEBUG(D_LFSCK, "%s: namespace LFSCK finds "
3711                                        "the object "DFID"'s nlink count %d "
3712                                        "does not match linkEA count %d, "
3713                                        "type %o, skip it.\n",
3714                                        lfsck_lfsck2name(lfsck),
3715                                        PFID(lfsck_dto2fid(child)),
3716                                        la->la_nlink, count,
3717                                        lfsck_object_type(child));
3718                         } else if (la->la_nlink < count &&
3719                                    likely(!ldata.ld_leh->leh_overflow_time)) {
3720                                 rc = lfsck_namespace_repair_nlink(env, com,
3721                                                                   child, la);
3722                                 if (rc > 0) {
3723                                         ns->ln_objs_nlink_repaired++;
3724                                         rc = 0;
3725                                 }
3726                         }
3727                 }
3728         }
3729
3730         if (repaired) {
3731                 if (la->la_nlink > 1)
3732                         ns->ln_mul_linked_repaired++;
3733
3734                 if (rc == 0)
3735                         rc = 1;
3736         }
3737
3738         return rc;
3739 }
3740
3741 static void lfsck_namespace_dump_statistics(struct seq_file *m,
3742                                             struct lfsck_namespace *ns,
3743                                             __u64 checked_phase1,
3744                                             __u64 checked_phase2,
3745                                             time64_t time_phase1,
3746                                             time64_t time_phase2, bool dryrun)
3747 {
3748         const char *postfix = dryrun ? "inconsistent" : "repaired";
3749
3750         seq_printf(m, "checked_phase1: %llu\n"
3751                    "checked_phase2: %llu\n"
3752                    "%s_phase1: %llu\n"
3753                    "%s_phase2: %llu\n"
3754                    "failed_phase1: %llu\n"
3755                    "failed_phase2: %llu\n"
3756                    "directories: %llu\n"
3757                    "dirent_%s: %llu\n"
3758                    "linkea_%s: %llu\n"
3759                    "nlinks_%s: %llu\n"
3760                    "multiple_linked_checked: %llu\n"
3761                    "multiple_linked_%s: %llu\n"
3762                    "unknown_inconsistency: %llu\n"
3763                    "unmatched_pairs_%s: %llu\n"
3764                    "dangling_%s: %llu\n"
3765                    "multiple_referenced_%s: %llu\n"
3766                    "bad_file_type_%s: %llu\n"
3767                    "lost_dirent_%s: %llu\n"
3768                    "local_lost_found_scanned: %llu\n"
3769                    "local_lost_found_moved: %llu\n"
3770                    "local_lost_found_skipped: %llu\n"
3771                    "local_lost_found_failed: %llu\n"
3772                    "striped_dirs_scanned: %llu\n"
3773                    "striped_dirs_%s: %llu\n"
3774                    "striped_dirs_failed: %llu\n"
3775                    "striped_dirs_disabled: %llu\n"
3776                    "striped_dirs_skipped: %llu\n"
3777                    "striped_shards_scanned: %llu\n"
3778                    "striped_shards_%s: %llu\n"
3779                    "striped_shards_failed: %llu\n"
3780                    "striped_shards_skipped: %llu\n"
3781                    "name_hash_%s: %llu\n"
3782                    "linkea_overflow_%s: %llu\n"
3783                    "success_count: %u\n"
3784                    "run_time_phase1: %lld seconds\n"
3785                    "run_time_phase2: %lld seconds\n",
3786                    checked_phase1,
3787                    checked_phase2,
3788                    dryrun ? "inconsistent" : "updated",
3789                    ns->ln_items_repaired,
3790                    dryrun ? "inconsistent" : "updated",
3791                    ns->ln_objs_repaired_phase2,
3792                    ns->ln_items_failed,
3793                    ns->ln_objs_failed_phase2,
3794                    ns->ln_dirs_checked,
3795                    postfix, ns->ln_dirent_repaired,
3796                    postfix, ns->ln_linkea_repaired,
3797                    postfix, ns->ln_objs_nlink_repaired,
3798                    ns->ln_mul_linked_checked,
3799                    postfix, ns->ln_mul_linked_repaired,
3800                    ns->ln_unknown_inconsistency,
3801                    postfix, ns->ln_unmatched_pairs_repaired,
3802                    postfix, ns->ln_dangling_repaired,
3803                    postfix, ns->ln_mul_ref_repaired,
3804                    postfix, ns->ln_bad_type_repaired,
3805                    postfix, ns->ln_lost_dirent_repaired,
3806                    ns->ln_local_lpf_scanned,
3807                    ns->ln_local_lpf_moved,
3808                    ns->ln_local_lpf_skipped,
3809                    ns->ln_local_lpf_failed,
3810                    ns->ln_striped_dirs_scanned,
3811                    postfix, ns->ln_striped_dirs_repaired,
3812                    ns->ln_striped_dirs_failed,
3813                    ns->ln_striped_dirs_disabled,
3814                    ns->ln_striped_dirs_skipped,
3815                    ns->ln_striped_shards_scanned,
3816                    postfix, ns->ln_striped_shards_repaired,
3817                    ns->ln_striped_shards_failed,
3818                    ns->ln_striped_shards_skipped,
3819                    postfix, ns->ln_name_hash_repaired,
3820                    dryrun ? "inconsistent" : "cleared",
3821                    ns->ln_linkea_overflow_cleared,
3822                    ns->ln_success_count,
3823                    time_phase1,
3824                    time_phase2);
3825 }
3826
3827 static void lfsck_namespace_release_lmv(const struct lu_env *env,
3828                                         struct lfsck_component *com)
3829 {
3830         struct lfsck_instance           *lfsck  = com->lc_lfsck;
3831         struct lfsck_namespace          *ns     = com->lc_file_ram;
3832
3833         while (!list_empty(&lfsck->li_list_lmv)) {
3834                 struct lfsck_lmv_unit   *llu;
3835                 struct lfsck_lmv        *llmv;
3836
3837                 llu = list_entry(lfsck->li_list_lmv.next,
3838                                  struct lfsck_lmv_unit, llu_link);
3839                 llmv = &llu->llu_lmv;
3840
3841                 LASSERTF(atomic_read(&llmv->ll_ref) == 1,
3842                          "still in using: %u\n",
3843                          atomic_read(&llmv->ll_ref));
3844
3845                 ns->ln_striped_dirs_skipped++;
3846                 lfsck_lmv_put(env, llmv);
3847         }
3848 }
3849
3850 static int lfsck_namespace_check_for_double_scan(const struct lu_env *env,
3851                                                  struct lfsck_component *com,
3852                                                  struct dt_object *obj)
3853 {
3854         struct lu_attr *la = &lfsck_env_info(env)->lti_la;
3855         int             rc;
3856
3857         rc = dt_attr_get(env, obj, la);
3858         if (rc != 0)
3859                 return rc;
3860
3861         /* zero-linkEA object may be orphan, but it also maybe because
3862          * of upgrading. Currently, we cannot record it for double scan.
3863          * Because it may cause the LFSCK trace file to be too large. */
3864
3865         /* "la_ctime" == 1 means that it has ever been removed from
3866          * backend /lost+found directory but not been added back to
3867          * the normal namespace yet. */
3868
3869         if ((S_ISREG(lfsck_object_type(obj)) && la->la_nlink > 1) ||
3870             unlikely(la->la_ctime == 1))
3871                 rc = lfsck_namespace_trace_update(env, com, lfsck_dto2fid(obj),
3872                                                   LNTF_CHECK_LINKEA, true);
3873
3874         return rc;
3875 }
3876
3877 /* namespace APIs */
3878
3879 static int lfsck_namespace_reset(const struct lu_env *env,
3880                                  struct lfsck_component *com, bool init)
3881 {
3882         struct lfsck_instance           *lfsck  = com->lc_lfsck;
3883         struct lfsck_namespace          *ns     = com->lc_file_ram;
3884         struct lfsck_assistant_data     *lad    = com->lc_data;
3885         struct dt_object                *root;
3886         int                              rc;
3887         ENTRY;
3888
3889         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
3890         if (IS_ERR(root))
3891                 GOTO(log, rc = PTR_ERR(root));
3892
3893         if (unlikely(!dt_try_as_dir(env, root)))
3894                 GOTO(put, rc = -ENOTDIR);
3895
3896         down_write(&com->lc_sem);
3897         if (init) {
3898                 memset(ns, 0, sizeof(*ns));
3899         } else {
3900                 __u32 count = ns->ln_success_count;
3901                 time64_t last_time = ns->ln_time_last_complete;
3902
3903                 memset(ns, 0, sizeof(*ns));
3904                 ns->ln_success_count = count;
3905                 ns->ln_time_last_complete = last_time;
3906         }
3907         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
3908         ns->ln_status = LS_INIT;
3909         ns->ln_time_latest_reset = ktime_get_real_seconds();
3910
3911         rc = lfsck_load_one_trace_file(env, com, root, &com->lc_obj,
3912                                        &dt_lfsck_namespace_features,
3913                                        LFSCK_NAMESPACE, true);
3914         if (rc)
3915                 GOTO(out, rc);
3916
3917         rc = lfsck_load_sub_trace_files(env, com, &dt_lfsck_namespace_features,
3918                                         LFSCK_NAMESPACE, true);
3919         if (rc != 0)
3920                 GOTO(out, rc);
3921
3922         lad->lad_incomplete = 0;
3923         CFS_RESET_BITMAP(lad->lad_bitmap);
3924
3925         rc = lfsck_namespace_store(env, com);
3926
3927         GOTO(out, rc);
3928
3929 out:
3930         up_write(&com->lc_sem);
3931
3932 put:
3933         lfsck_object_put(env, root);
3934 log:
3935         CDEBUG(D_LFSCK, "%s: namespace LFSCK reset: rc = %d\n",
3936                lfsck_lfsck2name(lfsck), rc);
3937         return rc;
3938 }
3939
3940 static void
3941 lfsck_namespace_fail(const struct lu_env *env, struct lfsck_component *com,
3942                      bool new_checked)
3943 {
3944         struct lfsck_namespace *ns = com->lc_file_ram;
3945
3946         down_write(&com->lc_sem);
3947         if (new_checked)
3948                 com->lc_new_checked++;
3949         lfsck_namespace_record_failure(env, com->lc_lfsck, ns);
3950         up_write(&com->lc_sem);
3951 }
3952
3953 static void lfsck_namespace_close_dir(const struct lu_env *env,
3954                                       struct lfsck_component *com)
3955 {
3956         struct lfsck_namespace          *ns     = com->lc_file_ram;
3957         struct lfsck_assistant_data     *lad    = com->lc_data;
3958         struct lfsck_assistant_object   *lso    = NULL;
3959         struct lfsck_instance           *lfsck  = com->lc_lfsck;
3960         struct lfsck_lmv                *llmv   = lfsck->li_lmv;
3961         struct lfsck_namespace_req      *lnr;
3962         __u32                            size   =
3963                                 sizeof(*lnr) + LFSCK_TMPBUF_LEN;
3964         bool                             wakeup = false;
3965         ENTRY;
3966
3967         if (llmv == NULL)
3968                 RETURN_EXIT;
3969
3970         OBD_ALLOC(lnr, size);
3971         if (lnr == NULL) {
3972                 ns->ln_striped_dirs_skipped++;
3973
3974                 RETURN_EXIT;
3975         }
3976
3977         lso = lfsck_assistant_object_init(env, lfsck_dto2fid(lfsck->li_obj_dir),
3978                         NULL, lfsck->li_pos_current.lp_oit_cookie, true);
3979         if (IS_ERR(lso)) {
3980                 OBD_FREE(lnr, size);
3981                 ns->ln_striped_dirs_skipped++;
3982
3983                 RETURN_EXIT;
3984         }
3985
3986         /* Generate a dummy request to indicate that all shards' name entry
3987          * in this striped directory has been scanned for the first time. */
3988         INIT_LIST_HEAD(&lnr->lnr_lar.lar_list);
3989         lnr->lnr_lar.lar_parent = lso;
3990         lnr->lnr_lmv = lfsck_lmv_get(llmv);
3991         lnr->lnr_fid = *lfsck_dto2fid(lfsck->li_obj_dir);
3992         lnr->lnr_dir_cookie = MDS_DIR_END_OFF;
3993         lnr->lnr_size = size;
3994
3995         spin_lock(&lad->lad_lock);
3996         if (lad->lad_assistant_status < 0) {
3997                 spin_unlock(&lad->lad_lock);
3998                 lfsck_namespace_assistant_req_fini(env, &lnr->lnr_lar);
3999                 ns->ln_striped_dirs_skipped++;
4000
4001                 RETURN_EXIT;
4002         }
4003
4004         list_add_tail(&lnr->lnr_lar.lar_list, &lad->lad_req_list);
4005         if (lad->lad_prefetched == 0)
4006                 wakeup = true;
4007
4008         lad->lad_prefetched++;
4009         spin_unlock(&lad->lad_lock);
4010         if (wakeup)
4011                 wake_up_all(&lad->lad_thread.t_ctl_waitq);
4012
4013         EXIT;
4014 }
4015
4016 static int lfsck_namespace_open_dir(const struct lu_env *env,
4017                                     struct lfsck_component *com)
4018 {
4019         struct lfsck_instance   *lfsck  = com->lc_lfsck;
4020         struct lfsck_namespace  *ns     = com->lc_file_ram;
4021         struct lfsck_lmv        *llmv   = lfsck->li_lmv;
4022         int                      rc     = 0;
4023         ENTRY;
4024
4025         if (llmv == NULL)
4026                 RETURN(0);
4027
4028         if (llmv->ll_lmv_master) {
4029                 struct lmv_mds_md_v1 *lmv = &llmv->ll_lmv;
4030
4031                 if (lmv->lmv_master_mdt_index != lfsck_dev_idx(lfsck)) {
4032                         lmv->lmv_master_mdt_index =
4033                                 lfsck_dev_idx(lfsck);
4034                         ns->ln_flags |= LF_INCONSISTENT;
4035                         llmv->ll_lmv_updated = 1;
4036                 }
4037         } else {
4038                 rc = lfsck_namespace_verify_stripe_slave(env, com,
4039                                         lfsck->li_obj_dir, llmv);
4040         }
4041
4042         RETURN(rc > 0 ? 0 : rc);
4043 }
4044
4045 static int lfsck_namespace_checkpoint(const struct lu_env *env,
4046                                       struct lfsck_component *com, bool init)
4047 {
4048         struct lfsck_instance   *lfsck = com->lc_lfsck;
4049         struct lfsck_namespace  *ns    = com->lc_file_ram;
4050         int                      rc;
4051
4052         if (!init) {
4053                 rc = lfsck_checkpoint_generic(env, com);
4054                 if (rc != 0)
4055                         goto log;
4056         }
4057
4058         down_write(&com->lc_sem);
4059         if (init) {
4060                 ns->ln_pos_latest_start = lfsck->li_pos_checkpoint;
4061         } else {
4062                 ns->ln_pos_last_checkpoint = lfsck->li_pos_checkpoint;
4063                 ns->ln_run_time_phase1 += ktime_get_seconds() -
4064                                           lfsck->li_time_last_checkpoint;
4065                 ns->ln_time_last_checkpoint = ktime_get_real_seconds();
4066                 ns->ln_items_checked += com->lc_new_checked;
4067                 com->lc_new_checked = 0;
4068         }
4069
4070         rc = lfsck_namespace_store(env, com);
4071         up_write(&com->lc_sem);
4072
4073 log:
4074         CDEBUG(D_LFSCK, "%s: namespace LFSCK checkpoint at the pos [%llu"
4075                ", "DFID", %#llx], status = %d: rc = %d\n",
4076                lfsck_lfsck2name(lfsck), lfsck->li_pos_current.lp_oit_cookie,
4077                PFID(&lfsck->li_pos_current.lp_dir_parent),
4078                lfsck->li_pos_current.lp_dir_cookie, ns->ln_status, rc);
4079
4080         return rc > 0 ? 0 : rc;
4081 }
4082
4083 static int lfsck_namespace_prep(const struct lu_env *env,
4084                                 struct lfsck_component *com,
4085                                 struct lfsck_start_param *lsp)
4086 {
4087         struct lfsck_instance   *lfsck  = com->lc_lfsck;
4088         struct lfsck_namespace  *ns     = com->lc_file_ram;
4089         struct lfsck_position   *pos    = &com->lc_pos_start;
4090         int                      rc;
4091
4092         rc = lfsck_namespace_load_bitmap(env, com);
4093         if (rc != 0 || ns->ln_status == LS_COMPLETED) {
4094                 rc = lfsck_namespace_reset(env, com, false);
4095                 if (rc == 0)
4096                         rc = lfsck_set_param(env, lfsck, lsp->lsp_start, true);
4097
4098                 if (rc != 0) {
4099                         CDEBUG(D_LFSCK, "%s: namespace LFSCK prep failed: "
4100                                "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
4101
4102                         return rc;
4103                 }
4104         }
4105
4106         down_write(&com->lc_sem);
4107         ns->ln_time_latest_start = ktime_get_real_seconds();
4108         spin_lock(&lfsck->li_lock);
4109
4110         if (ns->ln_flags & LF_SCANNED_ONCE) {
4111                 if (!lfsck->li_drop_dryrun ||
4112                     lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent)) {
4113                         ns->ln_status = LS_SCANNING_PHASE2;
4114                         list_move_tail(&com->lc_link,
4115                                        &lfsck->li_list_double_scan);
4116                         if (!list_empty(&com->lc_link_dir))
4117                                 list_del_init(&com->lc_link_dir);
4118                         lfsck_pos_set_zero(pos);
4119                 } else {
4120                         ns->ln_status = LS_SCANNING_PHASE1;
4121                         ns->ln_run_time_phase1 = 0;
4122                         ns->ln_run_time_phase2 = 0;
4123                         ns->ln_items_checked = 0;
4124                         ns->ln_items_repaired = 0;
4125                         ns->ln_items_failed = 0;
4126                         ns->ln_dirs_checked = 0;
4127                         ns->ln_objs_checked_phase2 = 0;
4128                         ns->ln_objs_repaired_phase2 = 0;
4129                         ns->ln_objs_failed_phase2 = 0;
4130                         ns->ln_objs_nlink_repaired = 0;
4131                         ns->ln_dirent_repaired = 0;
4132                         ns->ln_linkea_repaired = 0;
4133                         ns->ln_mul_linked_checked = 0;
4134                         ns->ln_mul_linked_repaired = 0;
4135                         ns->ln_unknown_inconsistency = 0;
4136                         ns->ln_unmatched_pairs_repaired = 0;
4137                         ns->ln_dangling_repaired = 0;
4138                         ns->ln_mul_ref_repaired = 0;
4139                         ns->ln_bad_type_repaired = 0;
4140                         ns->ln_lost_dirent_repaired = 0;
4141                         ns->ln_striped_dirs_scanned = 0;
4142                         ns->ln_striped_dirs_repaired = 0;
4143                         ns->ln_striped_dirs_failed = 0;
4144                         ns->ln_striped_dirs_disabled = 0;
4145                         ns->ln_striped_dirs_skipped = 0;
4146                         ns->ln_striped_shards_scanned = 0;
4147                         ns->ln_striped_shards_repaired = 0;
4148                         ns->ln_striped_shards_failed = 0;
4149                         ns->ln_striped_shards_skipped = 0;
4150                         ns->ln_name_hash_repaired = 0;
4151                         fid_zero(&ns->ln_fid_latest_scanned_phase2);
4152                         if (list_empty(&com->lc_link_dir))
4153                                 list_add_tail(&com->lc_link_dir,
4154                                               &lfsck->li_list_dir);
4155                         *pos = ns->ln_pos_first_inconsistent;
4156                 }
4157         } else {
4158                 ns->ln_status = LS_SCANNING_PHASE1;
4159                 if (list_empty(&com->lc_link_dir))
4160                         list_add_tail(&com->lc_link_dir,
4161                                       &lfsck->li_list_dir);
4162                 if (!lfsck->li_drop_dryrun ||
4163                     lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent)) {
4164                         *pos = ns->ln_pos_last_checkpoint;
4165                         pos->lp_oit_cookie++;
4166                 } else {
4167                         *pos = ns->ln_pos_first_inconsistent;
4168                 }
4169         }
4170
4171         spin_unlock(&lfsck->li_lock);
4172         up_write(&com->lc_sem);
4173
4174         rc = lfsck_start_assistant(env, com, lsp);
4175
4176         CDEBUG(D_LFSCK, "%s: namespace LFSCK prep done, start pos [%llu, "
4177                DFID", %#llx]: rc = %d\n",
4178                lfsck_lfsck2name(lfsck), pos->lp_oit_cookie,
4179                PFID(&pos->lp_dir_parent), pos->lp_dir_cookie, rc);
4180
4181         return rc;
4182 }
4183
4184 static int lfsck_namespace_exec_oit(const struct lu_env *env,
4185                                     struct lfsck_component *com,
4186                                     struct dt_object *obj)
4187 {
4188         struct lfsck_thread_info *info  = lfsck_env_info(env);
4189         struct lfsck_namespace   *ns    = com->lc_file_ram;
4190         struct lfsck_instance    *lfsck = com->lc_lfsck;
4191         const struct lu_fid      *fid   = lfsck_dto2fid(obj);
4192         struct lu_fid            *pfid  = &info->lti_fid2;
4193         struct lu_name           *cname = &info->lti_name;
4194         struct lu_seq_range      *range = &info->lti_range;
4195         struct seq_server_site   *ss    = lfsck_dev_site(lfsck);
4196         struct linkea_data        ldata = { NULL };
4197         __u32                     idx   = lfsck_dev_idx(lfsck);
4198         int                       rc;
4199         ENTRY;
4200
4201         rc = lfsck_links_read(env, obj, &ldata);
4202         if (rc == -ENOENT)
4203                 GOTO(out, rc = 0);
4204
4205         /* -EINVAL means crashed linkEA, should be verified. */
4206         if (rc == -EINVAL) {
4207                 rc = lfsck_namespace_trace_update(env, com, fid,
4208                                                   LNTF_CHECK_LINKEA, true);
4209                 if (rc == 0) {
4210                         struct lustre_handle lh = { 0 };
4211
4212                         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
4213                                               MDS_INODELOCK_UPDATE |
4214                                               MDS_INODELOCK_XATTR, LCK_EX);
4215                         if (rc == 0) {
4216                                 rc = lfsck_namespace_links_remove(env, com,
4217                                                                   obj);
4218                                 lfsck_ibits_unlock(&lh, LCK_EX);
4219                         }
4220                 }
4221
4222                 GOTO(out, rc = (rc == -ENOENT ? 0 : rc));
4223         }
4224
4225         if (rc == -ENODATA || unlikely(!ldata.ld_leh->leh_reccount)) {
4226                 rc = lfsck_namespace_check_for_double_scan(env, com, obj);
4227
4228                 GOTO(out, rc);
4229         }
4230
4231         if (rc != 0)
4232                 GOTO(out, rc);
4233
4234         /* Record multiple-linked object. */
4235         if (ldata.ld_leh->leh_reccount > 1) {
4236                 rc = lfsck_namespace_trace_update(env, com, fid,
4237                                                   LNTF_CHECK_LINKEA, true);
4238
4239                 GOTO(out, rc);
4240         }
4241
4242         linkea_first_entry(&ldata);
4243         linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen, cname, pfid);
4244         if (!fid_is_sane(pfid)) {
4245                 rc = lfsck_namespace_trace_update(env, com, fid,
4246                                                   LNTF_CHECK_PARENT, true);
4247         } else {
4248                 fld_range_set_mdt(range);
4249                 rc = fld_local_lookup(env, ss->ss_server_fld,
4250                                       fid_seq(pfid), range);
4251                 if ((rc == -ENOENT) ||
4252                     (rc == 0 && range->lsr_index != idx))
4253                         rc = lfsck_namespace_trace_update(env, com, fid,
4254                                                 LNTF_CHECK_LINKEA, true);
4255                 else
4256                         rc = lfsck_namespace_check_for_double_scan(env, com,
4257                                                                    obj);
4258         }
4259
4260         GOTO(out, rc);
4261
4262 out:
4263         down_write(&com->lc_sem);
4264         if (S_ISDIR(lfsck_object_type(obj)))
4265                 ns->ln_dirs_checked++;
4266         if (rc != 0)
4267                 lfsck_namespace_record_failure(env, com->lc_lfsck, ns);
4268         up_write(&com->lc_sem);
4269
4270         return rc;
4271 }
4272
4273 static int lfsck_namespace_exec_dir(const struct lu_env *env,
4274                                     struct lfsck_component *com,
4275                                     struct lfsck_assistant_object *lso,
4276                                     struct lu_dirent *ent, __u16 type)
4277 {
4278         struct lfsck_assistant_data     *lad     = com->lc_data;
4279         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4280         struct lfsck_namespace_req      *lnr;
4281         struct lfsck_bookmark           *bk      = &lfsck->li_bookmark_ram;
4282         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4283         struct ptlrpc_thread            *athread = &lad->lad_thread;
4284         struct l_wait_info               lwi     = { 0 };
4285         bool                             wakeup  = false;
4286
4287         l_wait_event(mthread->t_ctl_waitq,
4288                      lad->lad_prefetched < bk->lb_async_windows ||
4289                      !thread_is_running(mthread) ||
4290                      thread_is_stopped(athread),
4291                      &lwi);
4292
4293         if (unlikely(!thread_is_running(mthread)) ||
4294                      thread_is_stopped(athread))
4295                 return 0;
4296
4297         if (unlikely(lfsck_is_dead_obj(lfsck->li_obj_dir)))
4298                 return 0;
4299
4300         lnr = lfsck_namespace_assistant_req_init(com->lc_lfsck, lso, ent, type);
4301         if (IS_ERR(lnr)) {
4302                 struct lfsck_namespace *ns = com->lc_file_ram;
4303
4304                 lfsck_namespace_record_failure(env, com->lc_lfsck, ns);
4305                 return PTR_ERR(lnr);
4306         }
4307
4308         spin_lock(&lad->lad_lock);
4309         if (lad->lad_assistant_status < 0) {
4310                 spin_unlock(&lad->lad_lock);
4311                 lfsck_namespace_assistant_req_fini(env, &lnr->lnr_lar);
4312                 return lad->lad_assistant_status;
4313         }
4314
4315         list_add_tail(&lnr->lnr_lar.lar_list, &lad->lad_req_list);
4316         if (lad->lad_prefetched == 0)
4317                 wakeup = true;
4318
4319         lad->lad_prefetched++;
4320         spin_unlock(&lad->lad_lock);
4321         if (wakeup)
4322                 wake_up_all(&lad->lad_thread.t_ctl_waitq);
4323
4324         down_write(&com->lc_sem);
4325         com->lc_new_checked++;
4326         up_write(&com->lc_sem);
4327
4328         return 0;
4329 }
4330
4331 static int lfsck_namespace_post(const struct lu_env *env,
4332                                 struct lfsck_component *com,
4333                                 int result, bool init)
4334 {
4335         struct lfsck_instance   *lfsck = com->lc_lfsck;
4336         struct lfsck_namespace  *ns    = com->lc_file_ram;
4337         int                      rc;
4338         ENTRY;
4339
4340         lfsck_post_generic(env, com, &result);
4341
4342         down_write(&com->lc_sem);
4343         lfsck_namespace_release_lmv(env, com);
4344
4345         spin_lock(&lfsck->li_lock);
4346         if (!init)
4347                 ns->ln_pos_last_checkpoint = lfsck->li_pos_checkpoint;
4348         if (result > 0) {
4349                 ns->ln_status = LS_SCANNING_PHASE2;
4350                 ns->ln_flags |= LF_SCANNED_ONCE;
4351                 ns->ln_flags &= ~LF_UPGRADE;
4352                 list_del_init(&com->lc_link_dir);
4353                 list_move_tail(&com->lc_link, &lfsck->li_list_double_scan);
4354         } else if (result == 0) {
4355                 if (lfsck->li_status != 0)
4356                         ns->ln_status = lfsck->li_status;
4357                 else
4358                         ns->ln_status = LS_STOPPED;
4359                 if (ns->ln_status != LS_PAUSED) {
4360                         list_del_init(&com->lc_link_dir);
4361                         list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4362                 }
4363         } else {
4364                 ns->ln_status = LS_FAILED;
4365                 list_del_init(&com->lc_link_dir);
4366                 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4367         }
4368         spin_unlock(&lfsck->li_lock);
4369
4370         if (!init) {
4371                 ns->ln_run_time_phase1 += ktime_get_seconds() -
4372                                           lfsck->li_time_last_checkpoint;
4373                 ns->ln_time_last_checkpoint = ktime_get_real_seconds();
4374                 ns->ln_items_checked += com->lc_new_checked;
4375                 com->lc_new_checked = 0;
4376         }
4377
4378         rc = lfsck_namespace_store(env, com);
4379         up_write(&com->lc_sem);
4380
4381         CDEBUG(D_LFSCK, "%s: namespace LFSCK post done: rc = %d\n",
4382                lfsck_lfsck2name(lfsck), rc);
4383
4384         RETURN(rc);
4385 }
4386
4387 static void
4388 lfsck_namespace_dump(const struct lu_env *env, struct lfsck_component *com,
4389                      struct seq_file *m)
4390 {
4391         struct lfsck_instance   *lfsck = com->lc_lfsck;
4392         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
4393         struct lfsck_namespace  *ns    = com->lc_file_ram;
4394
4395         down_read(&com->lc_sem);
4396         seq_printf(m, "name: lfsck_namespace\n"
4397                    "magic: %#x\n"
4398                    "version: %d\n"
4399                    "status: %s\n",
4400                    ns->ln_magic,
4401                    bk->lb_version,
4402                    lfsck_status2name(ns->ln_status));
4403
4404         lfsck_bits_dump(m, ns->ln_flags, lfsck_flags_names, "flags");
4405
4406         lfsck_bits_dump(m, bk->lb_param, lfsck_param_names, "param");
4407
4408         lfsck_time_dump(m, ns->ln_time_last_complete, "last_completed");
4409
4410         lfsck_time_dump(m, ns->ln_time_latest_start, "latest_start");
4411
4412         lfsck_time_dump(m, ns->ln_time_last_checkpoint, "last_checkpoint");
4413
4414         lfsck_pos_dump(m, &ns->ln_pos_latest_start, "latest_start_position");
4415
4416         lfsck_pos_dump(m, &ns->ln_pos_last_checkpoint,
4417                        "last_checkpoint_position");
4418
4419         lfsck_pos_dump(m, &ns->ln_pos_first_inconsistent,
4420                        "first_failure_position");
4421
4422         if (ns->ln_status == LS_SCANNING_PHASE1) {
4423                 struct lfsck_position pos;
4424                 time64_t duration = ktime_get_seconds() -
4425                                     lfsck->li_time_last_checkpoint;
4426                 u64 checked = ns->ln_items_checked + com->lc_new_checked;
4427                 u64 speed = checked;
4428                 u64 new_checked = com->lc_new_checked;
4429                 time64_t rtime = ns->ln_run_time_phase1 + duration;
4430
4431                 if (duration != 0)
4432                         div_u64(new_checked, duration);
4433
4434                 if (rtime != 0)
4435                         div_u64(speed, rtime);
4436
4437                 lfsck_namespace_dump_statistics(m, ns, checked, 0, rtime, 0,
4438                                                 bk->lb_param & LPF_DRYRUN);
4439                 seq_printf(m, "average_speed_phase1: %llu items/sec\n"
4440                            "average_speed_phase2: N/A\n"
4441                            "average_speed_total: %llu items/sec\n"
4442                            "real_time_speed_phase1: %llu items/sec\n"
4443                            "real_time_speed_phase2: N/A\n",
4444                            speed,
4445                            speed,
4446                            new_checked);
4447
4448                 if (likely(lfsck->li_di_oit)) {
4449                         const struct dt_it_ops *iops =
4450                                 &lfsck->li_obj_oit->do_index_ops->dio_it;
4451
4452                         /* The low layer otable-based iteration position may NOT
4453                          * exactly match the namespace-based directory traversal
4454                          * cookie. Generally, it is not a serious issue. But the
4455                          * caller should NOT make assumption on that. */
4456                         pos.lp_oit_cookie = iops->store(env, lfsck->li_di_oit);
4457                         if (!lfsck->li_current_oit_processed)
4458                                 pos.lp_oit_cookie--;
4459
4460                         spin_lock(&lfsck->li_lock);
4461                         if (lfsck->li_di_dir) {
4462                                 pos.lp_dir_cookie = lfsck->li_cookie_dir;
4463                                 if (pos.lp_dir_cookie >= MDS_DIR_END_OFF) {
4464                                         fid_zero(&pos.lp_dir_parent);
4465                                         pos.lp_dir_cookie = 0;
4466                                 } else {
4467                                         pos.lp_dir_parent =
4468                                         *lfsck_dto2fid(lfsck->li_obj_dir);
4469                                 }
4470                         } else {
4471                                 fid_zero(&pos.lp_dir_parent);
4472                                 pos.lp_dir_cookie = 0;
4473                         }
4474                         spin_unlock(&lfsck->li_lock);
4475                 } else {
4476                         pos = ns->ln_pos_last_checkpoint;
4477                 }
4478
4479                 lfsck_pos_dump(m, &pos, "current_position");
4480         } else if (ns->ln_status == LS_SCANNING_PHASE2) {
4481                 time64_t duration = ktime_get_seconds() -
4482                                     com->lc_time_last_checkpoint;
4483                 __u64 checked = ns->ln_objs_checked_phase2 +
4484                                 com->lc_new_checked;
4485                 __u64 speed1 = ns->ln_items_checked;
4486                 __u64 speed2 = checked;
4487                 __u64 speed0 = speed1 + speed2;
4488                 __u64 new_checked = com->lc_new_checked;
4489                 time64_t rtime = ns->ln_run_time_phase2 + duration;
4490                 time64_t time0 = ns->ln_run_time_phase1 + rtime;
4491
4492                 if (duration != 0)
4493                         div_u64(new_checked, duration);
4494
4495                 if (ns->ln_run_time_phase1 != 0)
4496                         div_u64(speed1, ns->ln_run_time_phase1);
4497                 else if (ns->ln_items_checked != 0)
4498                         time0++;
4499
4500                 if (rtime != 0)
4501                         div_u64(speed2, rtime);
4502                 else if (checked != 0)
4503                         time0++;
4504
4505                 if (time0 != 0)
4506                         div_u64(speed0, time0);
4507
4508                 lfsck_namespace_dump_statistics(m, ns, ns->ln_items_checked,
4509                                                 checked,
4510                                                 ns->ln_run_time_phase1, rtime,
4511                                                 bk->lb_param & LPF_DRYRUN);
4512                 seq_printf(m, "average_speed_phase1: %llu items/sec\n"
4513                            "average_speed_phase2: %llu objs/sec\n"
4514                            "average_speed_total: %llu items/sec\n"
4515                            "real_time_speed_phase1: N/A\n"
4516                            "real_time_speed_phase2: %llu objs/sec\n"
4517                            "current_position: "DFID"\n",
4518                            speed1,
4519                            speed2,
4520                            speed0,
4521                            new_checked,
4522                            PFID(&ns->ln_fid_latest_scanned_phase2));
4523         } else {
4524                 __u64 speed1 = ns->ln_items_checked;
4525                 __u64 speed2 = ns->ln_objs_checked_phase2;
4526                 __u64 speed0 = speed1 + speed2;
4527                 time64_t time0 = ns->ln_run_time_phase1 + ns->ln_run_time_phase2;
4528
4529                 if (ns->ln_run_time_phase1 != 0)
4530                         div_u64(speed1, ns->ln_run_time_phase1);
4531                 else if (ns->ln_items_checked != 0)
4532                         time0++;
4533
4534                 if (ns->ln_run_time_phase2 != 0)
4535                         div_u64(speed2, ns->ln_run_time_phase2);
4536                 else if (ns->ln_objs_checked_phase2 != 0)
4537                         time0++;
4538
4539                 if (time0 != 0)
4540                         div_u64(speed0, time0);
4541
4542                 lfsck_namespace_dump_statistics(m, ns, ns->ln_items_checked,
4543                                                 ns->ln_objs_checked_phase2,
4544                                                 ns->ln_run_time_phase1,
4545                                                 ns->ln_run_time_phase2,
4546                                                 bk->lb_param & LPF_DRYRUN);
4547                 seq_printf(m, "average_speed_phase1: %llu items/sec\n"
4548                            "average_speed_phase2: %llu objs/sec\n"
4549                            "average_speed_total: %llu items/sec\n"
4550                            "real_time_speed_phase1: N/A\n"
4551                            "real_time_speed_phase2: N/A\n"
4552                            "current_position: N/A\n",
4553                            speed1,
4554                            speed2,
4555                            speed0);
4556         }
4557
4558         up_read(&com->lc_sem);
4559 }
4560
4561 static int lfsck_namespace_double_scan(const struct lu_env *env,
4562                                        struct lfsck_component *com)
4563 {
4564         struct lfsck_namespace          *ns     = com->lc_file_ram;
4565         struct lfsck_assistant_data     *lad    = com->lc_data;
4566         struct lfsck_tgt_descs          *ltds   = &com->lc_lfsck->li_mdt_descs;
4567         struct lfsck_tgt_desc           *ltd;
4568         struct lfsck_tgt_desc           *next;
4569         int                              rc;
4570
4571         rc = lfsck_double_scan_generic(env, com, ns->ln_status);
4572         if (thread_is_stopped(&lad->lad_thread)) {
4573                 LASSERT(list_empty(&lad->lad_req_list));
4574                 LASSERT(list_empty(&lad->lad_mdt_phase1_list));
4575
4576                 spin_lock(&ltds->ltd_lock);
4577                 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
4578                                          ltd_namespace_phase_list) {
4579                         list_del_init(&ltd->ltd_namespace_phase_list);
4580                 }
4581                 spin_unlock(&ltds->ltd_lock);
4582         }
4583
4584         return rc;
4585 }
4586
4587 static void lfsck_namespace_data_release(const struct lu_env *env,
4588                                          struct lfsck_component *com)
4589 {
4590         struct lfsck_assistant_data     *lad    = com->lc_data;
4591         struct lfsck_tgt_descs          *ltds   = &com->lc_lfsck->li_mdt_descs;
4592         struct lfsck_tgt_desc           *ltd;
4593         struct lfsck_tgt_desc           *next;
4594
4595         LASSERT(lad != NULL);
4596         LASSERT(thread_is_init(&lad->lad_thread) ||
4597                 thread_is_stopped(&lad->lad_thread));
4598         LASSERT(list_empty(&lad->lad_req_list));
4599
4600         com->lc_data = NULL;
4601         lfsck_namespace_release_lmv(env, com);
4602
4603         spin_lock(&ltds->ltd_lock);
4604         list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase1_list,
4605                                  ltd_namespace_phase_list) {
4606                 list_del_init(&ltd->ltd_namespace_phase_list);
4607         }
4608         list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
4609                                  ltd_namespace_phase_list) {
4610                 list_del_init(&ltd->ltd_namespace_phase_list);
4611         }
4612         list_for_each_entry_safe(ltd, next, &lad->lad_mdt_list,
4613                                  ltd_namespace_list) {
4614                 list_del_init(&ltd->ltd_namespace_list);
4615         }
4616         spin_unlock(&ltds->ltd_lock);
4617
4618         if (likely(lad->lad_bitmap != NULL))
4619                 CFS_FREE_BITMAP(lad->lad_bitmap);
4620
4621         OBD_FREE_PTR(lad);
4622 }
4623
4624 static void lfsck_namespace_quit(const struct lu_env *env,
4625                                  struct lfsck_component *com)
4626 {
4627         struct lfsck_assistant_data     *lad    = com->lc_data;
4628         struct lfsck_tgt_descs          *ltds   = &com->lc_lfsck->li_mdt_descs;
4629         struct lfsck_tgt_desc           *ltd;
4630         struct lfsck_tgt_desc           *next;
4631
4632         LASSERT(lad != NULL);
4633
4634         lfsck_quit_generic(env, com);
4635
4636         LASSERT(thread_is_init(&lad->lad_thread) ||
4637                 thread_is_stopped(&lad->lad_thread));
4638         LASSERT(list_empty(&lad->lad_req_list));
4639
4640         lfsck_namespace_release_lmv(env, com);
4641
4642         spin_lock(&ltds->ltd_lock);
4643         list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase1_list,
4644                                  ltd_namespace_phase_list) {
4645                 list_del_init(&ltd->ltd_namespace_phase_list);
4646         }
4647         list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
4648                                  ltd_namespace_phase_list) {
4649                 list_del_init(&ltd->ltd_namespace_phase_list);
4650         }
4651         spin_unlock(&ltds->ltd_lock);
4652 }
4653
4654 static int lfsck_namespace_in_notify(const struct lu_env *env,
4655                                      struct lfsck_component *com,
4656                                      struct lfsck_request *lr)
4657 {
4658         struct lfsck_instance *lfsck = com->lc_lfsck;
4659         struct lfsck_namespace *ns = com->lc_file_ram;
4660         struct lfsck_assistant_data *lad = com->lc_data;
4661         struct lfsck_tgt_descs *ltds = &lfsck->li_mdt_descs;
4662         struct lfsck_tgt_desc *ltd;
4663         int rc = 0;
4664         bool fail = false;
4665         ENTRY;
4666
4667         switch (lr->lr_event) {
4668         case LE_SET_LMV_MASTER: {
4669                 struct dt_object        *obj;
4670
4671                 obj = lfsck_object_find_bottom(env, lfsck, &lr->lr_fid);
4672                 if (IS_ERR(obj))
4673                         RETURN(PTR_ERR(obj));
4674
4675                 if (likely(dt_object_exists(obj)))
4676                         rc = lfsck_namespace_notify_lmv_master_local(env, com,
4677                                                                      obj);
4678
4679                 lfsck_object_put(env, obj);
4680
4681                 RETURN(rc > 0 ? 0 : rc);
4682         }
4683         case LE_SET_LMV_SLAVE: {
4684                 if (!(lr->lr_flags & LEF_RECHECK_NAME_HASH))
4685                         ns->ln_striped_shards_repaired++;
4686
4687                 rc = lfsck_namespace_trace_update(env, com, &lr->lr_fid,
4688                                                   LNTF_RECHECK_NAME_HASH, true);
4689
4690                 RETURN(rc > 0 ? 0 : rc);
4691         }
4692         case LE_PHASE1_DONE:
4693         case LE_PHASE2_DONE:
4694         case LE_PEER_EXIT:
4695                 break;
4696         default:
4697                 RETURN(-EINVAL);
4698         }
4699
4700         CDEBUG(D_LFSCK, "%s: namespace LFSCK handles notify %u from MDT %x, "
4701                "status %d, flags %x\n", lfsck_lfsck2name(lfsck), lr->lr_event,
4702                lr->lr_index, lr->lr_status, lr->lr_flags2);
4703
4704         spin_lock(&ltds->ltd_lock);
4705         ltd = lfsck_ltd2tgt(ltds, lr->lr_index);
4706         if (ltd == NULL) {
4707                 spin_unlock(&ltds->ltd_lock);
4708
4709                 RETURN(-ENXIO);
4710         }
4711
4712         list_del_init(&ltd->ltd_namespace_phase_list);
4713         switch (lr->lr_event) {
4714         case LE_PHASE1_DONE:
4715                 if (lr->lr_status <= 0) {
4716                         ltd->ltd_namespace_done = 1;
4717                         list_del_init(&ltd->ltd_namespace_list);
4718                         CDEBUG(D_LFSCK, "%s: MDT %x failed/stopped at "
4719                                "phase1 for namespace LFSCK: rc = %d.\n",
4720                                lfsck_lfsck2name(lfsck),
4721                                ltd->ltd_index, lr->lr_status);
4722                         ns->ln_flags |= LF_INCOMPLETE;
4723                         fail = true;
4724                         break;
4725                 }
4726
4727                 if (lr->lr_flags2 & LF_INCOMPLETE)
4728                         ns->ln_flags |= LF_INCOMPLETE;
4729
4730                 if (list_empty(&ltd->ltd_namespace_list))
4731                         list_add_tail(&ltd->ltd_namespace_list,
4732                                       &lad->lad_mdt_list);
4733                 list_add_tail(&ltd->ltd_namespace_phase_list,
4734                               &lad->lad_mdt_phase2_list);
4735                 break;
4736         case LE_PHASE2_DONE:
4737                 ltd->ltd_namespace_done = 1;
4738                 list_del_init(&ltd->ltd_namespace_list);
4739                 break;
4740         case LE_PEER_EXIT:
4741                 fail = true;
4742                 ltd->ltd_namespace_done = 1;
4743                 list_del_init(&ltd->ltd_namespace_list);
4744                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT)) {
4745                         CDEBUG(D_LFSCK,
4746                                "%s: the peer MDT %x exit namespace LFSCK\n",
4747                                lfsck_lfsck2name(lfsck), ltd->ltd_index);
4748                         ns->ln_flags |= LF_INCOMPLETE;
4749                 }
4750                 break;
4751         default:
4752                 break;
4753         }
4754         spin_unlock(&ltds->ltd_lock);
4755
4756         if (fail && lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
4757                 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
4758
4759                 memset(stop, 0, sizeof(*stop));
4760                 stop->ls_status = lr->lr_status;
4761                 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
4762                 lfsck_stop(env, lfsck->li_bottom, stop);
4763         } else if (lfsck_phase2_next_ready(lad)) {
4764                 wake_up_all(&lad->lad_thread.t_ctl_waitq);
4765         }
4766
4767         RETURN(0);
4768 }
4769
4770 static void lfsck_namespace_repaired(struct lfsck_namespace *ns, __u64 *count)
4771 {
4772         *count += ns->ln_objs_nlink_repaired;
4773         *count += ns->ln_dirent_repaired;
4774         *count += ns->ln_linkea_repaired;
4775         *count += ns->ln_mul_linked_repaired;
4776         *count += ns->ln_unmatched_pairs_repaired;
4777         *count += ns->ln_dangling_repaired;
4778         *count += ns->ln_mul_ref_repaired;
4779         *count += ns->ln_bad_type_repaired;
4780         *count += ns->ln_lost_dirent_repaired;
4781         *count += ns->ln_striped_dirs_disabled;
4782         *count += ns->ln_striped_dirs_repaired;
4783         *count += ns->ln_striped_shards_repaired;
4784         *count += ns->ln_name_hash_repaired;
4785         *count += ns->ln_local_lpf_moved;
4786 }
4787
4788 static int lfsck_namespace_query_all(const struct lu_env *env,
4789                                      struct lfsck_component *com,
4790                                      __u32 *mdts_count, __u64 *repaired)
4791 {
4792         struct lfsck_namespace *ns = com->lc_file_ram;
4793         struct lfsck_tgt_descs *ltds = &com->lc_lfsck->li_mdt_descs;
4794         struct lfsck_tgt_desc *ltd;
4795         int idx;
4796         int rc;
4797         ENTRY;
4798
4799         rc = lfsck_query_all(env, com);
4800         if (rc != 0)
4801                 RETURN(rc);
4802
4803         down_read(&ltds->ltd_rw_sem);
4804         cfs_foreach_bit(ltds->ltd_tgts_bitmap, idx) {
4805                 ltd = lfsck_ltd2tgt(ltds, idx);
4806                 LASSERT(ltd != NULL);
4807
4808                 mdts_count[ltd->ltd_namespace_status]++;
4809                 *repaired += ltd->ltd_namespace_repaired;
4810         }
4811         up_read(&ltds->ltd_rw_sem);
4812
4813         down_read(&com->lc_sem);
4814         mdts_count[ns->ln_status]++;
4815         lfsck_namespace_repaired(ns, repaired);
4816         up_read(&com->lc_sem);
4817
4818         RETURN(0);
4819 }
4820
4821 static int lfsck_namespace_query(const struct lu_env *env,
4822                                  struct lfsck_component *com,
4823                                  struct lfsck_request *req,
4824                                  struct lfsck_reply *rep,
4825                                  struct lfsck_query *que, int idx)
4826 {
4827         struct lfsck_namespace *ns = com->lc_file_ram;
4828         int rc = 0;
4829
4830         if (que != NULL) {
4831                 LASSERT(com->lc_lfsck->li_master);
4832
4833                 rc = lfsck_namespace_query_all(env, com,
4834                                                que->lu_mdts_count[idx],
4835                                                &que->lu_repaired[idx]);
4836         } else {
4837                 down_read(&com->lc_sem);
4838                 rep->lr_status = ns->ln_status;
4839                 if (req->lr_flags & LEF_QUERY_ALL)
4840                         lfsck_namespace_repaired(ns, &rep->lr_repaired);
4841                 up_read(&com->lc_sem);
4842         }
4843
4844         return rc;
4845 }
4846
4847 static struct lfsck_operations lfsck_namespace_ops = {
4848         .lfsck_reset            = lfsck_namespace_reset,
4849         .lfsck_fail             = lfsck_namespace_fail,
4850         .lfsck_close_dir        = lfsck_namespace_close_dir,
4851         .lfsck_open_dir         = lfsck_namespace_open_dir,
4852         .lfsck_checkpoint       = lfsck_namespace_checkpoint,
4853         .lfsck_prep             = lfsck_namespace_prep,
4854         .lfsck_exec_oit         = lfsck_namespace_exec_oit,
4855         .lfsck_exec_dir         = lfsck_namespace_exec_dir,
4856         .lfsck_post             = lfsck_namespace_post,
4857         .lfsck_dump             = lfsck_namespace_dump,
4858         .lfsck_double_scan      = lfsck_namespace_double_scan,
4859         .lfsck_data_release     = lfsck_namespace_data_release,
4860         .lfsck_quit             = lfsck_namespace_quit,
4861         .lfsck_in_notify        = lfsck_namespace_in_notify,
4862         .lfsck_query            = lfsck_namespace_query,
4863 };
4864
4865 /**
4866  * Repair dangling name entry.
4867  *
4868  * For the name entry with dangling reference, we need to repare the
4869  * inconsistency according to the LFSCK sponsor's requirement:
4870  *
4871  * 1) Keep the inconsistency there and report the inconsistency case,
4872  *    then give the chance to the application to find related issues,
4873  *    and the users can make the decision about how to handle it with
4874  *    more human knownledge. (by default)
4875  *
4876  * 2) Re-create the missing MDT-object with the FID information.
4877  *
4878  * \param[in] env       pointer to the thread context
4879  * \param[in] com       pointer to the lfsck component
4880  * \param[in] parent    pointer to the dir object that contains the dangling
4881  *                      name entry
4882  * \param[in] child     pointer to the object corresponding to the dangling
4883  *                      name entry
4884  * \param[in] lnr       pointer to the namespace request that contains the
4885  *                      name's name, parent object, parent's LMV, and ect.
4886  *
4887  * \retval              positive number if no need to repair
4888  * \retval              zero for repaired successfully
4889  * \retval              negative error number on failure
4890  */
4891 int lfsck_namespace_repair_dangling(const struct lu_env *env,
4892                                     struct lfsck_component *com,
4893                                     struct dt_object *parent,
4894                                     struct dt_object *child,
4895                                     struct lfsck_namespace_req *lnr)
4896 {
4897         struct lfsck_thread_info        *info   = lfsck_env_info(env);
4898         struct lu_attr                  *la     = &info->lti_la;
4899         struct dt_allocation_hint       *hint   = &info->lti_hint;
4900         struct dt_object_format         *dof    = &info->lti_dof;
4901         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
4902         struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
4903         const struct lu_name            *cname;
4904         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
4905         const struct lu_fid             *cfid   = lfsck_dto2fid(child);
4906         struct linkea_data               ldata  = { NULL };
4907         struct lfsck_lock_handle        *llh    = &info->lti_llh;
4908         struct lu_buf                    linkea_buf;
4909         struct lu_buf                    lmv_buf;
4910         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4911         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
4912         struct dt_device                *dev    = lfsck->li_next;
4913         struct thandle                  *th     = NULL;
4914         int                              rc     = 0;
4915         __u16                            type   = lnr->lnr_type;
4916         bool                             create;
4917         ENTRY;
4918
4919         cname = lfsck_name_get_const(env, lnr->lnr_name, lnr->lnr_namelen);
4920         if (bk->lb_param & LPF_CREATE_MDTOBJ)
4921                 create = true;
4922         else
4923                 create = false;
4924
4925         if (!create || bk->lb_param & LPF_DRYRUN)
4926                 GOTO(log, rc = 0);
4927
4928         /* We may need to create the sub-objects of the @child via LOD,
4929          * so make the modification based on lfsck->li_next. */
4930
4931         parent = lfsck_object_locate(dev, parent);
4932         if (IS_ERR(parent))
4933                 GOTO(log, rc = PTR_ERR(parent));
4934
4935         if (unlikely(!dt_try_as_dir(env, parent)))
4936                 GOTO(log, rc = -ENOTDIR);
4937
4938         child = lfsck_object_locate(dev, child);
4939         if (IS_ERR(child))
4940                 GOTO(log, rc = PTR_ERR(child));
4941
4942         rc = linkea_links_new(&ldata, &info->lti_linkea_buf2,
4943                               cname, pfid);
4944         if (rc != 0)
4945                 GOTO(log, rc);
4946
4947         rc = lfsck_lock(env, lfsck, parent, lnr->lnr_name, llh,
4948                         MDS_INODELOCK_UPDATE, LCK_PR);
4949         if (rc != 0)
4950                 GOTO(log, rc);
4951
4952         rc = lfsck_namespace_check_exist(env, parent, child, lnr->lnr_name);
4953         if (rc != 0)
4954                 GOTO(log, rc);
4955
4956         /* Set the ctime as zero, then others can know it is created for
4957          * repairing dangling name entry by LFSCK. And if the LFSCK made
4958          * wrong decision and the real MDT-object has been found later,
4959          * then the LFSCK has chance to fix the incosistency properly. */
4960         memset(la, 0, sizeof(*la));
4961         la->la_mode = (type & S_IFMT) | 0600;
4962         la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
4963                         LA_ATIME | LA_MTIME | LA_CTIME;
4964
4965         child->do_ops->do_ah_init(env, hint, parent, child,
4966                                  la->la_mode & S_IFMT);
4967
4968         memset(dof, 0, sizeof(*dof));
4969         dof->dof_type = dt_mode_to_dft(type);
4970         /* If the target is a regular file, then the LFSCK will only create
4971          * the MDT-object without stripes (dof->dof_reg.striped = 0). related
4972          * OST-objects will be created when write open. */
4973
4974         th = dt_trans_create(env, dev);
4975         if (IS_ERR(th))
4976                 GOTO(log, rc = PTR_ERR(th));
4977
4978         /* 1a. create child. */
4979         rc = dt_declare_create(env, child, la, hint, dof, th);
4980         if (rc != 0)
4981                 GOTO(stop, rc);
4982
4983         if (S_ISDIR(type)) {
4984                 if (unlikely(!dt_try_as_dir(env, child)))
4985                         GOTO(stop, rc = -ENOTDIR);
4986
4987                 /* 2a. increase child nlink */
4988                 rc = dt_declare_ref_add(env, child, th);
4989                 if (rc != 0)
4990                         GOTO(stop, rc);
4991
4992                 /* 3a. insert dot into child dir */
4993                 rec->rec_type = S_IFDIR;
4994                 rec->rec_fid = cfid;
4995                 rc = dt_declare_insert(env, child,
4996                                        (const struct dt_rec *)rec,
4997                                        (const struct dt_key *)dot, th);
4998                 if (rc != 0)
4999                         GOTO(stop, rc);
5000
5001                 /* 4a. insert dotdot into child dir */
5002                 rec->rec_fid = pfid;
5003                 rc = dt_declare_insert(env, child,
5004                                        (const struct dt_rec *)rec,
5005                                        (const struct dt_key *)dotdot, th);
5006                 if (rc != 0)
5007                         GOTO(stop, rc);
5008
5009                 /* 5a. generate slave LMV EA. */
5010                 if (lnr->lnr_lmv != NULL && lnr->lnr_lmv->ll_lmv_master) {
5011                         int idx;
5012
5013                         idx = lfsck_shard_name_to_index(env,
5014                                         lnr->lnr_name, lnr->lnr_namelen,
5015                                         type, cfid);
5016                         if (unlikely(idx < 0))
5017                                 GOTO(stop, rc = idx);
5018
5019                         *lmv2 = lnr->lnr_lmv->ll_lmv;
5020                         lmv2->lmv_magic = LMV_MAGIC_STRIPE;
5021                         lmv2->lmv_master_mdt_index = idx;
5022
5023                         lfsck_lmv_header_cpu_to_le(lmv2, lmv2);
5024                         lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
5025                         rc = dt_declare_xattr_set(env, child, &lmv_buf,
5026                                                   XATTR_NAME_LMV, 0, th);
5027                         if (rc != 0)
5028                                 GOTO(stop, rc);
5029                 }
5030         }
5031
5032         /* 6a. insert linkEA for child */
5033         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
5034                        ldata.ld_leh->leh_len);
5035         rc = dt_declare_xattr_set(env, child, &linkea_buf,
5036                                   XATTR_NAME_LINK, 0, th);
5037         if (rc != 0)
5038                 GOTO(stop, rc);
5039
5040         rc = dt_trans_start_local(env, dev, th);
5041         if (rc != 0)
5042                 GOTO(stop, rc = (rc == -EEXIST ? 1 : rc));
5043
5044         dt_write_lock(env, child, 0);
5045         /* 1b. create child */
5046         rc = dt_create(env, child, la, hint, dof, th);
5047         if (rc != 0)
5048                 GOTO(unlock, rc = (rc == -EEXIST ? 1 : rc));
5049
5050         if (S_ISDIR(type)) {
5051                 /* 2b. increase child nlink */
5052                 rc = dt_ref_add(env, child, th);
5053                 if (rc != 0)
5054                         GOTO(unlock, rc);
5055
5056                 /* 3b. insert dot into child dir */
5057                 rec->rec_type = S_IFDIR;
5058                 rec->rec_fid = cfid;
5059                 rc = dt_insert(env, child, (const struct dt_rec *)rec,
5060                                (const struct dt_key *)dot, th, 1);
5061                 if (rc != 0)
5062                         GOTO(unlock, rc);
5063
5064                 /* 4b. insert dotdot into child dir */
5065                 rec->rec_fid = pfid;
5066                 rc = dt_insert(env, child, (const struct dt_rec *)rec,
5067                                (const struct dt_key *)dotdot, th, 1);
5068                 if (rc != 0)
5069                         GOTO(unlock, rc);
5070
5071                 /* 5b. generate slave LMV EA. */
5072                 if (lnr->lnr_lmv != NULL && lnr->lnr_lmv->ll_lmv_master) {
5073                         rc = dt_xattr_set(env, child, &lmv_buf, XATTR_NAME_LMV,
5074                                           0, th);
5075                         if (rc != 0)
5076                                 GOTO(unlock, rc);
5077                 }
5078         }
5079
5080         /* 6b. insert linkEA for child. */
5081         rc = dt_xattr_set(env, child, &linkea_buf,
5082                           XATTR_NAME_LINK, 0, th);
5083
5084         GOTO(unlock, rc);
5085
5086 unlock:
5087         dt_write_unlock(env, child);
5088
5089 stop:
5090         dt_trans_stop(env, dev, th);
5091
5092 log:
5093         lfsck_unlock(llh);
5094         CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found dangling "
5095                "reference for: parent "DFID", child "DFID", type %u, "
5096                "name %s. %s: rc = %d\n", lfsck_lfsck2name(lfsck),
5097                PFID(pfid), PFID(cfid), type, cname->ln_name,
5098                create ? "Create the lost MDT-object as required" :
5099                         "Keep the MDT-object there by default", rc);
5100
5101         if (rc <= 0) {
5102                 struct lfsck_namespace *ns = com->lc_file_ram;
5103
5104                 ns->ln_flags |= LF_INCONSISTENT;
5105         }
5106
5107         return rc;
5108 }
5109
5110 static int lfsck_namespace_assistant_handler_p1(const struct lu_env *env,
5111                                                 struct lfsck_component *com,
5112                                                 struct lfsck_assistant_req *lar)
5113 {
5114         struct lfsck_thread_info   *info     = lfsck_env_info(env);
5115         struct lu_attr             *la       = &info->lti_la;
5116         struct lfsck_instance      *lfsck    = com->lc_lfsck;
5117         struct lfsck_bookmark      *bk       = &lfsck->li_bookmark_ram;
5118         struct lfsck_namespace     *ns       = com->lc_file_ram;
5119         struct lfsck_assistant_data *lad     = com->lc_data;
5120         struct linkea_data          ldata    = { NULL };
5121         const struct lu_name       *cname;
5122         struct thandle             *handle   = NULL;
5123         struct lfsck_namespace_req *lnr      =
5124                         container_of0(lar, struct lfsck_namespace_req, lnr_lar);
5125         struct dt_object           *dir      = NULL;
5126         struct dt_object           *obj      = NULL;
5127         struct lfsck_assistant_object *lso   = lar->lar_parent;
5128         const struct lu_fid        *pfid     = &lso->lso_fid;
5129         struct dt_device           *dev      = NULL;
5130         struct lustre_handle        lh       = { 0 };
5131         bool                        repaired = false;
5132         bool                        dtlocked = false;
5133         bool                        remove;
5134         bool                        newdata;
5135         bool                        log      = false;
5136         bool                        bad_hash = false;
5137         bool                        bad_linkea = false;
5138         int                         idx      = 0;
5139         int                         count    = 0;
5140         int                         rc       = 0;
5141         enum lfsck_namespace_inconsistency_type type = LNIT_NONE;
5142         ENTRY;
5143
5144         if (lso->lso_dead)
5145                 RETURN(0);
5146
5147         la->la_nlink = 0;
5148         if (lnr->lnr_attr & LUDA_UPGRADE) {
5149                 ns->ln_flags |= LF_UPGRADE;
5150                 ns->ln_dirent_repaired++;
5151                 repaired = true;
5152         } else if (lnr->lnr_attr & LUDA_REPAIR) {
5153                 ns->ln_flags |= LF_INCONSISTENT;
5154                 ns->ln_dirent_repaired++;
5155                 repaired = true;
5156         }
5157
5158         if (unlikely(fid_is_zero(&lnr->lnr_fid))) {
5159                 if (strcmp(lnr->lnr_name, dotdot) != 0)
5160                         LBUG();
5161                 else
5162                         rc = lfsck_namespace_trace_update(env, com, pfid,
5163                                                 LNTF_CHECK_PARENT, true);
5164
5165                 GOTO(out, rc);
5166         }
5167
5168         if (unlikely(!fid_is_sane(&lnr->lnr_fid))) {
5169                 CDEBUG(D_LFSCK, "%s: dir scan find invalid FID "DFID
5170                        " for the name entry %.*s under "DFID"\n",
5171                        lfsck_lfsck2name(lfsck), PFID(&lnr->lnr_fid),
5172                        lnr->lnr_namelen, lnr->lnr_name, PFID(pfid));
5173
5174                 if (strcmp(lnr->lnr_name, dotdot) != 0)
5175                         /* invalid FID means bad name entry, remove it. */
5176                         type = LNIT_BAD_DIRENT;
5177                 else
5178                         /* If the parent FID is invalid, we cannot remove
5179                          * the ".." entry directly. */
5180                         rc = lfsck_namespace_trace_update(env, com, pfid,
5181                                                 LNTF_CHECK_PARENT, true);
5182
5183                 GOTO(out, rc);
5184         }
5185
5186         if (unlikely(lnr->lnr_dir_cookie == MDS_DIR_END_OFF)) {
5187                 rc = lfsck_namespace_striped_dir_rescan(env, com, lnr);
5188
5189                 RETURN(rc);
5190         }
5191
5192         if (fid_seq_is_dot(fid_seq(&lnr->lnr_fid)))
5193                 GOTO(out, rc = 0);
5194
5195         if (lnr->lnr_lmv != NULL && lnr->lnr_lmv->ll_lmv_master) {
5196                 rc = lfsck_namespace_handle_striped_master(env, com, lnr);
5197
5198                 RETURN(rc);
5199         }
5200
5201         idx = lfsck_find_mdt_idx_by_fid(env, lfsck, &lnr->lnr_fid);
5202         if (idx < 0)
5203                 GOTO(out, rc = idx);
5204
5205         if (idx == lfsck_dev_idx(lfsck)) {
5206                 if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
5207                         GOTO(out, rc = 0);
5208
5209                 dev = lfsck->li_bottom;
5210         } else {
5211                 struct lfsck_tgt_desc *ltd;
5212
5213                 /* Usually, some local filesystem consistency verification
5214                  * tools can guarantee the local namespace tree consistenct.
5215                  * So the LFSCK will only verify the remote directory. */
5216                 if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0)) {
5217                         rc = lfsck_namespace_trace_update(env, com, pfid,
5218                                                 LNTF_CHECK_PARENT, true);
5219
5220                         GOTO(out, rc);
5221                 }
5222
5223                 ltd = lfsck_ltd2tgt(&lfsck->li_mdt_descs, idx);
5224                 if (unlikely(ltd == NULL)) {
5225                         CDEBUG(D_LFSCK, "%s: cannot talk with MDT %x which "
5226                                "did not join the namespace LFSCK\n",
5227                                lfsck_lfsck2name(lfsck), idx);
5228                         lfsck_lad_set_bitmap(env, com, idx);
5229
5230                         GOTO(out, rc = -ENODEV);
5231                 }
5232
5233                 dev = ltd->ltd_tgt;
5234         }
5235
5236         obj = lfsck_object_find_by_dev(env, dev, &lnr->lnr_fid);
5237         if (IS_ERR(obj))
5238                 GOTO(out, rc = PTR_ERR(obj));
5239
5240         cname = lfsck_name_get_const(env, lnr->lnr_name, lnr->lnr_namelen);
5241         if (dt_object_exists(obj) == 0) {
5242
5243 dangling:
5244                 if (dir == NULL) {
5245                         dir = lfsck_assistant_object_load(env, lfsck, lso);
5246                         if (IS_ERR(dir)) {
5247                                 rc = PTR_ERR(dir);
5248
5249                                 GOTO(trace, rc == -ENOENT ? 0 : rc);
5250                         }
5251                 }
5252
5253                 rc = lfsck_namespace_check_exist(env, dir, obj, lnr->lnr_name);
5254                 if (rc == 0) {
5255                         if (!lfsck_is_valid_slave_name_entry(env, lnr->lnr_lmv,
5256                                         lnr->lnr_name, lnr->lnr_namelen)) {
5257                                 type = LNIT_BAD_DIRENT;
5258
5259                                 GOTO(out, rc);
5260                         }
5261
5262                         type = LNIT_DANGLING;
5263                         rc = lfsck_namespace_repair_dangling(env, com, dir,
5264                                                              obj, lnr);
5265                         if (rc == 0)
5266                                 repaired = true;
5267                 }
5268
5269                 GOTO(out, rc);
5270         }
5271
5272         if (!(bk->lb_param & LPF_DRYRUN) && lad->lad_advance_lock) {
5273
5274 again:
5275                 rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
5276                                       MDS_INODELOCK_UPDATE |
5277                                       MDS_INODELOCK_XATTR, LCK_EX);
5278                 if (rc != 0)
5279                         GOTO(out, rc);
5280
5281                 handle = dt_trans_create(env, dev);
5282                 if (IS_ERR(handle))
5283                         GOTO(out, rc = PTR_ERR(handle));
5284
5285                 rc = lfsck_declare_namespace_exec_dir(env, obj, handle);
5286                 if (rc != 0)
5287                         GOTO(stop, rc);
5288
5289                 rc = dt_trans_start_local(env, dev, handle);
5290                 if (rc != 0)
5291                         GOTO(stop, rc);
5292
5293                 dt_write_lock(env, obj, 0);
5294                 dtlocked = true;
5295         }
5296
5297         rc = lfsck_links_read(env, obj, &ldata);
5298         if (unlikely(rc == -ENOENT)) {
5299                 if (handle != NULL) {
5300                         dt_write_unlock(env, obj);
5301                         dtlocked = false;
5302
5303                         dt_trans_stop(env, dev, handle);
5304                         handle = NULL;
5305
5306                         lfsck_ibits_unlock(&lh, LCK_EX);
5307                 }
5308
5309                 /* It may happen when the remote object has been removed,
5310                  * but the local MDT is not aware of that. */
5311                 goto dangling;
5312         } else if (rc == 0) {
5313                 count = ldata.ld_leh->leh_reccount;
5314                 rc = linkea_links_find(&ldata, cname, pfid);
5315                 if ((rc == 0) &&
5316                     (count == 1 || !S_ISDIR(lfsck_object_type(obj)))) {
5317                         if ((lfsck_object_type(obj) & S_IFMT) !=
5318                             lnr->lnr_type) {
5319                                 ns->ln_flags |= LF_INCONSISTENT;
5320                                 type = LNIT_BAD_TYPE;
5321                         }
5322
5323                         goto stop;
5324                 }
5325
5326                 /* If the name entry hash does not match the slave striped
5327                  * directory, and the name entry does not match also, then
5328                  * it is quite possible that name entry is corrupted. */
5329                 if (!lfsck_is_valid_slave_name_entry(env, lnr->lnr_lmv,
5330                                         lnr->lnr_name, lnr->lnr_namelen)) {
5331                         ns->ln_flags |= LF_INCONSISTENT;
5332                         type = LNIT_BAD_DIRENT;
5333
5334                         GOTO(stop, rc = 0);
5335                 }
5336
5337                 /* If the file type stored in the name entry does not match
5338                  * the file type claimed by the object, and the object does
5339                  * not recognize the name entry, then it is quite possible
5340                  * that the name entry is corrupted. */
5341                 if ((lfsck_object_type(obj) & S_IFMT) != lnr->lnr_type) {
5342                         ns->ln_flags |= LF_INCONSISTENT;
5343                         type = LNIT_BAD_DIRENT;
5344
5345                         GOTO(stop, rc = 0);
5346                 }
5347
5348                 /* For sub-dir object, we cannot make sure whether the sub-dir
5349                  * back references the parent via ".." name entry correctly or
5350                  * not in the LFSCK first-stage scanning. It may be that the
5351                  * (remote) sub-dir ".." name entry has no parent FID after
5352                  * file-level backup/restore and its linkEA may be wrong.
5353                  * So under such case, we should replace the linkEA according
5354                  * to current name entry. But this needs to be done during the
5355                  * LFSCK second-stage scanning. The LFSCK will record the name
5356                  * entry for further possible using. */
5357                 remove = false;
5358                 newdata = false;
5359                 goto nodata;
5360         } else if (unlikely(rc == -EINVAL)) {
5361                 if ((lfsck_object_type(obj) & S_IFMT) != lnr->lnr_type)
5362                         type = LNIT_BAD_TYPE;
5363
5364                 count = 1;
5365                 /* The magic crashed, we are not sure whether there are more
5366                  * corrupt data in the linkea, so remove all linkea entries. */
5367                 remove = true;
5368                 newdata = true;
5369                 goto nodata;
5370         } else if (rc == -ENODATA) {
5371                 if ((lfsck_object_type(obj) & S_IFMT) != lnr->lnr_type)
5372                         type = LNIT_BAD_TYPE;
5373
5374                 count = 1;
5375                 remove = false;
5376                 newdata = true;
5377
5378 nodata:
5379                 if (bk->lb_param & LPF_DRYRUN) {
5380                         if (rc == -ENODATA)
5381                                 ns->ln_flags |= LF_UPGRADE;
5382                         else
5383                                 ns->ln_flags |= LF_INCONSISTENT;
5384                         ns->ln_linkea_repaired++;
5385                         repaired = true;
5386                         log = true;
5387                         goto stop;
5388                 }
5389
5390                 if (!lustre_handle_is_used(&lh)) {
5391                         remove = false;
5392                         newdata = false;
5393                         type = LNIT_NONE;
5394
5395                         goto again;
5396                 }
5397
5398                 LASSERT(handle != NULL);
5399
5400                 if (dir == NULL) {
5401                         dir = lfsck_assistant_object_load(env, lfsck, lso);
5402                         if (IS_ERR(dir)) {
5403                                 rc = PTR_ERR(dir);
5404
5405                                 GOTO(stop, rc == -ENOENT ? 0 : rc);
5406                         }
5407                 }
5408
5409                 rc = lfsck_namespace_check_exist(env, dir, obj, lnr->lnr_name);
5410                 if (rc != 0)
5411                         GOTO(stop, rc);
5412
5413                 bad_linkea = true;
5414                 if (!remove && newdata)
5415                         ns->ln_flags |= LF_UPGRADE;
5416                 else if (remove || !(ns->ln_flags & LF_UPGRADE))
5417                         ns->ln_flags |= LF_INCONSISTENT;
5418
5419                 if (remove) {
5420                         LASSERT(newdata);
5421
5422                         rc = dt_xattr_del(env, obj, XATTR_NAME_LINK, handle);
5423                         if (rc != 0 && rc != -ENOENT && rc != -ENODATA)
5424                                 GOTO(stop, rc);
5425                 }
5426
5427                 if (newdata) {
5428                         rc = linkea_data_new(&ldata,
5429                                         &lfsck_env_info(env)->lti_linkea_buf);
5430                         if (rc != 0)
5431                                 GOTO(stop, rc);
5432                 }
5433
5434                 rc = linkea_add_buf(&ldata, cname, pfid);
5435                 if (rc == 0)
5436                         rc = lfsck_links_write(env, obj, &ldata, handle);
5437                 if (rc != 0)
5438                         GOTO(stop, rc);
5439
5440                 count = ldata.ld_leh->leh_reccount;
5441                 if (!S_ISDIR(lfsck_object_type(obj)) ||
5442                     !dt_object_remote(obj)) {
5443                         ns->ln_linkea_repaired++;
5444                         repaired = true;
5445                         log = true;
5446                 }
5447         } else {
5448                 GOTO(stop, rc);
5449         }
5450
5451 stop:
5452         if (dtlocked)
5453                 dt_write_unlock(env, obj);
5454
5455         if (handle != NULL && !IS_ERR(handle))
5456                 dt_trans_stop(env, dev, handle);
5457
5458 out:
5459         lfsck_ibits_unlock(&lh, LCK_EX);
5460
5461         if (!name_is_dot_or_dotdot(lnr->lnr_name, lnr->lnr_namelen) &&
5462             !lfsck_is_valid_slave_name_entry(env, lnr->lnr_lmv,
5463                                              lnr->lnr_name, lnr->lnr_namelen) &&
5464             type != LNIT_BAD_DIRENT) {
5465                 ns->ln_flags |= LF_INCONSISTENT;
5466
5467                 log = false;
5468                 if (dir == NULL) {
5469                         dir = lfsck_assistant_object_load(env, lfsck, lso);
5470                         if (IS_ERR(dir)) {
5471                                 rc = PTR_ERR(dir);
5472
5473                                 GOTO(trace, rc == -ENOENT ? 0 : rc);
5474                         }
5475                 }
5476
5477                 rc = lfsck_namespace_repair_bad_name_hash(env, com, dir,
5478                                                 lnr->lnr_lmv, lnr->lnr_name);
5479                 if (rc == 0)
5480                         bad_hash = true;
5481         }
5482
5483         if (rc >= 0) {
5484                 if (type != LNIT_NONE && dir == NULL) {
5485                         dir = lfsck_assistant_object_load(env, lfsck, lso);
5486                         if (IS_ERR(dir)) {
5487                                 rc = PTR_ERR(dir);
5488
5489                                 GOTO(trace, rc == -ENOENT ? 0 : rc);
5490                         }
5491                 }
5492
5493                 switch (type) {
5494                 case LNIT_BAD_TYPE:
5495                         log = false;
5496                         rc = lfsck_namespace_repair_dirent(env, com, dir,
5497                                         obj, lnr->lnr_name, lnr->lnr_name,
5498                                         lnr->lnr_type, true, false);
5499                         if (rc > 0)
5500                                 repaired = true;
5501                         break;
5502                 case LNIT_BAD_DIRENT:
5503                         log = false;
5504                         /* XXX: This is a bad dirent, we do not know whether
5505                          *      the original name entry reference a regular
5506                          *      file or a directory, then keep the parent's
5507                          *      nlink count unchanged here. */
5508                         rc = lfsck_namespace_repair_dirent(env, com, dir,
5509                                         obj, lnr->lnr_name, lnr->lnr_name,
5510                                         lnr->lnr_type, false, false);
5511                         if (rc > 0)
5512                                 repaired = true;
5513                         break;
5514                 default:
5515                         break;
5516                 }
5517
5518                 if (obj != NULL && count == 1 &&
5519                     S_ISREG(lfsck_object_type(obj)))
5520                         dt_attr_get(env, obj, la);
5521         }
5522
5523 trace:
5524         down_write(&com->lc_sem);
5525         if (rc < 0) {
5526                 CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant fail to handle "
5527                        "the entry: "DFID", parent "DFID", name %.*s: rc = %d\n",
5528                        lfsck_lfsck2name(lfsck), PFID(&lnr->lnr_fid), PFID(pfid),
5529                        lnr->lnr_namelen, lnr->lnr_name, rc);
5530
5531                 lfsck_namespace_record_failure(env, lfsck, ns);
5532                 if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
5533                      rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
5534                      rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&
5535                     dev != NULL && dev != lfsck->li_bottom)
5536                         lfsck_lad_set_bitmap(env, com, idx);
5537
5538                 if (!(bk->lb_param & LPF_FAILOUT))
5539                         rc = 0;
5540         } else {
5541                 if (repaired) {
5542                         ns->ln_items_repaired++;
5543                         if (log)
5544                                 CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant "
5545                                        "repaired the entry: "DFID", parent "DFID
5546                                        ", name %.*s, type %d\n",
5547                                        lfsck_lfsck2name(lfsck),
5548                                        PFID(&lnr->lnr_fid), PFID(pfid),
5549                                        lnr->lnr_namelen, lnr->lnr_name, type);
5550
5551                         switch (type) {
5552                         case LNIT_DANGLING:
5553                                 ns->ln_dangling_repaired++;
5554                                 break;
5555                         case LNIT_BAD_TYPE:
5556                                 ns->ln_bad_type_repaired++;
5557                                 break;
5558                         case LNIT_BAD_DIRENT:
5559                                 ns->ln_dirent_repaired++;
5560                                 break;
5561                         default:
5562                                 break;
5563                         }
5564
5565                         if (bk->lb_param & LPF_DRYRUN &&
5566                             lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent))
5567                                 lfsck_pos_fill(env, lfsck,
5568                                                &ns->ln_pos_first_inconsistent,
5569                                                false);
5570                 }
5571
5572                 if (bad_hash) {
5573                         ns->ln_name_hash_repaired++;
5574
5575                         /* Not count repeatedly. */
5576                         if (!repaired) {
5577                                 ns->ln_items_repaired++;
5578                                 if (log)
5579                                         CDEBUG(D_LFSCK, "%s: namespace LFSCK "
5580                                                "assistant repaired the entry: "
5581                                                DFID", parent "DFID
5582                                                ", name %.*s\n",
5583                                                lfsck_lfsck2name(lfsck),
5584                                                PFID(&lnr->lnr_fid), PFID(pfid),
5585                                                lnr->lnr_namelen, lnr->lnr_name);
5586                         }
5587
5588                         if (bk->lb_param & LPF_DRYRUN &&
5589                             lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent))
5590                                 lfsck_pos_fill(env, lfsck,
5591                                                &ns->ln_pos_first_inconsistent,
5592                                                false);
5593                 }
5594
5595                 rc = 0;
5596         }
5597
5598         if (count > 1 || la->la_nlink > 1)
5599                 ns->ln_mul_linked_checked++;
5600
5601         up_write(&com->lc_sem);
5602
5603         if (obj != NULL && !IS_ERR(obj))
5604                 lfsck_object_put(env, obj);
5605
5606         if (dir != NULL && !IS_ERR(dir))
5607                 lfsck_object_put(env, dir);
5608
5609         lad->lad_advance_lock = bad_linkea;
5610
5611         return rc;
5612 }
5613
5614 /**
5615  * Handle one orphan under the backend /lost+found directory
5616  *
5617  * Insert the orphan FID into the namespace LFSCK trace file for further
5618  * processing (via the subsequent namespace LFSCK second-stage scanning).
5619  * At the same time, remove the orphan name entry from backend /lost+found
5620  * directory. There is an interval between the orphan name entry removed
5621  * from the backend /lost+found directory and the orphan FID in the LFSCK
5622  * trace file handled. In such interval, the LFSCK can be reset, then
5623  * all the FIDs recorded in the namespace LFSCK trace file will be dropped.
5624  * To guarantee that the orphans can be found when LFSCK run next time
5625  * without e2fsck again, when remove the orphan name entry, the LFSCK
5626  * will set the orphan's ctime attribute as 1. Since normal applications
5627  * cannot change the object's ctime attribute as 1. Then when LFSCK run
5628  * next time, it can record the object (that ctime is 1) in the namespace
5629  * LFSCK trace file during the first-stage scanning.
5630  *
5631  * \param[in] env       pointer to the thread context
5632  * \param[in] com       pointer to the lfsck component
5633  * \param[in] parent    pointer to the object for the backend /lost+found
5634  * \param[in] ent       pointer to the name entry for the target under the
5635  *                      backend /lost+found
5636  *
5637  * \retval              positive for repaired
5638  * \retval              0 if needs to repair nothing
5639  * \retval              negative error number on failure
5640  */
5641 static int lfsck_namespace_scan_local_lpf_one(const struct lu_env *env,
5642                                               struct lfsck_component *com,
5643                                               struct dt_object *parent,
5644                                               struct lu_dirent *ent)
5645 {
5646         struct lfsck_thread_info        *info   = lfsck_env_info(env);
5647         struct lu_fid                   *key    = &info->lti_fid;
5648         struct lu_attr                  *la     = &info->lti_la;
5649         struct lfsck_instance           *lfsck  = com->lc_lfsck;
5650         struct dt_object                *obj;
5651         struct dt_device                *dev    = lfsck->li_bottom;
5652         struct dt_object                *child  = NULL;
5653         struct thandle                  *th     = NULL;
5654         int                              idx;
5655         int                              rc     = 0;
5656         __u8                             flags  = 0;
5657         bool                             exist  = false;
5658         ENTRY;
5659
5660         child = lfsck_object_find_by_dev(env, dev, &ent->lde_fid);
5661         if (IS_ERR(child))
5662                 RETURN(PTR_ERR(child));
5663
5664         LASSERT(dt_object_exists(child));
5665         LASSERT(!dt_object_remote(child));
5666
5667         idx = lfsck_sub_trace_file_fid2idx(&ent->lde_fid);
5668         obj = com->lc_sub_trace_objs[idx].lsto_obj;
5669         fid_cpu_to_be(key, &ent->lde_fid);
5670         rc = dt_lookup(env, obj, (struct dt_rec *)&flags,
5671                        (const struct dt_key *)key);
5672         if (rc == 0) {
5673                 exist = true;
5674                 flags |= LNTF_CHECK_ORPHAN;
5675         } else if (rc == -ENOENT) {
5676                 flags = LNTF_CHECK_ORPHAN;
5677         } else {
5678                 GOTO(out, rc);
5679         }
5680
5681         th = dt_trans_create(env, dev);
5682         if (IS_ERR(th))
5683                 GOTO(out, rc = PTR_ERR(th));
5684
5685         /* a1. remove name entry from backend /lost+found */
5686         rc = dt_declare_delete(env, parent,
5687                                (const struct dt_key *)ent->lde_name, th);
5688         if (rc != 0)
5689                 GOTO(stop, rc);
5690
5691         if (S_ISDIR(lfsck_object_type(child))) {
5692                 /* a2. decrease parent's nlink */
5693                 rc = dt_declare_ref_del(env, parent, th);
5694                 if (rc != 0)
5695                         GOTO(stop, rc);
5696         }
5697
5698         if (exist) {
5699                 /* a3. remove child's FID from the LFSCK trace file. */
5700                 rc = dt_declare_delete(env, obj,
5701                                        (const struct dt_key *)key, th);
5702                 if (rc != 0)
5703                         GOTO(stop, rc);
5704         } else {
5705                 /* a4. set child's ctime as 1 */
5706                 memset(la, 0, sizeof(*la));
5707                 la->la_ctime = 1;
5708                 la->la_valid = LA_CTIME;
5709                 rc = dt_declare_attr_set(env, child, la, th);
5710                 if (rc != 0)
5711                         GOTO(stop, rc);
5712         }
5713
5714         /* a5. insert child's FID into the LFSCK trace file. */
5715         rc = dt_declare_insert(env, obj, (const struct dt_rec *)&flags,
5716                                (const struct dt_key *)key, th);
5717         if (rc != 0)
5718                 GOTO(stop, rc);
5719
5720         rc = dt_trans_start_local(env, dev, th);
5721         if (rc != 0)
5722                 GOTO(stop, rc);
5723
5724         /* b1. remove name entry from backend /lost+found */
5725         rc = dt_delete(env, parent, (const struct dt_key *)ent->lde_name, th);
5726         if (rc != 0)
5727                 GOTO(stop, rc);
5728
5729         if (S_ISDIR(lfsck_object_type(child))) {
5730                 /* b2. decrease parent's nlink */
5731                 dt_write_lock(env, parent, 0);
5732                 rc = dt_ref_del(env, parent, th);
5733                 dt_write_unlock(env, parent);
5734                 if (rc != 0)
5735                         GOTO(stop, rc);
5736         }
5737
5738         if (exist) {
5739                 /* a3. remove child's FID from the LFSCK trace file. */
5740                 rc = dt_delete(env, obj, (const struct dt_key *)key, th);
5741                 if (rc != 0)
5742                         GOTO(stop, rc);
5743         } else {
5744                 /* b4. set child's ctime as 1 */
5745                 rc = dt_attr_set(env, child, la, th);
5746                 if (rc != 0)
5747                         GOTO(stop, rc);
5748         }
5749
5750         /* b5. insert child's FID into the LFSCK trace file. */
5751         rc = dt_insert(env, obj, (const struct dt_rec *)&flags,
5752                        (const struct dt_key *)key, th, 1);
5753
5754         GOTO(stop, rc = (rc == 0 ? 1 : rc));
5755
5756 stop:
5757         dt_trans_stop(env, dev, th);
5758
5759 out:
5760         lfsck_object_put(env, child);
5761
5762         return rc;
5763 }
5764
5765 /**
5766  * Handle orphans under the backend /lost+found directory
5767  *
5768  * Some backend checker, such as e2fsck for ldiskfs may find some orphans
5769  * and put them under the backend /lost+found directory that is invisible
5770  * to client. The LFSCK will scan such directory, for the original client
5771  * visible orphans, add their fids into the namespace LFSCK trace file,
5772  * then the subsenquent namespace LFSCK second-stage scanning can handle
5773  * them as other objects to be double scanned: either move back to normal
5774  * namespace, or to the global visible orphan directory:
5775  * /ROOT/.lustre/lost+found/MDTxxxx/
5776  *
5777  * \param[in] env       pointer to the thread context
5778  * \param[in] com       pointer to the lfsck component
5779  */
5780 static void lfsck_namespace_scan_local_lpf(const struct lu_env *env,
5781                                            struct lfsck_component *com)
5782 {
5783         struct lfsck_thread_info        *info   = lfsck_env_info(env);
5784         struct lu_dirent                *ent    =
5785                                         (struct lu_dirent *)info->lti_key;
5786         struct lu_seq_range             *range  = &info->lti_range;
5787         struct lfsck_instance           *lfsck  = com->lc_lfsck;
5788         struct ptlrpc_thread            *thread = &lfsck->li_thread;
5789         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
5790         struct lfsck_namespace          *ns     = com->lc_file_ram;
5791         struct dt_object                *parent;
5792         const struct dt_it_ops          *iops;
5793         struct dt_it                    *di;
5794         struct seq_server_site          *ss     = lfsck_dev_site(lfsck);
5795         __u64                            cookie;
5796         __u32                            idx    = lfsck_dev_idx(lfsck);
5797         int                              rc     = 0;
5798         __u16                            type;
5799         ENTRY;
5800
5801         parent = lfsck_object_find_by_dev(env, lfsck->li_bottom,
5802                                           &LU_BACKEND_LPF_FID);
5803         if (IS_ERR(parent)) {
5804                 CERROR("%s: fail to find backend /lost+found: rc = %ld\n",
5805                        lfsck_lfsck2name(lfsck), PTR_ERR(parent));
5806                 RETURN_EXIT;
5807         }
5808
5809         /* It is normal that the /lost+found does not exist for ZFS backend. */
5810         if (!dt_object_exists(parent))
5811                 GOTO(out, rc = 0);
5812
5813         if (unlikely(!dt_try_as_dir(env, parent)))
5814                 GOTO(out, rc = -ENOTDIR);
5815
5816         CDEBUG(D_LFSCK, "%s: start to scan backend /lost+found\n",
5817                lfsck_lfsck2name(lfsck));
5818
5819         com->lc_new_scanned = 0;
5820         iops = &parent->do_index_ops->dio_it;
5821         di = iops->init(env, parent, LUDA_64BITHASH | LUDA_TYPE);
5822         if (IS_ERR(di))
5823                 GOTO(out, rc = PTR_ERR(di));
5824
5825         rc = iops->load(env, di, 0);
5826         if (rc == 0)
5827                 rc = iops->next(env, di);
5828         else if (rc > 0)
5829                 rc = 0;
5830
5831         while (rc == 0) {
5832                 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
5833                     unlikely(!thread_is_running(thread)))
5834                         break;
5835
5836                 rc = iops->rec(env, di, (struct dt_rec *)ent,
5837                                LUDA_64BITHASH | LUDA_TYPE);
5838                 if (rc == 0)
5839                         rc = lfsck_unpack_ent(ent, &cookie, &type);
5840
5841                 if (unlikely(rc != 0)) {
5842                         CDEBUG(D_LFSCK, "%s: fail to iterate backend "
5843                                "/lost+found: rc = %d\n",
5844                                lfsck_lfsck2name(lfsck), rc);
5845
5846                         goto skip;
5847                 }
5848
5849                 /* skip dot and dotdot entries */
5850                 if (name_is_dot_or_dotdot(ent->lde_name, ent->lde_namelen))
5851                         goto next;
5852
5853                 if (!fid_seq_in_fldb(fid_seq(&ent->lde_fid)))
5854                         goto skip;
5855
5856                 if (fid_is_norm(&ent->lde_fid)) {
5857                         fld_range_set_mdt(range);
5858                         rc = fld_local_lookup(env, ss->ss_server_fld,
5859                                               fid_seq(&ent->lde_fid), range);
5860                         if (rc != 0)
5861                                 goto skip;
5862                 } else if (idx != 0) {
5863                         /* If the returned FID is IGIF, then there are three
5864                          * possible cases:
5865                          *
5866                          * 1) The object is upgraded from old Lustre-1.8 with
5867                          *    IGIF assigned to such object.
5868                          * 2) The object is a backend local object and is
5869                          *    invisible to client.
5870                          * 3) The object lost its LMV EA, and since there is
5871                          *    no FID-in-dirent for the orphan in the backend
5872                          *    /lost+found directory, then the low layer will
5873                          *    return IGIF for such object.
5874                          *
5875                          * For MDTx (x != 0), it is either case 2) or case 3),
5876                          * but from the LFSCK view, they are indistinguishable.
5877                          * To be safe, the LFSCK will keep it there and report
5878                          * some message, then the adminstrator can handle that
5879                          * furtherly.
5880                          *
5881                          * For MDT0, it is more possible the case 1). The LFSCK
5882                          * will handle the orphan as an upgraded object. */
5883                         CDEBUG(D_LFSCK, "%s: the orphan %.*s with IGIF "DFID
5884                                "in the backend /lost+found on the MDT %04x, "
5885                                "to be safe, skip it.\n",
5886                                lfsck_lfsck2name(lfsck), ent->lde_namelen,
5887                                ent->lde_name, PFID(&ent->lde_fid), idx);
5888                         goto skip;
5889                 }
5890
5891                 rc = lfsck_namespace_scan_local_lpf_one(env, com, parent, ent);
5892
5893 skip:
5894                 down_write(&com->lc_sem);
5895                 com->lc_new_scanned++;
5896                 ns->ln_local_lpf_scanned++;
5897                 if (rc > 0)
5898                         ns->ln_local_lpf_moved++;
5899                 else if (rc == 0)
5900                         ns->ln_local_lpf_skipped++;
5901                 else
5902                         ns->ln_local_lpf_failed++;
5903                 up_write(&com->lc_sem);
5904
5905                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
5906                         break;
5907
5908 next:
5909                 lfsck_control_speed_by_self(com);
5910                 if (unlikely(!thread_is_running(thread))) {
5911                         rc = 0;
5912                         break;
5913                 }
5914
5915                 rc = iops->next(env, di);
5916         }
5917
5918         iops->put(env, di);
5919         iops->fini(env, di);
5920
5921         EXIT;
5922
5923 out:
5924         CDEBUG(D_LFSCK, "%s: stop to scan backend /lost+found: rc = %d\n",
5925                lfsck_lfsck2name(lfsck), rc);
5926
5927         lfsck_object_put(env, parent);
5928 }
5929
5930 /**
5931  * Rescan the striped directory after the master LMV EA reset.
5932  *
5933  * Sometimes, the master LMV EA of the striped directory maybe lost, so when
5934  * the namespace LFSCK engine scan the striped directory for the first time,
5935  * it will be regarded as a normal directory. As the LFSCK processing, some
5936  * other LFSCK instance on other MDT will find the shard of this striped dir,
5937  * and find that the master MDT-object of the striped directory lost its LMV
5938  * EA, then such remote LFSCK instance will regenerate the master LMV EA and
5939  * notify the LFSCK instance on this MDT to rescan the striped directory.
5940  *
5941  * \param[in] env       pointer to the thread context
5942  * \param[in] com       pointer to the lfsck component
5943  * \param[in] llu       the lfsck_lmv_unit that contains the striped directory
5944  *                      to be rescanned.
5945  *
5946  * \retval              positive number for success
5947  * \retval              0 for LFSCK stopped/paused
5948  * \retval              negative error number on failure
5949  */
5950 static int lfsck_namespace_rescan_striped_dir(const struct lu_env *env,
5951                                               struct lfsck_component *com,
5952                                               struct lfsck_lmv_unit *llu)
5953 {
5954         struct lfsck_thread_info        *info   = lfsck_env_info(env);
5955         struct lfsck_instance           *lfsck  = com->lc_lfsck;
5956         struct lfsck_assistant_data     *lad    = com->lc_data;
5957         struct dt_object                *dir;
5958         const struct dt_it_ops          *iops;
5959         struct dt_it                    *di;
5960         struct lu_dirent                *ent    =
5961                         (struct lu_dirent *)info->lti_key;
5962         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
5963         struct ptlrpc_thread            *thread = &lfsck->li_thread;
5964         struct lfsck_assistant_object   *lso    = NULL;
5965         struct lfsck_namespace_req      *lnr;
5966         struct lfsck_assistant_req      *lar;
5967         int                              rc;
5968         __u16                            type;
5969         ENTRY;
5970
5971         LASSERT(list_empty(&lad->lad_req_list));
5972
5973         lfsck->li_lmv = &llu->llu_lmv;
5974         lfsck->li_obj_dir = lfsck_object_get(llu->llu_obj);
5975         rc = lfsck_open_dir(env, lfsck, 0);
5976         if (rc != 0)
5977                 RETURN(rc);
5978
5979         dir = lfsck->li_obj_dir;
5980         di = lfsck->li_di_dir;
5981         iops = &dir->do_index_ops->dio_it;
5982         do {
5983                 rc = iops->rec(env, di, (struct dt_rec *)ent,
5984                                lfsck->li_args_dir);
5985                 if (rc == 0)
5986                         rc = lfsck_unpack_ent(ent, &lfsck->li_cookie_dir,
5987                                               &type);
5988
5989                 if (rc != 0) {
5990                         if (bk->lb_param & LPF_FAILOUT)
5991                                 GOTO(out, rc);
5992
5993                         goto next;
5994                 }
5995
5996                 if (name_is_dot_or_dotdot(ent->lde_name, ent->lde_namelen))
5997                         goto next;
5998
5999                 if (lso == NULL) {
6000                         lso = lfsck_assistant_object_init(env,
6001                                 lfsck_dto2fid(dir), NULL,
6002                                 lfsck->li_pos_current.lp_oit_cookie, true);
6003                         if (IS_ERR(lso)) {
6004                                 if (bk->lb_param & LPF_FAILOUT)
6005                                         GOTO(out, rc = PTR_ERR(lso));
6006
6007                                 lso = NULL;
6008                                 goto next;
6009                         }
6010                 }
6011
6012                 lnr = lfsck_namespace_assistant_req_init(lfsck, lso, ent, type);
6013                 if (IS_ERR(lnr)) {
6014                         if (bk->lb_param & LPF_FAILOUT)
6015                                 GOTO(out, rc = PTR_ERR(lnr));
6016
6017                         goto next;
6018                 }
6019
6020                 lar = &lnr->lnr_lar;
6021                 rc = lfsck_namespace_assistant_handler_p1(env, com, lar);
6022                 lfsck_namespace_assistant_req_fini(env, lar);
6023                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
6024                         GOTO(out, rc);
6025
6026                 if (unlikely(!thread_is_running(thread)))
6027                         GOTO(out, rc = 0);
6028
6029 next:
6030                 rc = iops->next(env, di);
6031         } while (rc == 0);
6032
6033 out:
6034         if (lso != NULL && !IS_ERR(lso))
6035                 lfsck_assistant_object_put(env, lso);
6036
6037         lfsck_close_dir(env, lfsck, rc);
6038         if (rc <= 0)
6039                 RETURN(rc);
6040
6041         /* The close_dir() may insert a dummy lnr in the lad->lad_req_list. */
6042         if (list_empty(&lad->lad_req_list))
6043                 RETURN(1);
6044
6045         spin_lock(&lad->lad_lock);
6046         lar = list_entry(lad->lad_req_list.next, struct lfsck_assistant_req,
6047                           lar_list);
6048         list_del_init(&lar->lar_list);
6049         spin_unlock(&lad->lad_lock);
6050
6051         rc = lfsck_namespace_assistant_handler_p1(env, com, lar);
6052         lfsck_namespace_assistant_req_fini(env, lar);
6053
6054         RETURN(rc == 0 ? 1 : rc);
6055 }
6056
6057 static int
6058 lfsck_namespace_double_scan_one_trace_file(const struct lu_env *env,
6059                                            struct lfsck_component *com,
6060                                            struct dt_object *obj, bool first)
6061 {
6062         struct lfsck_instance   *lfsck  = com->lc_lfsck;
6063         struct ptlrpc_thread    *thread = &lfsck->li_thread;
6064         struct lfsck_bookmark   *bk     = &lfsck->li_bookmark_ram;
6065         struct lfsck_namespace  *ns     = com->lc_file_ram;
6066         const struct dt_it_ops  *iops   = &obj->do_index_ops->dio_it;
6067         struct dt_object        *target;
6068         struct dt_it            *di;
6069         struct dt_key           *key;
6070         struct lu_fid            fid;
6071         int                      rc;
6072         __u8                     flags  = 0;
6073         ENTRY;
6074
6075         di = iops->init(env, obj, 0);
6076         if (IS_ERR(di))
6077                 RETURN(PTR_ERR(di));
6078
6079         if (first)
6080                 fid_cpu_to_be(&fid, &ns->ln_fid_latest_scanned_phase2);
6081         else
6082                 fid_zero(&fid);
6083         rc = iops->get(env, di, (const struct dt_key *)&fid);
6084         if (rc < 0)
6085                 GOTO(fini, rc);
6086
6087         if (first) {
6088                 /* The start one either has been processed or does not exist,
6089                  * skip it. */
6090                 rc = iops->next(env, di);
6091                 if (rc != 0)
6092                         GOTO(put, rc);
6093         }
6094
6095         do {
6096                 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
6097                     unlikely(!thread_is_running(thread)))
6098                         GOTO(put, rc = 0);
6099
6100                 key = iops->key(env, di);
6101                 if (IS_ERR(key)) {
6102                         rc = PTR_ERR(key);
6103                         if (rc == -ENOENT)
6104                                 GOTO(put, rc = 1);
6105
6106                         goto checkpoint;
6107                 }
6108
6109                 fid_be_to_cpu(&fid, (const struct lu_fid *)key);
6110                 if (!fid_is_sane(&fid)) {
6111                         rc = 0;
6112                         goto checkpoint;
6113                 }
6114
6115                 target = lfsck_object_find_bottom(env, lfsck, &fid);
6116                 if (IS_ERR(target)) {
6117                         rc = PTR_ERR(target);
6118                         goto checkpoint;
6119                 }
6120
6121                 if (dt_object_exists(target)) {
6122                         rc = iops->rec(env, di, (struct dt_rec *)&flags, 0);
6123                         if (rc == 0) {
6124                                 rc = lfsck_namespace_double_scan_one(env, com,
6125                                                                 target, flags);
6126                                 if (rc == -ENOENT)
6127                                         rc = 0;
6128                         }
6129                 }
6130
6131                 lfsck_object_put(env, target);
6132
6133 checkpoint:
6134                 down_write(&com->lc_sem);
6135                 com->lc_new_checked++;
6136                 com->lc_new_scanned++;
6137                 if (rc >= 0)
6138                         ns->ln_fid_latest_scanned_phase2 = fid;
6139
6140                 if (rc > 0)
6141                         ns->ln_objs_repaired_phase2++;
6142                 else if (rc < 0)
6143                         ns->ln_objs_failed_phase2++;
6144                 up_write(&com->lc_sem);
6145
6146                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
6147                         GOTO(put, rc);
6148
6149                 if (unlikely(com->lc_time_next_checkpoint <=
6150                              ktime_get_seconds()) &&
6151                     com->lc_new_checked != 0) {
6152                         down_write(&com->lc_sem);
6153                         ns->ln_run_time_phase2 += ktime_get_seconds() -
6154                                                   com->lc_time_last_checkpoint;
6155                         ns->ln_time_last_checkpoint = ktime_get_real_seconds();
6156                         ns->ln_objs_checked_phase2 += com->lc_new_checked;
6157                         com->lc_new_checked = 0;
6158                         lfsck_namespace_store(env, com);
6159                         up_write(&com->lc_sem);
6160
6161                         com->lc_time_last_checkpoint = ktime_get_seconds();
6162                         com->lc_time_next_checkpoint =
6163                                 com->lc_time_last_checkpoint +
6164                                 LFSCK_CHECKPOINT_INTERVAL;
6165                 }
6166
6167                 lfsck_control_speed_by_self(com);
6168                 if (unlikely(!thread_is_running(thread)))
6169                         GOTO(put, rc = 0);
6170
6171                 rc = iops->next(env, di);
6172         } while (rc == 0);
6173
6174         GOTO(put, rc);
6175
6176 put:
6177         iops->put(env, di);
6178
6179 fini:
6180         iops->fini(env, di);
6181
6182         return rc;
6183 }
6184
6185 static int lfsck_namespace_assistant_handler_p2(const struct lu_env *env,
6186                                                 struct lfsck_component *com)
6187 {
6188         struct lfsck_instance   *lfsck  = com->lc_lfsck;
6189         struct lfsck_namespace  *ns     = com->lc_file_ram;
6190         int                      rc;
6191         int                      i;
6192         ENTRY;
6193
6194         while (!list_empty(&lfsck->li_list_lmv)) {
6195                 struct lfsck_lmv_unit *llu;
6196
6197                 spin_lock(&lfsck->li_lock);
6198                 llu = list_entry(lfsck->li_list_lmv.next,
6199                                  struct lfsck_lmv_unit, llu_link);
6200                 list_del_init(&llu->llu_link);
6201                 spin_unlock(&lfsck->li_lock);
6202
6203                 rc = lfsck_namespace_rescan_striped_dir(env, com, llu);
6204                 if (rc <= 0)
6205                         RETURN(rc);
6206         }
6207
6208         CDEBUG(D_LFSCK, "%s: namespace LFSCK phase2 scan start\n",
6209                lfsck_lfsck2name(lfsck));
6210
6211         lfsck_namespace_scan_local_lpf(env, com);
6212
6213         com->lc_new_checked = 0;
6214         com->lc_new_scanned = 0;
6215         com->lc_time_last_checkpoint = ktime_get_seconds();
6216         com->lc_time_next_checkpoint = com->lc_time_last_checkpoint +
6217                                        LFSCK_CHECKPOINT_INTERVAL;
6218
6219         i = lfsck_sub_trace_file_fid2idx(&ns->ln_fid_latest_scanned_phase2);
6220         rc = lfsck_namespace_double_scan_one_trace_file(env, com,
6221                                 com->lc_sub_trace_objs[i].lsto_obj, true);
6222         while (rc > 0 && ++i < LFSCK_STF_COUNT)
6223                 rc = lfsck_namespace_double_scan_one_trace_file(env, com,
6224                                 com->lc_sub_trace_objs[i].lsto_obj, false);
6225
6226         CDEBUG(D_LFSCK, "%s: namespace LFSCK phase2 scan stop at the No. %d "
6227                "trace file: rc = %d\n", lfsck_lfsck2name(lfsck), i, rc);
6228
6229         RETURN(rc);
6230 }
6231
6232 static void lfsck_namespace_assistant_fill_pos(const struct lu_env *env,
6233                                                struct lfsck_component *com,
6234                                                struct lfsck_position *pos)
6235 {
6236         struct lfsck_assistant_data     *lad = com->lc_data;
6237         struct lfsck_namespace_req      *lnr;
6238
6239         if (((struct lfsck_namespace *)(com->lc_file_ram))->ln_status !=
6240             LS_SCANNING_PHASE1)
6241                 return;
6242
6243         if (list_empty(&lad->lad_req_list))
6244                 return;
6245
6246         lnr = list_entry(lad->lad_req_list.next,
6247                          struct lfsck_namespace_req,
6248                          lnr_lar.lar_list);
6249         pos->lp_oit_cookie = lnr->lnr_lar.lar_parent->lso_oit_cookie;
6250         pos->lp_dir_cookie = lnr->lnr_dir_cookie - 1;
6251         pos->lp_dir_parent = lnr->lnr_lar.lar_parent->lso_fid;
6252 }
6253
6254 static int lfsck_namespace_double_scan_result(const struct lu_env *env,
6255                                               struct lfsck_component *com,
6256                                               int rc)
6257 {
6258         struct lfsck_instance   *lfsck  = com->lc_lfsck;
6259         struct lfsck_namespace  *ns     = com->lc_file_ram;
6260
6261         down_write(&com->lc_sem);
6262         ns->ln_run_time_phase2 += ktime_get_seconds() -
6263                                   com->lc_time_last_checkpoint;
6264         ns->ln_time_last_checkpoint = ktime_get_real_seconds();
6265         ns->ln_objs_checked_phase2 += com->lc_new_checked;
6266         com->lc_new_checked = 0;
6267
6268         if (rc > 0) {
6269                 if (ns->ln_flags & LF_INCOMPLETE)
6270                         ns->ln_status = LS_PARTIAL;
6271                 else
6272                         ns->ln_status = LS_COMPLETED;
6273                 ns->ln_flags &= ~LF_SCANNED_ONCE;
6274                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN))
6275                         ns->ln_flags &= ~LF_INCONSISTENT;
6276                 ns->ln_time_last_complete = ns->ln_time_last_checkpoint;
6277                 ns->ln_success_count++;
6278         } else if (rc == 0) {
6279                 if (lfsck->li_status != 0)
6280                         ns->ln_status = lfsck->li_status;
6281                 else
6282                         ns->ln_status = LS_STOPPED;
6283         } else {
6284                 ns->ln_status = LS_FAILED;
6285         }
6286
6287         rc = lfsck_namespace_store(env, com);
6288         up_write(&com->lc_sem);
6289
6290         return rc;
6291 }
6292
6293 static int
6294 lfsck_namespace_assistant_sync_failures_interpret(const struct lu_env *env,
6295                                                   struct ptlrpc_request *req,
6296                                                   void *args, int rc)
6297 {
6298         if (rc == 0) {
6299                 struct lfsck_async_interpret_args *laia = args;
6300                 struct lfsck_tgt_desc             *ltd  = laia->laia_ltd;
6301
6302                 ltd->ltd_synced_failures = 1;
6303         }
6304
6305         return 0;
6306 }
6307
6308 /**
6309  * Notify remote LFSCK instances about former failures.
6310  *
6311  * The local LFSCK instance has recorded which MDTs have ever failed to respond
6312  * some LFSCK verification requests (maybe because of network issues or the MDT
6313  * itself trouble). During the respond gap the MDT may missed some name entries
6314  * verification, then the MDT cannot know whether related MDT-objects have been
6315  * referenced by related name entries or not, then in the second-stage scanning,
6316  * these MDT-objects will be regarded as orphan, if the MDT-object contains bad
6317  * linkEA for back reference, then it will misguide the LFSCK to generate wrong
6318  * name entry for repairing the orphan.
6319  *
6320  * To avoid above trouble, when layout LFSCK finishes the first-stage scanning,
6321  * it will scan the bitmap for the ever failed MDTs, and notify them that they
6322  * have ever missed some name entries verification and should skip the handling
6323  * for orphan MDT-objects.
6324  *
6325  * \param[in] env       pointer to the thread context
6326  * \param[in] com       pointer to the lfsck component
6327  * \param[in] lr        pointer to the lfsck request
6328  */
6329 static void lfsck_namespace_assistant_sync_failures(const struct lu_env *env,
6330                                                     struct lfsck_component *com,
6331                                                     struct lfsck_request *lr)
6332 {
6333         struct lfsck_async_interpret_args *laia  =
6334                                 &lfsck_env_info(env)->lti_laia2;
6335         struct lfsck_assistant_data       *lad   = com->lc_data;
6336         struct lfsck_namespace            *ns    = com->lc_file_ram;
6337         struct lfsck_instance             *lfsck = com->lc_lfsck;
6338         struct lfsck_tgt_descs            *ltds  = &lfsck->li_mdt_descs;
6339         struct lfsck_tgt_desc             *ltd;
6340         struct ptlrpc_request_set         *set;
6341         __u32                              idx;
6342         int                                rc    = 0;
6343         ENTRY;
6344
6345         if (!lad->lad_incomplete)
6346                 RETURN_EXIT;
6347
6348         set = ptlrpc_prep_set();
6349         if (set == NULL)
6350                 GOTO(out, rc = -ENOMEM);
6351
6352         lr->lr_flags2 = ns->ln_flags | LF_INCOMPLETE;
6353         memset(laia, 0, sizeof(*laia));
6354         lad->lad_touch_gen++;
6355
6356         down_read(&ltds->ltd_rw_sem);
6357         cfs_foreach_bit(lad->lad_bitmap, idx) {
6358                 ltd = lfsck_ltd2tgt(ltds, idx);
6359                 if (unlikely(!ltd))
6360                         continue;
6361
6362                 laia->laia_ltd = ltd;
6363                 rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
6364                         lfsck_namespace_assistant_sync_failures_interpret,
6365                         laia, LFSCK_NOTIFY);
6366                 if (rc != 0)
6367                         CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant fail "
6368                                "to sync failure with MDT %x: rc = %d\n",
6369                                lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
6370         }
6371         up_read(&ltds->ltd_rw_sem);
6372
6373         rc = ptlrpc_set_wait(set);
6374         ptlrpc_set_destroy(set);
6375
6376         GOTO(out, rc);
6377
6378 out:
6379         if (rc != 0)
6380                 CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant fail "
6381                        "to sync failure with MDTs, and related MDTs "
6382                        "may handle orphan improperly: rc = %d\n",
6383                        lfsck_lfsck2name(lfsck), rc);
6384
6385         EXIT;
6386 }
6387
6388 struct lfsck_assistant_operations lfsck_namespace_assistant_ops = {
6389         .la_handler_p1          = lfsck_namespace_assistant_handler_p1,
6390         .la_handler_p2          = lfsck_namespace_assistant_handler_p2,
6391         .la_fill_pos            = lfsck_namespace_assistant_fill_pos,
6392         .la_double_scan_result  = lfsck_namespace_double_scan_result,
6393         .la_req_fini            = lfsck_namespace_assistant_req_fini,
6394         .la_sync_failures       = lfsck_namespace_assistant_sync_failures,
6395 };
6396
6397 /**
6398  * Verify the specified linkEA entry for the given directory object.
6399  * If the object has no such linkEA entry or it has more other linkEA
6400  * entries, then re-generate the linkEA with the given information.
6401  *
6402  * \param[in] env       pointer to the thread context
6403  * \param[in] obj       pointer to the dt_object to be handled
6404  * \param[in] cname     the name for the child in the parent directory
6405  * \param[in] pfid      the parent directory's FID for the linkEA
6406  *
6407  * \retval              0 for success
6408  * \retval              negative error number on failure
6409  */
6410 int lfsck_verify_linkea(const struct lu_env *env, struct dt_object *obj,
6411                         const struct lu_name *cname, const struct lu_fid *pfid)
6412 {
6413         struct dt_device        *dev    = lfsck_obj2dev(obj);
6414         struct linkea_data       ldata  = { NULL };
6415         struct lu_buf            linkea_buf;
6416         struct thandle          *th;
6417         int                      rc;
6418         int                      fl     = LU_XATTR_CREATE;
6419         bool                     dirty  = false;
6420         ENTRY;
6421
6422         LASSERT(S_ISDIR(lfsck_object_type(obj)));
6423
6424         rc = lfsck_links_read_with_rec(env, obj, &ldata);
6425         if (rc == -ENODATA) {
6426                 dirty = true;
6427         } else if (rc == 0) {
6428                 fl = LU_XATTR_REPLACE;
6429                 if (ldata.ld_leh->leh_reccount != 1) {
6430                         dirty = true;
6431                 } else {
6432                         rc = linkea_links_find(&ldata, cname, pfid);
6433                         if (rc != 0)
6434                                 dirty = true;
6435                 }
6436         }
6437
6438         if (!dirty)
6439                 RETURN(rc);
6440
6441         rc = linkea_links_new(&ldata, &lfsck_env_info(env)->lti_linkea_buf,
6442                               cname, pfid);
6443         if (rc != 0)
6444                 RETURN(rc);
6445
6446         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
6447                        ldata.ld_leh->leh_len);
6448         th = dt_trans_create(env, dev);
6449         if (IS_ERR(th))
6450                 RETURN(PTR_ERR(th));
6451
6452         rc = dt_declare_xattr_set(env, obj, &linkea_buf,
6453                                   XATTR_NAME_LINK, fl, th);
6454         if (rc != 0)
6455                 GOTO(stop, rc);
6456
6457         rc = dt_trans_start_local(env, dev, th);
6458         if (rc != 0)
6459                 GOTO(stop, rc);
6460
6461         dt_write_lock(env, obj, 0);
6462         rc = dt_xattr_set(env, obj, &linkea_buf,
6463                           XATTR_NAME_LINK, fl, th);
6464         dt_write_unlock(env, obj);
6465
6466         GOTO(stop, rc);
6467
6468 stop:
6469         dt_trans_stop(env, dev, th);
6470         return rc;
6471 }
6472
6473 /**
6474  * Get the name and parent directory's FID from the first linkEA entry.
6475  *
6476  * \param[in] env       pointer to the thread context
6477  * \param[in] obj       pointer to the object which get linkEA from
6478  * \param[out] name     pointer to the buffer to hold the name
6479  *                      in the first linkEA entry
6480  * \param[out] pfid     pointer to the buffer to hold the parent
6481  *                      directory's FID in the first linkEA entry
6482  *
6483  * \retval              0 for success
6484  * \retval              negative error number on failure
6485  */
6486 int lfsck_links_get_first(const struct lu_env *env, struct dt_object *obj,
6487                           char *name, struct lu_fid *pfid)
6488 {
6489         struct lu_name           *cname = &lfsck_env_info(env)->lti_name;
6490         struct linkea_data        ldata = { NULL };
6491         int                       rc;
6492
6493         rc = lfsck_links_read_with_rec(env, obj, &ldata);
6494         if (rc)
6495                 return rc;
6496
6497         linkea_first_entry(&ldata);
6498         linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen, cname, pfid);
6499         /* To guarantee the 'name' is terminated with '0'. */
6500         memcpy(name, cname->ln_name, cname->ln_namelen);
6501         name[cname->ln_namelen] = 0;
6502
6503         return 0;
6504 }
6505
6506 /**
6507  * Update the object's name entry with the given FID.
6508  *
6509  * \param[in] env       pointer to the thread context
6510  * \param[in] lfsck     pointer to the lfsck instance
6511  * \param[in] dir       pointer to the directory that holds
6512  *                      the name entry
6513  * \param[in] name      the name for the entry to be updated
6514  * \param[in] fid       the new FID for the name entry referenced
6515  * \param[in] type      the type for the name entry to be updated
6516  *
6517  * \retval              0 for success
6518  * \retval              negative error number on failure
6519  */
6520 int lfsck_update_name_entry(const struct lu_env *env,
6521                             struct lfsck_instance *lfsck,
6522                             struct dt_object *dir, const char *name,
6523                             const struct lu_fid *fid, __u32 type)
6524 {
6525         struct lfsck_thread_info *info   = lfsck_env_info(env);
6526         struct dt_insert_rec     *rec    = &info->lti_dt_rec;
6527         struct lfsck_lock_handle *llh    = &info->lti_llh;
6528         struct dt_device         *dev    = lfsck_obj2dev(dir);
6529         struct thandle           *th;
6530         int                       rc;
6531         bool                      exists = true;
6532         ENTRY;
6533
6534         rc = lfsck_lock(env, lfsck, dir, name, llh,
6535                         MDS_INODELOCK_UPDATE, LCK_PW);
6536         if (rc != 0)
6537                 RETURN(rc);
6538
6539         th = dt_trans_create(env, dev);
6540         if (IS_ERR(th))
6541                 GOTO(unlock, rc = PTR_ERR(th));
6542
6543         rc = dt_declare_delete(env, dir, (const struct dt_key *)name, th);
6544         if (rc != 0)
6545                 GOTO(stop, rc);
6546
6547         rec->rec_type = type;
6548         rec->rec_fid = fid;
6549         rc = dt_declare_insert(env, dir, (const struct dt_rec *)rec,
6550                                (const struct dt_key *)name, th);
6551         if (rc != 0)
6552                 GOTO(stop, rc);
6553
6554         rc = dt_declare_ref_add(env, dir, th);
6555         if (rc != 0)
6556                 GOTO(stop, rc);
6557
6558         rc = dt_trans_start_local(env, dev, th);
6559         if (rc != 0)
6560                 GOTO(stop, rc);
6561
6562         rc = dt_delete(env, dir, (const struct dt_key *)name, th);
6563         if (rc == -ENOENT) {
6564                 exists = false;
6565                 rc = 0;
6566         }
6567
6568         if (rc != 0)
6569                 GOTO(stop, rc);
6570
6571         rc = dt_insert(env, dir, (const struct dt_rec *)rec,
6572                        (const struct dt_key *)name, th, 1);
6573         if (rc == 0 && S_ISDIR(type) && !exists) {
6574                 dt_write_lock(env, dir, 0);
6575                 rc = dt_ref_add(env, dir, th);
6576                 dt_write_unlock(env, dir);
6577         }
6578
6579         GOTO(stop, rc);
6580
6581 stop:
6582         dt_trans_stop(env, dev, th);
6583
6584 unlock:
6585         lfsck_unlock(llh);
6586         CDEBUG(D_LFSCK, "%s: update name entry "DFID"/%s with the FID "DFID
6587                " and the type %o: rc = %d\n", lfsck_lfsck2name(lfsck),
6588                PFID(lfsck_dto2fid(dir)), name, PFID(fid), type, rc);
6589
6590         return rc;
6591 }
6592
6593 int lfsck_namespace_setup(const struct lu_env *env,
6594                           struct lfsck_instance *lfsck)
6595 {
6596         struct lfsck_component  *com;
6597         struct lfsck_namespace  *ns;
6598         struct dt_object        *root = NULL;
6599         struct dt_object        *obj;
6600         int                      i;
6601         int                      rc;
6602         ENTRY;
6603
6604         LASSERT(lfsck->li_master);
6605
6606         OBD_ALLOC_PTR(com);
6607         if (com == NULL)
6608                 RETURN(-ENOMEM);
6609
6610         INIT_LIST_HEAD(&com->lc_link);
6611         INIT_LIST_HEAD(&com->lc_link_dir);
6612         init_rwsem(&com->lc_sem);
6613         atomic_set(&com->lc_ref, 1);
6614         com->lc_lfsck = lfsck;
6615         com->lc_type = LFSCK_TYPE_NAMESPACE;
6616         com->lc_ops = &lfsck_namespace_ops;
6617         com->lc_data = lfsck_assistant_data_init(
6618                         &lfsck_namespace_assistant_ops,
6619                         LFSCK_NAMESPACE);
6620         if (com->lc_data == NULL)
6621                 GOTO(out, rc = -ENOMEM);
6622
6623         com->lc_file_size = sizeof(struct lfsck_namespace);
6624         OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
6625         if (com->lc_file_ram == NULL)
6626                 GOTO(out, rc = -ENOMEM);
6627
6628         OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
6629         if (com->lc_file_disk == NULL)
6630                 GOTO(out, rc = -ENOMEM);
6631
6632         for (i = 0; i < LFSCK_STF_COUNT; i++)
6633                 mutex_init(&com->lc_sub_trace_objs[i].lsto_mutex);
6634
6635         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
6636         if (IS_ERR(root))
6637                 GOTO(out, rc = PTR_ERR(root));
6638
6639         if (unlikely(!dt_try_as_dir(env, root)))
6640                 GOTO(out, rc = -ENOTDIR);
6641
6642         obj = local_index_find_or_create(env, lfsck->li_los, root,
6643                                          LFSCK_NAMESPACE,
6644                                          S_IFREG | S_IRUGO | S_IWUSR,
6645                                          &dt_lfsck_namespace_features);
6646         if (IS_ERR(obj))
6647                 GOTO(out, rc = PTR_ERR(obj));
6648
6649         com->lc_obj = obj;
6650         rc = lfsck_namespace_load(env, com);
6651         if (rc == -ENODATA)
6652                 rc = lfsck_namespace_init(env, com);
6653         else if (rc < 0)
6654                 rc = lfsck_namespace_reset(env, com, true);
6655         else
6656                 rc = lfsck_load_sub_trace_files(env, com,
6657                         &dt_lfsck_namespace_features, LFSCK_NAMESPACE, false);
6658         if (rc != 0)
6659                 GOTO(out, rc);
6660
6661         ns = com->lc_file_ram;
6662         switch (ns->ln_status) {
6663         case LS_INIT:
6664         case LS_COMPLETED:
6665         case LS_FAILED:
6666         case LS_STOPPED:
6667                 spin_lock(&lfsck->li_lock);
6668                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
6669                 spin_unlock(&lfsck->li_lock);
6670                 break;
6671         default:
6672                 CERROR("%s: unknown lfsck_namespace status %d\n",
6673                        lfsck_lfsck2name(lfsck), ns->ln_status);
6674                 /* fall through */
6675         case LS_SCANNING_PHASE1:
6676         case LS_SCANNING_PHASE2:
6677                 /* No need to store the status to disk right now.
6678                  * If the system crashed before the status stored,
6679                  * it will be loaded back when next time. */
6680                 ns->ln_status = LS_CRASHED;
6681                 /* fall through */
6682         case LS_PAUSED:
6683         case LS_CRASHED:
6684                 spin_lock(&lfsck->li_lock);
6685                 list_add_tail(&com->lc_link, &lfsck->li_list_scan);
6686                 list_add_tail(&com->lc_link_dir, &lfsck->li_list_dir);
6687                 spin_unlock(&lfsck->li_lock);
6688                 break;
6689         }
6690
6691         GOTO(out, rc = 0);
6692
6693 out:
6694         if (root != NULL && !IS_ERR(root))
6695                 lfsck_object_put(env, root);
6696         if (rc != 0) {
6697                 lfsck_component_cleanup(env, com);
6698                 CERROR("%s: fail to init namespace LFSCK component: rc = %d\n",
6699                        lfsck_lfsck2name(lfsck), rc);
6700         }
6701         return rc;
6702 }