lustre/lfsck/lfsck_striped_dir.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) 2014, 2017, Intel Corporation.
  24  */
  25 /*
  26  * lustre/lfsck/lfsck_striped_dir.c
  27  *
  28  * Author: Fan, Yong <fan.yong@intel.com>
  29  */
  30
  31 /*
  32  * About the verification for striped directory. Some rules and assumptions:
  33  *
  34  * 1) lmv_magic: The magic may be wrong. But it is almost impossible (1/2^32
  35  *    probability) that a master LMV EA claims as a slave LMV EA by wrong,
  36  *    so we can ignore such race case and the reverse case.
  37  *
  38  * 2) lmv_master_mdt_index: The master index can be self-verified by compared
  39  *    with the MDT index directly. The slave stripe index can be verified by
  40  *    compared with the file name. Although both the name entry and the LMV EA
  41  *    can be wrong, it is almost impossible that they hit the same bad data
  42  *    So if they match each other, then trust them. Similarly, for the shard,
  43  *    it stores index in both slave LMV EA and in linkEA, if the two copies
  44  *    match, then trust them.
  45  *
  46  * 3) lmv_hash_type: The valid hash type should be LMV_HASH_TYPE_ALL_CHARS or
  47  *    LMV_HASH_TYPE_FNV_1A_64. If the LFSCK instance on some slave finds that
  48  *    the name hash against the hash function does not match the MDT, then it
  49  *    will change the master LMV EA hash type as LMV_HASH_TYPE_UNKNOWN. With
  50  *    such hash type, the whole striped directory still can be accessed via
  51  *    lookup/readdir, and also support unlink, but cannot add new name entry.
  52  *
  53  * 3.1) If the master hash type is one of the valid values, then trust the
  54  *      master LMV EA. Because:
  55  *
  56  * 3.1.1) The master hash type is visible to the client and used by the client.
  57  *
  58  * 3.1.2) For a given name, different hash types may map the name entry to the
  59  *        same MDT. So simply checking one name entry or some name entries may
  60  *        cannot verify whether the hash type is correct or not.
  61  *
  62  * 3.1.3) Different shards can claim different hash types, it is not easy to
  63  *        distinguish which ones are correct. Even though the master is wrong,
  64  *        as the LFSCK processing, some LFSCK instance on other MDT may finds
  65  *        unmatched name hash, then it will change the master hash type to
  66  *        LMV_HASH_TYPE_UNKNOWN as described above. The worst case is euqal
  67  *        to the case without the LFSCK.
  68  *
  69  * 3.2) If the master hash type is invalid, nor LMV_HASH_TYPE_UNKNOWN, then
  70  *      trust the first shard with valid hash type (ALL_CHARS or FNV_1A_64).
  71  *      If the shard is also worng, means there are double failures, then as
  72  *      the LFSCK processing, other LFSCK instances on the other MDTs may
  73  *      find unmatched name hash, and then, the master hash type will be
  74  *      changed to LMV_HASH_TYPE_UNKNOWN as described in the 3).
  75  *
  76  * 3.3) If the master hash type is LMV_HASH_TYPE_UNKNOWN, then it is possible
  77  *      that some other LFSCK instance on other MDT found bad name hash, then
  78  *      changed the master hash type to LMV_HASH_TYPE_UNKNOWN as described in
  79  *      the 3). But it also maybe because of data corruption in master LMV EA.
  80  *      To make such two cases to be distinguishable, when the LFSCK changes
  81  *      the master hash type to LMV_HASH_TYPE_UNKNOWN, it will mark in the
  82  *      master LMV EA (new lmv flags LMV_HASH_FLAG_BAD_TYPE). Then subsequent
  83  *      LFSCK checking can distinguish them: for former case, turst the master
  84  *      LMV EA with nothing to be done; otherwise, trust the first shard with
  85  *      valid hash type (ALL_CHARS or FNV_1A_64) as the 3.2) does.
  86  *
  87  * 4) lmv_stripe_count: For a shard of a striped directory, if its index has
  88  *    been verified as the 2), then the stripe count must be larger than its
  89  *    index. For the master object, by scanning each shard's index, the LFSCK
  90  *    can know the highest index, and the stripe count must be larger than the
  91  *    known highest index. If the stipe count in the LMV EA matches above two
  92  *    rules, then it is may be trustable. If both the master claimed stripe
  93  *    count and the slave claimed stripe count match each own rule, but they
  94  *    are not the same, then trust the master. Because the stripe count in
  95  *    the master LMV EA is visible to client and used to distribute the name
  96  *    entry to some shard, but the slave LMV EA is only used for verification
  97  *    and invisible to client.
  98  *
  99  * 5) If the master LMV EA is lost, then there are two possible cases:
 100  *
 101  * 5.1) The slave claims slave LMV EA by wrong, means that the parent was not
 102  *      a striped directory, but its sub-directory has a wrong slave LMV EA.
 103  *      It is very very race case, similar as the 1), can be ignored.
 104  *
 105  * 5.2) The parent directory is a striped directory, but the master LMV EA
 106  *      is lost or crashed. Then the LFSCK needs to re-generate the master
 107  *      LMV EA: the lmv_master_mdt_index is from the MDT device index; the
 108  *      lmv_hash_type is from the first valid shard; the lmv_stripe_count
 109  *      will be calculated via scanning all the shards.
 110  *
 111  * 5.2.1) Before re-generating the master LMV EA, the LFSCK needs to check
 112  *        whether someone has created some file(s) under the master object
 113  *        after the master LMV EA disappear. If yes, the LFSCK will cannot
 114  *        re-generate the master LMV EA, otherwise, such new created files
 115  *        will be invisible to client. Under such case, the LFSCK will mark
 116  *        the master object as read only (without master LMV EA). Then all
 117  *        things under the master MDT-object, including those new created
 118  *        files and the shards themselves, will be visibile to client. And
 119  *        then the administrator can handle the bad striped directory with
 120  *        more human knowledge.
 121  *
 122  * 5.2.2) If someone created some special sub-directory under the master
 123  *        MDT-object with the same naming rule as shard name $FID:$index,
 124  *        as to the LFSCK cannot detect it before re-generating the master
 125  *        LMV EA, then such sub-directory itself will be invisible after
 126  *        the LFSCK re-generating the master LMV EA. The sub-items under
 127  *        such sub-directory are still visible to client. As the LFSCK
 128  *        processing, if such sub-directory cause some conflict with other
 129  *        normal shard, such as the index conflict, then the LFSCK will
 130  *        remove the master LMV EA and change the master MDT-object to
 131  *        read-only mode as the 5.2.1). But if there is no conflict, the
 132  *        LFSCK will regard such sub-directory as a striped shard that
 133  *        lost its slave LMV EA, and will re-generate slave LMV EA for it.
 134  *
 135  * 5.2.3) Anytime, if the LFSCK found some shards name/index conflict,
 136  *        and cannot make the distinguish which one is right, then it
 137  *        will remove the master LMV EA and change the MDT-object to
 138  *        read-only mode as the 5.2.2).
 139  */
 140
 141 #define DEBUG_SUBSYSTEM S_LFSCK
 142
 143 #include <lu_object.h>
 144 #include <dt_object.h>
 145 #include <md_object.h>
 146 #include <lustre_fid.h>
 147 #include <lustre_lib.h>
 148 #include <lustre_net.h>
 149 #include <lustre_lmv.h>
 150
 151 #include "lfsck_internal.h"
 152
 153 void lfsck_lmv_put(const struct lu_env *env, struct lfsck_lmv *llmv)
 154 {
 155         if (llmv != NULL && atomic_dec_and_test(&llmv->ll_ref)) {
 156                 if (llmv->ll_inline) {
 157                         struct lfsck_lmv_unit   *llu;
 158                         struct lfsck_instance   *lfsck;
 159
 160                         llu = list_entry(llmv, struct lfsck_lmv_unit, llu_lmv);
 161                         lfsck = llu->llu_lfsck;
 162
 163                         spin_lock(&lfsck->li_lock);
 164                         list_del(&llu->llu_link);
 165                         spin_unlock(&lfsck->li_lock);
 166
 167                         lfsck_object_put(env, llu->llu_obj);
 168
 169                         LASSERT(llmv->ll_lslr != NULL);
 170
 171                         OBD_FREE_PTR_ARRAY_LARGE(llmv->ll_lslr,
 172                                                  llmv->ll_stripes_allocated);
 173                         OBD_FREE_PTR(llu);
 174                 } else {
 175                         if (llmv->ll_lslr != NULL)
 176                                 OBD_FREE_PTR_ARRAY_LARGE(
 177                                         llmv->ll_lslr,
 178                                         llmv->ll_stripes_allocated);
 179
 180                         OBD_FREE_PTR(llmv);
 181                 }
 182         }
 183 }
 184
 185 /**
 186  * Mark the specified directory as read-only by set LUSTRE_IMMUTABLE_FL.
 187  *
 188  * The caller has taken the ldlm lock on the @obj already.
 189  *
 190  * \param[in] env       pointer to the thread context
 191  * \param[in] com       pointer to the lfsck component
 192  * \param[in] obj       pointer to the object to be handled
 193  * \param[in] del_lmv   true if need to drop the LMV EA
 194  *
 195  * \retval              positive number if nothing to be done
 196  * \retval              zero for success
 197  * \retval              negative error number on failure
 198  */
 199 static int lfsck_disable_master_lmv(const struct lu_env *env,
 200                                     struct lfsck_component *com,
 201                                     struct dt_object *obj, bool del_lmv)
 202 {
 203         struct lfsck_thread_info *info = lfsck_env_info(env);
 204         struct lu_attr *la = &info->lti_la;
 205         struct lfsck_instance *lfsck = com->lc_lfsck;
 206         struct dt_device *dev = lfsck_obj2dev(obj);
 207         struct thandle *th = NULL;
 208         int rc = 0;
 209
 210         ENTRY;
 211         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
 212                 GOTO(log, rc = 0);
 213
 214         th = lfsck_trans_create(env, dev, lfsck);
 215         if (IS_ERR(th))
 216                 GOTO(log, rc = PTR_ERR(th));
 217
 218         if (del_lmv) {
 219                 rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LMV, th);
 220                 if (rc != 0)
 221                         GOTO(stop, rc);
 222         }
 223
 224         la->la_valid = LA_FLAGS;
 225         rc = dt_declare_attr_set(env, obj, la, th);
 226         if (rc != 0)
 227                 GOTO(stop, rc);
 228
 229         rc = dt_trans_start_local(env, dev, th);
 230         if (rc != 0)
 231                 GOTO(stop, rc);
 232
 233         dt_write_lock(env, obj, 0);
 234         if (unlikely(lfsck_is_dead_obj(obj)))
 235                 GOTO(unlock, rc = 1);
 236
 237         if (del_lmv) {
 238                 rc = dt_xattr_del(env, obj, XATTR_NAME_LMV, th);
 239                 if (rc != 0)
 240                         GOTO(unlock, rc);
 241         }
 242
 243         rc = dt_attr_get(env, obj, la);
 244         if (rc == 0 && !(la->la_flags & LUSTRE_IMMUTABLE_FL)) {
 245                 la->la_valid = LA_FLAGS;
 246                 la->la_flags |= LUSTRE_IMMUTABLE_FL;
 247                 rc = dt_attr_set(env, obj, la, th);
 248         }
 249
 250         GOTO(unlock, rc);
 251
 252 unlock:
 253         dt_write_unlock(env, obj);
 254
 255 stop:
 256         dt_trans_stop(env, dev, th);
 257
 258 log:
 259         CDEBUG(D_LFSCK,
 260                "%s: namespace LFSCK set the master MDT-object of the striped directory "DFID" as read-only: rc = %d\n",
 261                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
 262
 263         if (rc <= 0) {
 264                 struct lfsck_namespace *ns = com->lc_file_ram;
 265
 266                 ns->ln_flags |= LF_INCONSISTENT;
 267                 if (rc == 0)
 268                         ns->ln_striped_dirs_disabled++;
 269         }
 270
 271         return rc;
 272 }
 273
 274 static inline bool lfsck_is_valid_slave_lmv(struct lmv_mds_md_v1 *lmv)
 275 {
 276         return lmv->lmv_stripe_count >= 1 &&
 277                lmv->lmv_stripe_count <= LFSCK_LMV_MAX_STRIPES &&
 278                lmv->lmv_stripe_count > lmv->lmv_master_mdt_index &&
 279                lmv_is_known_hash_type(lmv->lmv_hash_type);
 280 }
 281
 282 /**
 283  * Remove the striped directory's master LMV EA and mark it as read-only.
 284  *
 285  * Take ldlm lock on the striped directory before calling the
 286  * lfsck_disable_master_lmv().
 287  *
 288  * \param[in] env       pointer to the thread context
 289  * \param[in] com       pointer to the lfsck component
 290  * \param[in] obj       pointer to the striped directory to be handled
 291  * \param[in] lnr       pointer to the namespace request that contains the
 292  *                      striped directory to be handled and other information
 293  *
 294  * \retval              positive number if nothing to be done
 295  * \retval              zero for success
 296  * \retval              negative error number on failure
 297  */
 298 static int lfsck_remove_lmv(const struct lu_env *env,
 299                             struct lfsck_component *com,
 300                             struct dt_object *obj,
 301                             struct lfsck_namespace_req *lnr)
 302 {
 303         struct lustre_handle lh = { 0 };
 304         int rc;
 305
 306         lnr->lnr_lmv->ll_ignore = 1;
 307         rc = lfsck_ibits_lock(env, com->lc_lfsck, obj, &lh,
 308                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
 309                               LCK_EX);
 310         if (rc == 0) {
 311                 rc = lfsck_disable_master_lmv(env, com, obj, true);
 312                 lfsck_ibits_unlock(&lh, LCK_EX);
 313         }
 314
 315         return rc;
 316 }
 317
 318 /**
 319  * Remove the name entry from the striped directory's master MDT-object.
 320  *
 321  * \param[in] env       pointer to the thread context
 322  * \param[in] com       pointer to the lfsck component
 323  * \param[in] dir       pointer to the striped directory
 324  * \param[in] fid       the shard's FID which name entry will be removed
 325  * \param[in] index     the shard's index which name entry will be removed
 326  *
 327  * \retval              positive number for repaired successfully
 328  * \retval              0 if nothing to be repaired
 329  * \retval              negative error number on failure
 330  */
 331 static int lfsck_remove_dirent(const struct lu_env *env,
 332                                struct lfsck_component *com,
 333                                struct dt_object *dir,
 334                                const struct lu_fid *fid, __u32 index)
 335 {
 336         struct lfsck_thread_info *info = lfsck_env_info(env);
 337         struct dt_object *obj;
 338         int rc;
 339
 340         snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2), DFID":%u",
 341                  PFID(fid), index);
 342         obj = lfsck_object_find_bottom(env, com->lc_lfsck, fid);
 343         if (IS_ERR(obj))
 344                 return PTR_ERR(obj);
 345
 346         rc = lfsck_namespace_repair_dirent(env, com, dir, obj,
 347                                         info->lti_tmpbuf2, info->lti_tmpbuf2,
 348                                         S_IFDIR, false, false);
 349         lfsck_object_put(env, obj);
 350         if (rc > 0) {
 351                 struct lfsck_namespace *ns = com->lc_file_ram;
 352
 353                 ns->ln_dirent_repaired++;
 354         }
 355
 356         return rc;
 357 }
 358
 359 /**
 360  * Remove old shard's name entry and refill the @lslr slot with new shard.
 361  *
 362  * Some old shard held the specified @lslr slot, but it is an invalid shard.
 363  * This function will remove the bad shard's name entry, and refill the @lslr
 364  * slot with the new shard.
 365  *
 366  * \param[in] env       pointer to the thread context
 367  * \param[in] com       pointer to the lfsck component
 368  * \param[in] dir       pointer to the striped directory to be handled
 369  * \param[in] lslr      pointer to lfsck_disable_master_lmv slot which content
 370  *                      will be replaced by the given information
 371  * \param[in] lnr       contain the shard's FID to be used to fill the
 372  *                      @lslr slot, it also records the known max filled index
 373  *                      and the known max stripe count
 374  * \param[in] lmv       contain the slave LMV EA to be used to fill the
 375  *                      @lslr slot
 376  * \param[in] index     the old shard's index in the striped directory
 377  * \param[in] flags     the new shard's flags in the @lslr slot
 378  *
 379  * \retval              zero for success
 380  * \retval              negative error number on failure
 381  */
 382 static int lfsck_replace_lmv(const struct lu_env *env,
 383                              struct lfsck_component *com,
 384                              struct dt_object *dir,
 385                              struct lfsck_slave_lmv_rec *lslr,
 386                              struct lfsck_namespace_req *lnr,
 387                              struct lmv_mds_md_v1 *lmv,
 388                              __u32 index, __u32 flags)
 389 {
 390         struct lfsck_lmv *llmv = lnr->lnr_lmv;
 391         int rc;
 392
 393         rc = lfsck_remove_dirent(env, com, dir,
 394                                  &lslr->lslr_fid, index);
 395         if (rc < 0)
 396                 return rc;
 397
 398         lslr->lslr_fid = lnr->lnr_fid;
 399         lslr->lslr_flags = flags;
 400         lslr->lslr_stripe_count = lmv->lmv_stripe_count;
 401         lslr->lslr_index = lmv->lmv_master_mdt_index;
 402         lslr->lslr_hash_type = lmv->lmv_hash_type;
 403         if (flags == LSLF_NONE) {
 404                 if (llmv->ll_hash_type == LMV_HASH_TYPE_UNKNOWN &&
 405                     lmv_is_known_hash_type(lmv->lmv_hash_type))
 406                         llmv->ll_hash_type = lmv->lmv_hash_type;
 407
 408                 if (lslr->lslr_stripe_count <= LFSCK_LMV_MAX_STRIPES &&
 409                     llmv->ll_max_stripe_count < lslr->lslr_stripe_count)
 410                         llmv->ll_max_stripe_count = lslr->lslr_stripe_count;
 411         }
 412
 413         return 0;
 414 }
 415
 416 /**
 417  * Record the slave LMV EA in the lfsck_lmv::ll_lslr.
 418  *
 419  * If the lfsck_lmv::ll_lslr slot corresponding to the given @shard_idx is free,
 420  * then fill the slot with the given @lnr/@lmv/@flags directly (maybe need to
 421  * extend the lfsck_lmv::ll_lslr buffer).
 422  *
 423  * If the lfsck_lmv::ll_lslr slot corresponding to the given @shard_idx is taken
 424  * by other shard, then the LFSCK will try to resolve the conflict by checking
 425  * the two conflict shards' flags, and try other possible slot (if one of them
 426  * claims another possible @shard_idx).
 427  *
 428  * 1) If one of the two conflict shards can be recorded in another slot, then
 429  *    it is OK, go ahead. Otherwise,
 430  *
 431  * 2) If one of them is dangling name entry, then remove (one of) the dangling
 432  *    name entry (and replace related @lslr slot if needed). Otherwise,
 433  *
 434  * 3) If one of them has no slave LMV EA, then check whether the master LMV
 435  *    EA has ever been lost and re-generated (LMV_HASH_FLAG_LOST_LMV in the
 436  *    master LMV EA).
 437  *
 438  * 3.1) If yes, then it is possible that such object is not a real shard of
 439  *      the striped directory, instead, it was created by someone after the
 440  *      master LMV EA lost with the name that matches the shard naming rule.
 441  *      Then the LFSCK will remove the master LMV EA and mark the striped
 442  *      directory as read-only to allow those non-shard files to be visible
 443  *      to client.
 444  *
 445  * 3.2) If no, then remove (one of) the object what has no slave LMV EA.
 446  *
 447  * 4) If all above efforts cannot work, then the LFSCK cannot know how to
 448  *    recover the striped directory. To make the administrator can see the
 449  *    conflicts, the LFSCK will remove the master LMV EA and mark the striped
 450  *    directory as read-only.
 451  *
 452  * This function may be called recursively, to prevent overflow, we define
 453  * LFSCK_REC_LMV_MAX_DEPTH to restrict the recursive call depth.
 454  *
 455  * \param[in] env       pointer to the thread context
 456  * \param[in] com       pointer to the lfsck component
 457  * \param[in] dir       pointer to the striped directory to be handled
 458  * \param[in] lnr       contain the shard's FID to fill the @lslr slot,
 459  *                      it also records the known max filled index and
 460  *                      the known max stripe count
 461  * \param[in] lmv       pointer to the slave LMV EA to be recorded
 462  * \param[in] shard_idx the shard's index used for locating the @lslr slot,
 463  *                      it can be the index stored in the shard's name,
 464  *                      it also can be the index stored in the slave LMV EA
 465  *                      (for recursive case)
 466  * \param[in] flags     the shard's flags to be recorded in the @lslr slot
 467  *                      to indicate the shard status, such as whether has
 468  *                      slave LMV EA, whether dangling name entry, whether
 469  *                      the name entry and slave LMV EA unmatched, and ect
 470  * \param[in] flags2    when be called recursively, the @flags2 tells the
 471  *                      former conflict shard's flags in the @lslr slot.
 472  * \param[in,out] depth To prevent to be called recurisively too deep,
 473  *                      we define the max depth can be called recursively
 474  *                      (LFSCK_REC_LMV_MAX_DEPTH)
 475  *
 476  * \retval              zero for success
 477  * \retval              "-ERANGE" for invalid @shard_idx
 478  * \retval              "-EEXIST" for the required lslr slot has been
 479  *                      occupied by other shard
 480  * \retval              other negative error number on failure
 481  */
 482 static int lfsck_record_lmv(const struct lu_env *env,
 483                             struct lfsck_component *com,
 484                             struct dt_object *dir,
 485                             struct lfsck_namespace_req *lnr,
 486                             struct lmv_mds_md_v1 *lmv, __u32 shard_idx,
 487                             __u32 flags, __u32 flags2, __u32 *depth)
 488 {
 489         struct lfsck_instance *lfsck = com->lc_lfsck;
 490         struct lfsck_lmv *llmv = lnr->lnr_lmv;
 491         const struct lu_fid *fid = &lnr->lnr_fid;
 492         struct lfsck_slave_lmv_rec *lslr;
 493         struct lfsck_rec_lmv_save  *lrls;
 494         int index = shard_idx;
 495         int rc = 0;
 496
 497         ENTRY;
 498         CDEBUG(D_INFO,
 499                "%s: record slave LMV EA for the striped directory "DFID": shard = "DFID", index = %u, flags = %u, flags2 = %u, depth = %d\n",
 500                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(dir)), PFID(fid),
 501                index, flags, flags2, *depth);
 502
 503         if (index < 0 || index >= LFSCK_LMV_MAX_STRIPES)
 504                 RETURN(-ERANGE);
 505
 506         if (index >= llmv->ll_stripes_allocated) {
 507                 struct lfsck_slave_lmv_rec *new_lslr;
 508                 int new_stripes = index + 1;
 509                 size_t old_size = sizeof(*lslr) * llmv->ll_stripes_allocated;
 510
 511                 OBD_ALLOC_PTR_ARRAY_LARGE(new_lslr, new_stripes);
 512                 if (new_lslr == NULL) {
 513                         llmv->ll_failed = 1;
 514
 515                         RETURN(-ENOMEM);
 516                 }
 517
 518                 memcpy(new_lslr, llmv->ll_lslr, old_size);
 519                 OBD_FREE_LARGE(llmv->ll_lslr, old_size);
 520                 llmv->ll_stripes_allocated = new_stripes;
 521                 llmv->ll_lslr = new_lslr;
 522         }
 523
 524         lslr = llmv->ll_lslr + index;
 525         if (unlikely(lu_fid_eq(&lslr->lslr_fid, fid)))
 526                 RETURN(0);
 527
 528         if (fid_is_zero(&lslr->lslr_fid)) {
 529                 lslr->lslr_fid = *fid;
 530                 lslr->lslr_stripe_count = lmv->lmv_stripe_count;
 531                 lslr->lslr_index = lmv->lmv_master_mdt_index;
 532                 lslr->lslr_hash_type = lmv->lmv_hash_type;
 533                 lslr->lslr_flags = flags;
 534                 llmv->ll_stripes_filled++;
 535                 if (flags == LSLF_NONE) {
 536                         if (llmv->ll_hash_type == LMV_HASH_TYPE_UNKNOWN &&
 537                             lmv_is_known_hash_type(lmv->lmv_hash_type))
 538                                 llmv->ll_hash_type = lmv->lmv_hash_type;
 539
 540                         if (lslr->lslr_stripe_count <= LFSCK_LMV_MAX_STRIPES &&
 541                             llmv->ll_max_stripe_count < lslr->lslr_stripe_count)
 542                                 llmv->ll_max_stripe_count =
 543                                                         lslr->lslr_stripe_count;
 544                 }
 545
 546                 if (llmv->ll_max_filled_off < index)
 547                         llmv->ll_max_filled_off = index;
 548
 549                 RETURN(0);
 550         }
 551
 552         (*depth)++;
 553         if (flags != LSLF_BAD_INDEX2)
 554                 LASSERTF(*depth == 1, "depth = %d\n", *depth);
 555
 556         /* Handle conflict cases. */
 557         switch (lslr->lslr_flags) {
 558         case LSLF_NONE:
 559         case LSLF_BAD_INDEX2:
 560                 /* The existing one is a normal valid object. */
 561                 switch (flags) {
 562                 case LSLF_NONE:
 563                         /* The two 'valid' name entries claims the same index,
 564                          * the LFSCK cannot distinguish which one is correct.
 565                          * Then remove the master LMV EA to make all shards to
 566                          * be visible to client, and mark the master MDT-object
 567                          * as read-only. The administrator can handle the
 568                          * conflict with more human knowledge.
 569                          */
 570                         rc = lfsck_remove_lmv(env, com, dir, lnr);
 571                         break;
 572                 case LSLF_BAD_INDEX2:
 573                         GOTO(out, rc = -EEXIST);
 574                 case LSLF_NO_LMVEA:
 575
 576 no_lmvea:
 577                         if (llmv->ll_lmv.lmv_hash_type &
 578                             LMV_HASH_FLAG_LOST_LMV) {
 579                                 /* If the master LMV EA was re-generated by the
 580                                  * former LFSCK reparation, and before such
 581                                  * reparation, someone has created the conflict
 582                                  * object, but the LFSCK did not detect such
 583                                  * conflict, then we have to remove the master
 584                                  * LMV EA and mark the master MDT-object as
 585                                  * read-only. The administrator can handle the
 586                                  * conflict with more human knowledge.
 587                                  */
 588                                 rc = lfsck_remove_lmv(env, com, dir, lnr);
 589                         } else {
 590                                 /* Otherwise, remove the current name entry,
 591                                  * and add its FID in the LFSCK tracing file
 592                                  * for further processing.
 593                                  */
 594                                 rc = lfsck_namespace_trace_update(env, com, fid,
 595                                                 LNTF_CHECK_PARENT, true);
 596                                 if (rc == 0)
 597                                         rc = lfsck_remove_dirent(env, com, dir,
 598                                                                  fid, index);
 599                         }
 600
 601                         break;
 602                 case LSLF_DANGLING:
 603                         /* Remove the current dangling name entry. */
 604                         rc = lfsck_remove_dirent(env, com, dir, fid, index);
 605                         break;
 606                 case LSLF_BAD_INDEX1:
 607                         index = lmv->lmv_master_mdt_index;
 608                         lmv->lmv_master_mdt_index = shard_idx;
 609                         /* The name entry claims an index that is conflict
 610                          * with a valid existing name entry, then try the
 611                          * index in the lmv recursively.
 612                          */
 613                         rc = lfsck_record_lmv(env, com, dir, lnr, lmv, index,
 614                                 LSLF_BAD_INDEX2, lslr->lslr_flags, depth);
 615                         lmv->lmv_master_mdt_index = index;
 616                         if (rc == -ERANGE || rc == -EEXIST)
 617                                 /* The index in the lmv is invalid or conflict
 618                                  * with other. Then we do not know how to solve
 619                                  * the conflict. We will handle it as handle the
 620                                  * case of 'LSLF_NONE' vs 'LSLF_NONE'.
 621                                  */
 622                                 rc = lfsck_remove_lmv(env, com, dir, lnr);
 623
 624                         break;
 625                 default:
 626                         break;
 627                 }
 628
 629                 break;
 630         case LSLF_NO_LMVEA:
 631                 /* The existing one has no slave LMV EA. */
 632                 switch (flags) {
 633                 case LSLF_NONE:
 634
 635 none:
 636                         if (llmv->ll_lmv.lmv_hash_type &
 637                             LMV_HASH_FLAG_LOST_LMV) {
 638                                 /* If the master LMV EA was re-generated by the
 639                                  * former LFSCK reparation, and before such
 640                                  * reparation, someone has created the conflict
 641                                  * object, but the LFSCK did not detect such
 642                                  * conflict, then we have to remove the master
 643                                  * LMV EA and mark the master MDT-object as
 644                                  * read-only. The administrator can handle the
 645                                  * conflict with more human knowledge.
 646                                  */
 647                                 rc = lfsck_remove_lmv(env, com, dir, lnr);
 648                         } else {
 649                                 lrls = &lfsck->li_rec_lmv_save[*depth - 1];
 650                                 lrls->lrls_fid = lslr->lslr_fid;
 651                                 /* Otherwise, remove the existing name entry,
 652                                  * and add its FID in the LFSCK tracing file for
 653                                  * further processing. Refill the slot with
 654                                  * current slave LMV EA.
 655                                  */
 656                                 rc = lfsck_namespace_trace_update(env,
 657                                                 com, &lrls->lrls_fid,
 658                                                 LNTF_CHECK_PARENT, true);
 659                                 if (rc == 0)
 660                                         rc = lfsck_replace_lmv(env, com, dir,
 661                                                 lslr, lnr, lmv, index, flags);
 662                         }
 663
 664                         break;
 665                 case LSLF_BAD_INDEX2:
 666                         if (flags2 >= lslr->lslr_flags)
 667                                 GOTO(out, rc = -EEXIST);
 668
 669                         goto none;
 670                 case LSLF_NO_LMVEA:
 671                         goto no_lmvea;
 672                 case LSLF_DANGLING:
 673                         /* Remove the current dangling name entry. */
 674                         rc = lfsck_remove_dirent(env, com, dir, fid, index);
 675                         break;
 676                 case LSLF_BAD_INDEX1:
 677                         index = lmv->lmv_master_mdt_index;
 678                         lmv->lmv_master_mdt_index = shard_idx;
 679                         /* The name entry claims an index that is conflict
 680                          * with a valid existing name entry, then try the
 681                          * index in the lmv recursively.
 682                          */
 683                         rc = lfsck_record_lmv(env, com, dir, lnr, lmv, index,
 684                                 LSLF_BAD_INDEX2, lslr->lslr_flags, depth);
 685                         lmv->lmv_master_mdt_index = index;
 686                         if (rc == -ERANGE || rc == -EEXIST) {
 687                                 index = shard_idx;
 688                                 goto no_lmvea;
 689                         }
 690
 691                         break;
 692                 default:
 693                         break;
 694                 }
 695
 696                 break;
 697         case LSLF_DANGLING:
 698                 /* The existing one is a dangling name entry. */
 699                 switch (flags) {
 700                 case LSLF_NONE:
 701                 case LSLF_BAD_INDEX2:
 702                 case LSLF_NO_LMVEA:
 703                         /* Remove the existing dangling name entry.
 704                          * Refill the lslr slot with the given LMV.
 705                          */
 706                         rc = lfsck_replace_lmv(env, com, dir, lslr, lnr,
 707                                                lmv, index, flags);
 708                         break;
 709                 case LSLF_DANGLING:
 710                         /* Two dangling name entries conflict,
 711                          * remove the current one.
 712                          */
 713                         rc = lfsck_remove_dirent(env, com, dir, fid, index);
 714                         break;
 715                 case LSLF_BAD_INDEX1:
 716                         index = lmv->lmv_master_mdt_index;
 717                         lmv->lmv_master_mdt_index = shard_idx;
 718                         /* The name entry claims an index that is conflict
 719                          * with a valid existing name entry, then try the
 720                          * index in the lmv recursively.
 721                          */
 722                         rc = lfsck_record_lmv(env, com, dir, lnr, lmv, index,
 723                                 LSLF_BAD_INDEX2, lslr->lslr_flags, depth);
 724                         lmv->lmv_master_mdt_index = index;
 725                         if (rc == -ERANGE || rc == -EEXIST)
 726                                 /* If the index in the lmv is invalid or
 727                                  * also conflict with other, then remove
 728                                  * the existing dangling name entry.
 729                                  * Refill the lslr slot with the given LMV.
 730                                  */
 731                                 rc = lfsck_replace_lmv(env, com, dir, lslr, lnr,
 732                                                        lmv, shard_idx, flags);
 733
 734                         break;
 735                 default:
 736                         break;
 737                 }
 738
 739                 break;
 740         case LSLF_BAD_INDEX1: {
 741                 if (*depth >= LFSCK_REC_LMV_MAX_DEPTH)
 742                         goto conflict;
 743
 744                 lrls = &lfsck->li_rec_lmv_save[*depth - 1];
 745                 lrls->lrls_fid = lnr->lnr_fid;
 746                 lrls->lrls_lmv = *lmv;
 747
 748                 lnr->lnr_fid = lslr->lslr_fid;
 749                 lmv->lmv_master_mdt_index = index;
 750                 lmv->lmv_stripe_count = lslr->lslr_stripe_count;
 751                 lmv->lmv_hash_type = lslr->lslr_hash_type;
 752                 index = lslr->lslr_index;
 753
 754                 /* Existing one has another possible slot, try recursively. */
 755                 rc = lfsck_record_lmv(env, com, dir, lnr, lmv, index,
 756                                       LSLF_BAD_INDEX2, flags, depth);
 757                 *lmv = lrls->lrls_lmv;
 758                 lnr->lnr_fid = lrls->lrls_fid;
 759                 index = shard_idx;
 760                 if (rc != 0) {
 761                         if (rc == -ERANGE || rc == -EEXIST)
 762                                 goto conflict;
 763
 764                         break;
 765                 }
 766
 767                 lslr->lslr_fid = *fid;
 768                 lslr->lslr_flags = flags;
 769                 lslr->lslr_stripe_count = lmv->lmv_stripe_count;
 770                 lslr->lslr_index = lmv->lmv_master_mdt_index;
 771                 lslr->lslr_hash_type = lmv->lmv_hash_type;
 772                 if (flags == LSLF_NONE) {
 773                         if (llmv->ll_hash_type == LMV_HASH_TYPE_UNKNOWN &&
 774                             lmv_is_known_hash_type(lmv->lmv_hash_type))
 775                                 llmv->ll_hash_type = lmv->lmv_hash_type;
 776
 777                         if (lslr->lslr_stripe_count <= LFSCK_LMV_MAX_STRIPES &&
 778                             llmv->ll_max_stripe_count < lslr->lslr_stripe_count)
 779                                 llmv->ll_max_stripe_count =
 780                                                         lslr->lslr_stripe_count;
 781                 }
 782
 783                 break;
 784
 785 conflict:
 786                 switch (flags) {
 787                 case LSLF_NONE:
 788                         /* The two 'valid' name entries claims the same
 789                          * index, the LFSCK cannot distinguish which one
 790                          * is correct. Then remove the master LMV EA to
 791                          * make all shards to be visible to client, and
 792                          * mark the master MDT-object as read-only. The
 793                          * administrator can handle the conflict with
 794                          * more human knowledge.
 795                          */
 796                         rc = lfsck_remove_lmv(env, com, dir, lnr);
 797                         break;
 798                 case LSLF_BAD_INDEX2:
 799                         GOTO(out, rc = -EEXIST);
 800                 case LSLF_NO_LMVEA:
 801                         goto no_lmvea;
 802                 case LSLF_DANGLING:
 803                         /* Remove the current dangling name entry. */
 804                         rc = lfsck_remove_dirent(env, com, dir, fid, index);
 805                         break;
 806                 case LSLF_BAD_INDEX1:
 807                         index = lmv->lmv_master_mdt_index;
 808                         lmv->lmv_master_mdt_index = shard_idx;
 809                         /* The name entry claims an index that is conflict
 810                          * with a valid existing name entry, then try the
 811                          * index in the lmv recursively.
 812                          */
 813                         rc = lfsck_record_lmv(env, com, dir, lnr, lmv, index,
 814                                 LSLF_BAD_INDEX2, lslr->lslr_flags, depth);
 815                         lmv->lmv_master_mdt_index = index;
 816                         if (rc == -ERANGE || rc == -EEXIST)
 817                                 /* The index in the lmv is invalid or also
 818                                  * conflict with other. Then we do not know how
 819                                  * to resolve the conflict. We will handle it as
 820                                  * handle the case of 'LSLF_NONE' vs 'LSLF_NONE'
 821                                  */
 822                                 rc = lfsck_remove_lmv(env, com, dir, lnr);
 823
 824                         break;
 825                 }
 826
 827                 break;
 828         }
 829         default:
 830                 break;
 831         }
 832
 833         if (rc < 0)
 834                 llmv->ll_failed = 1;
 835
 836         GOTO(out, rc);
 837
 838 out:
 839         (*depth)--;
 840
 841         return rc > 0 ? 0 : rc;
 842 }
 843
 844 /**
 845  * Read LMV from bottom object, so it doesn't contain stripe FIDs.
 846  *
 847  * TODO: test migrating/foreign directory lfsck
 848  *
 849  * \param[in] env       thread env
 850  * \param[in] lfsck     lfsck instance
 851  * \param[in] obj       dt object
 852  * \param[out] lmv      LMV data pointer
 853  *
 854  * \retval              0 on success
 855  * \retval              -ENODATA on no LMV, corrupt LMV, dir is dead or foreign
 856  *                      -ev on other failures
 857  */
 858 int lfsck_read_stripe_lmv(const struct lu_env *env,
 859                           struct lfsck_instance *lfsck,
 860                           struct dt_object *obj,
 861                           struct lmv_mds_md_v1 *lmv)
 862 {
 863         struct lfsck_thread_info *info = lfsck_env_info(env);
 864         struct lu_buf *buf = &info->lti_buf;
 865         struct lmv_foreign_md *lfm;
 866         int rc;
 867
 868         /* use bottom object to avoid reading in shard FIDs */
 869         obj = lfsck_object_find_bottom(env, lfsck, lu_object_fid(&obj->do_lu));
 870         if (IS_ERR(obj))
 871                 return PTR_ERR(obj);
 872
 873         dt_read_lock(env, obj, 0);
 874         buf->lb_buf = lmv;
 875         buf->lb_len = sizeof(*lmv);
 876         rc = dt_xattr_get(env, obj, buf, XATTR_NAME_LMV);
 877         if (unlikely(rc == -ERANGE)) {
 878                 buf = &info->lti_big_buf;
 879                 /* this may be a foreign LMV */
 880                 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LMV);
 881                 if (rc > sizeof(*lmv)) {
 882                         int rc1;
 883
 884                         lu_buf_check_and_alloc(buf, rc);
 885                         rc1 = dt_xattr_get(env, obj, buf, XATTR_NAME_LMV);
 886                         if (rc != rc1)
 887                                 rc = -ENODATA;
 888                 } else {
 889                         rc = -ENODATA;
 890                 }
 891         }
 892         dt_read_unlock(env, obj);
 893
 894         lfsck_object_put(env, obj);
 895
 896         if (rc > offsetof(typeof(*lfm), lfm_value) &&
 897             *((__u32 *)buf->lb_buf) == LMV_MAGIC_FOREIGN) {
 898                 __u32 value_len;
 899
 900                 lfm = buf->lb_buf;
 901                 value_len = le32_to_cpu(lfm->lfm_length);
 902                 CDEBUG(D_INFO,
 903                        "foreign LMV EA, magic %x, len %u, type %x, flags %x, for dir "DFID"\n",
 904                        le32_to_cpu(lfm->lfm_magic), value_len,
 905                        le32_to_cpu(lfm->lfm_type), le32_to_cpu(lfm->lfm_flags),
 906                        PFID(lfsck_dto2fid(obj)));
 907
 908                 if (rc != value_len + offsetof(typeof(*lfm), lfm_value))
 909                         CDEBUG(D_LFSCK,
 910                                "foreign LMV EA internal size %u does not match EA full size %d for dir "DFID"\n",
 911                                value_len, rc, PFID(lfsck_dto2fid(obj)));
 912
 913                 /* no further usage/decode of foreign LMV outside */
 914                 return -ENODATA;
 915         }
 916
 917         if (rc == sizeof(*lmv)) {
 918                 rc = 0;
 919                 lfsck_lmv_header_le_to_cpu(lmv, lmv);
 920                 /* if LMV is corrupt, return -ENODATA */
 921                 if (lmv->lmv_magic != LMV_MAGIC_V1 &&
 922                     lmv->lmv_magic != LMV_MAGIC_STRIPE)
 923                         rc = -ENODATA;
 924         } else if (rc >= 0) {
 925                 /* LMV is corrupt */
 926                 rc = -ENODATA;
 927         }
 928
 929         return rc;
 930 }
 931
 932 /**
 933  * Parse the shard's index from the given shard name.
 934  *
 935  * The valid shard name/type should be:
 936  * 1) The type must be S_IFDIR
 937  * 2) The name should be $FID:$index
 938  * 3) the index should within valid range.
 939  *
 940  * \param[in] env       pointer to the thread context
 941  * \param[in] name      the shard name
 942  * \param[in] namelen   the name length
 943  * \param[in] type      the entry's type
 944  * \param[in] fid       the entry's FID
 945  *
 946  * \retval              zero or positive number for the index from the name
 947  * \retval              negative error number on failure
 948  */
 949 int lfsck_shard_name_to_index(const struct lu_env *env, const char *name,
 950                               int namelen, __u16 type, const struct lu_fid *fid)
 951 {
 952         char *name2 = lfsck_env_info(env)->lti_tmpbuf2;
 953         int len;
 954         int idx = 0;
 955
 956         if (!S_ISDIR(type))
 957                 return -ENOTDIR;
 958
 959         LASSERT(name != name2);
 960
 961         len = snprintf(name2, sizeof(lfsck_env_info(env)->lti_tmpbuf2),
 962                        DFID":", PFID(fid));
 963         if (namelen < len + 1 || memcmp(name, name2, len) != 0)
 964                 return -EINVAL;
 965
 966         do {
 967                 if (!isdigit(name[len]))
 968                         return -EINVAL;
 969
 970                 idx = idx * 10 + name[len++] - '0';
 971         } while (len < namelen);
 972
 973         if (idx >= LFSCK_LMV_MAX_STRIPES)
 974                 return -EINVAL;
 975
 976         return idx;
 977 }
 978
 979 static inline bool lfsck_name_hash_match(struct lmv_mds_md_v1 *lmv,
 980                                          const char *name, int namelen)
 981 {
 982         int idx;
 983
 984         idx = lmv_name_to_stripe_index_old(lmv, name, namelen);
 985         if (idx == lmv->lmv_master_mdt_index)
 986                 return true;
 987
 988         if (!lmv_hash_is_layout_changing(lmv->lmv_hash_type))
 989                 return false;
 990
 991         idx = lmv_name_to_stripe_index(lmv, name, namelen);
 992         return (idx == lmv->lmv_master_mdt_index);
 993 }
 994
 995 bool lfsck_is_valid_slave_name_entry(const struct lu_env *env,
 996                                      struct lfsck_lmv *llmv,
 997                                      const char *name, int namelen)
 998 {
 999         if (llmv == NULL || !llmv->ll_lmv_slave || !llmv->ll_lmv_verified)
1000                 return true;
1001
1002         return lfsck_name_hash_match(&llmv->ll_lmv, name, namelen);
1003 }
1004
1005 /**
1006  * Check whether the given name is a valid entry under the @parent.
1007  *
1008  * If the @parent is a striped directory then the @child should one
1009  * shard of the striped directory, its name should be $FID:$index.
1010  *
1011  * If the @parent is a shard of a striped directory, then the name hash
1012  * should match the MDT, otherwise it is invalid.
1013  *
1014  * \param[in] env       pointer to the thread context
1015  * \param[in] parent    the parent directory
1016  * \param[in] child     the child object to be checked
1017  * \param[in] cname     the name for the @child in the parent directory
1018  *
1019  * \retval              positive number for invalid name entry
1020  * \retval              0 if the name is valid or uncertain
1021  * \retval              negative error number on failure
1022  */
1023 int lfsck_namespace_check_name(const struct lu_env *env,
1024                                struct lfsck_instance *lfsck,
1025                                struct dt_object *parent,
1026                                struct dt_object *child,
1027                                const struct lu_name *cname)
1028 {
1029         struct lmv_mds_md_v1 *lmv = &lfsck_env_info(env)->lti_lmv;
1030         int rc;
1031
1032         rc = lfsck_read_stripe_lmv(env, lfsck, parent, lmv);
1033         if (rc != 0)
1034                 RETURN(rc == -ENODATA ? 0 : rc);
1035
1036         if (lmv->lmv_magic == LMV_MAGIC_STRIPE) {
1037                 if (!lfsck_is_valid_slave_lmv(lmv))
1038                         return 0;
1039
1040                 if (!lfsck_name_hash_match(lmv, cname->ln_name,
1041                                            cname->ln_namelen))
1042                         return 1;
1043         } else if (lfsck_shard_name_to_index(env, cname->ln_name,
1044                         cname->ln_namelen, lfsck_object_type(child),
1045                         lfsck_dto2fid(child)) < 0) {
1046                 return 1;
1047         }
1048
1049         return 0;
1050 }
1051
1052 /**
1053  * Update the object's LMV EA with the given @lmv.
1054  *
1055  * \param[in] env       pointer to the thread context
1056  * \param[in] com       pointer to the lfsck component
1057  * \param[in] obj       pointer to the object which LMV EA will be updated
1058  * \param[in] lmv       pointer to buffer holding the new LMV EA
1059  * \param[in] locked    whether the caller has held ldlm lock on the @obj or not
1060  *
1061  * \retval              positive number for nothing to be done
1062  * \retval              zero if updated successfully
1063  * \retval              negative error number on failure
1064  */
1065 int lfsck_namespace_update_lmv(const struct lu_env *env,
1066                                struct lfsck_component *com,
1067                                struct dt_object *obj,
1068                                struct lmv_mds_md_v1 *lmv, bool locked)
1069 {
1070         struct lfsck_thread_info *info = lfsck_env_info(env);
1071         struct lmv_mds_md_v1 *lmv4 = &info->lti_lmv4;
1072         struct lu_buf *buf = &info->lti_buf;
1073         struct lfsck_instance *lfsck = com->lc_lfsck;
1074         struct dt_device *dev = lfsck_obj2dev(obj);
1075         struct thandle *th = NULL;
1076         struct lustre_handle lh = { 0 };
1077         int rc = 0;
1078         int rc1 = 0;
1079
1080         ENTRY;
1081         LASSERT(lmv4 != lmv);
1082
1083         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1084                 GOTO(out, rc = 0);
1085
1086         lfsck_lmv_header_cpu_to_le(lmv4, lmv);
1087         lfsck_buf_init(buf, lmv4, sizeof(*lmv4));
1088
1089         if (!locked) {
1090                 rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
1091                                       MDS_INODELOCK_UPDATE |
1092                                       MDS_INODELOCK_XATTR, LCK_EX);
1093                 if (rc != 0)
1094                         GOTO(log, rc);
1095         }
1096
1097         th = lfsck_trans_create(env, dev, lfsck);
1098         if (IS_ERR(th))
1099                 GOTO(log, rc = PTR_ERR(th));
1100
1101         /* For remote updating LMV EA, there will be further LFSCK action on
1102          * remote MDT after the updating, so update the LMV EA synchronously.
1103          */
1104         if (dt_object_remote(obj))
1105                 th->th_sync = 1;
1106
1107         rc = dt_declare_xattr_set(env, obj, buf, XATTR_NAME_LMV, 0, th);
1108         if (rc != 0)
1109                 GOTO(stop, rc);
1110
1111         rc = dt_trans_start_local(env, dev, th);
1112         if (rc != 0)
1113                 GOTO(stop, rc);
1114
1115         dt_write_lock(env, obj, 0);
1116         if (unlikely(lfsck_is_dead_obj(obj)))
1117                 GOTO(unlock, rc = 1);
1118
1119         rc = dt_xattr_set(env, obj, buf, XATTR_NAME_LMV, 0, th);
1120
1121         GOTO(unlock, rc);
1122
1123 unlock:
1124         dt_write_unlock(env, obj);
1125
1126 stop:
1127         rc1 = dt_trans_stop(env, dev, th);
1128         if (rc == 0)
1129                 rc = rc1;
1130
1131 log:
1132         lfsck_ibits_unlock(&lh, LCK_EX);
1133
1134 out:
1135         CDEBUG(D_LFSCK,
1136                "%s: namespace LFSCK updated the %s LMV EA for the object "DFID": rc = %d\n",
1137                lfsck_lfsck2name(lfsck),
1138                lmv->lmv_magic == LMV_MAGIC ? "master" : "slave",
1139                PFID(lfsck_dto2fid(obj)), rc);
1140
1141         return rc;
1142 }
1143
1144 /**
1145  * Check whether allow to re-genereate the lost master LMV EA.
1146  *
1147  * If the master MDT-object of the striped directory lost its master LMV EA,
1148  * then before the LFSCK repaired the striped directory, some ones may have
1149  * created some objects (that are not normal shards of the striped directory)
1150  * under the master MDT-object. If such case happend, then the LFSCK cannot
1151  * re-generate the lost master LMV EA to keep those objects to be visible to
1152  * client.
1153  *
1154  * \param[in] env       pointer to the thread context
1155  * \param[in] com       pointer to the lfsck component
1156  * \param[in] obj       pointer to the master MDT-object to be checked
1157  * \param[in] cfid      the shard's FID used for verification
1158  * \param[in] cidx      the shard's index used for verification
1159  *
1160  * \retval              positive number if not allow to re-generate LMV EA
1161  * \retval              zero if allow to re-generate LMV EA
1162  * \retval              negative error number on failure
1163  */
1164 static int lfsck_allow_regenerate_master_lmv(const struct lu_env *env,
1165                                              struct lfsck_component *com,
1166                                              struct dt_object *obj,
1167                                              const struct lu_fid *cfid,
1168                                              __u32 cidx)
1169 {
1170         struct lfsck_thread_info *info = lfsck_env_info(env);
1171         struct lu_fid *tfid = &info->lti_fid3;
1172         struct lfsck_instance *lfsck = com->lc_lfsck;
1173         struct lu_dirent *ent = (struct lu_dirent *)info->lti_key;
1174         const struct dt_it_ops *iops;
1175         struct dt_it *di;
1176         __u64 cookie;
1177         __u32 args;
1178         int rc;
1179         __u16 type;
1180
1181         ENTRY;
1182         if (unlikely(!dt_try_as_dir(env, obj, true)))
1183                 RETURN(-ENOTDIR);
1184
1185         /* Check whether the shard and the master MDT-object matches or not. */
1186         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf), DFID":%u",
1187                  PFID(cfid), cidx);
1188         rc = dt_lookup(env, obj, (struct dt_rec *)tfid,
1189                        (const struct dt_key *)info->lti_tmpbuf);
1190         if (rc != 0)
1191                 RETURN(rc);
1192
1193         if (!lu_fid_eq(tfid, cfid))
1194                 RETURN(-ENOENT);
1195
1196         args = lfsck->li_args_dir & ~(LUDA_VERIFY | LUDA_VERIFY_DRYRUN);
1197         iops = &obj->do_index_ops->dio_it;
1198         di = iops->init(env, obj, args);
1199         if (IS_ERR(di))
1200                 RETURN(PTR_ERR(di));
1201
1202         rc = iops->load(env, di, 0);
1203         if (rc == 0)
1204                 rc = iops->next(env, di);
1205         else if (rc > 0)
1206                 rc = 0;
1207
1208         if (rc != 0)
1209                 GOTO(out, rc);
1210
1211         do {
1212                 rc = iops->rec(env, di, (struct dt_rec *)ent, args);
1213                 if (rc == 0)
1214                         rc = lfsck_unpack_ent(ent, &cookie, &type);
1215
1216                 if (rc != 0)
1217                         GOTO(out, rc);
1218
1219                 /* skip dot and dotdot entries */
1220                 if (name_is_dot_or_dotdot(ent->lde_name, ent->lde_namelen))
1221                         goto next;
1222
1223                 /* If the subdir name does not match the shard name rule, then
1224                  * it is quite possible that it is NOT a shard, but created by
1225                  * someone after the master MDT-object lost the master LMV EA.
1226                  * But it is also possible that the subdir name entry crashed,
1227                  * under such double failure cases, the LFSCK cannot know how
1228                  * to repair the inconsistency. For data safe, the LFSCK will
1229                  * mark the master MDT-object as read-only. The administrator
1230                  * can fix the bad shard name manually, then run LFSCK again.
1231                  *
1232                  * XXX: If the subdir name matches the shard name rule, but it
1233                  *      is not a real shard of the striped directory, instead,
1234                  *      it was created by someone after the master MDT-object
1235                  *      lost the LMV EA, then re-generating the master LMV EA
1236                  *      will cause such subdir to be invisible to client, and
1237                  *      if its index occupies some lost shard index, then the
1238                  *      LFSCK will use it to replace the bad shard, and cause
1239                  *      the subdir (itself) to be invisible for ever.
1240                  */
1241                 if (lfsck_shard_name_to_index(env, ent->lde_name,
1242                                 ent->lde_namelen, type, &ent->lde_fid) < 0)
1243                         GOTO(out, rc = 1);
1244
1245 next:
1246                 rc = iops->next(env, di);
1247         } while (rc == 0);
1248
1249         GOTO(out, rc = 0);
1250
1251 out:
1252         iops->put(env, di);
1253         iops->fini(env, di);
1254
1255         return rc;
1256 }
1257
1258 /**
1259  * Notify remote LFSCK instance that the object's LMV EA has been updated.
1260  *
1261  * \param[in] env       pointer to the thread context
1262  * \param[in] com       pointer to the lfsck component
1263  * \param[in] obj       pointer to the object on which the LMV EA will be set
1264  * \param[in] event     indicate either master or slave LMV EA has been updated
1265  * \param[in] flags     indicate which element(s) in the LMV EA has been updated
1266  * \param[in] index     the MDT index on which the LFSCK instance to be notified
1267  *
1268  * \retval              positive number if nothing to be done
1269  * \retval              zero for success
1270  * \retval              negative error number on failure
1271  */
1272 static int lfsck_namespace_notify_lmv_remote(const struct lu_env *env,
1273                                              struct lfsck_component *com,
1274                                              struct dt_object *obj,
1275                                              __u32 event, __u32 flags,
1276                                              __u32 index)
1277 {
1278         struct lfsck_request *lr = &lfsck_env_info(env)->lti_lr;
1279         const struct lu_fid *fid = lfsck_dto2fid(obj);
1280         struct lfsck_instance *lfsck = com->lc_lfsck;
1281         struct lfsck_tgt_desc *ltd = NULL;
1282         struct ptlrpc_request *req = NULL;
1283         int rc;
1284
1285         ENTRY;
1286         ltd = lfsck_tgt_get(&lfsck->li_mdt_descs, index);
1287         if (ltd == NULL)
1288                 GOTO(out, rc = -ENODEV);
1289
1290         req = ptlrpc_request_alloc(class_exp2cliimp(ltd->ltd_exp),
1291                                    &RQF_LFSCK_NOTIFY);
1292         if (req == NULL)
1293                 GOTO(out, rc = -ENOMEM);
1294
1295         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
1296         if (rc != 0) {
1297                 ptlrpc_request_free(req);
1298
1299                 GOTO(out, rc);
1300         }
1301
1302         lr = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
1303         memset(lr, 0, sizeof(*lr));
1304         lr->lr_event = event;
1305         lr->lr_index = lfsck_dev_idx(lfsck);
1306         lr->lr_active = LFSCK_TYPE_NAMESPACE;
1307         lr->lr_fid = *fid;
1308         lr->lr_flags = flags;
1309
1310         ptlrpc_request_set_replen(req);
1311         rc = ptlrpc_queue_wait(req);
1312         ptlrpc_req_put(req);
1313
1314         GOTO(out, rc = (rc == -ENOENT ? 1 : rc));
1315
1316 out:
1317         CDEBUG(D_LFSCK,
1318                "%s: namespace LFSCK notify LMV EA updated for the object "DFID" on MDT %x remotely with event %u, flags %u: rc = %d\n",
1319                lfsck_lfsck2name(lfsck), PFID(fid), index, event, flags, rc);
1320
1321         if (ltd != NULL)
1322                 lfsck_tgt_put(ltd);
1323
1324         return rc;
1325 }
1326
1327 /**
1328  * Generate request for local LFSCK instance to rescan the striped directory.
1329  *
1330  * \param[in] env       pointer to the thread context
1331  * \param[in] com       pointer to the lfsck component
1332  * \param[in] obj       pointer to the striped directory to be rescanned
1333  *
1334  * \retval              positive number if nothing to be done
1335  * \retval              zero for success
1336  * \retval              negative error number on failure
1337  */
1338 int lfsck_namespace_notify_lmv_master_local(const struct lu_env *env,
1339                                             struct lfsck_component *com,
1340                                             struct dt_object *obj)
1341 {
1342         struct lfsck_instance *lfsck = com->lc_lfsck;
1343         struct lfsck_namespace *ns = com->lc_file_ram;
1344         struct lmv_mds_md_v1 *lmv4 = &lfsck_env_info(env)->lti_lmv4;
1345         struct lfsck_lmv_unit *llu;
1346         struct lfsck_lmv *llmv;
1347         struct lfsck_slave_lmv_rec *lslr;
1348         int count = 0;
1349         int rc;
1350
1351         ENTRY;
1352         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1353                 RETURN(0);
1354
1355         rc = lfsck_read_stripe_lmv(env, lfsck, obj, lmv4);
1356         if (rc != 0)
1357                 RETURN(rc);
1358
1359         OBD_ALLOC_PTR(llu);
1360         if (unlikely(llu == NULL))
1361                 RETURN(-ENOMEM);
1362
1363         if (lmv4->lmv_stripe_count < 1)
1364                 count = LFSCK_LMV_DEF_STRIPES;
1365         else if (lmv4->lmv_stripe_count > LFSCK_LMV_MAX_STRIPES)
1366                 count = LFSCK_LMV_MAX_STRIPES;
1367         else
1368                 count = lmv4->lmv_stripe_count;
1369
1370         OBD_ALLOC_PTR_ARRAY_LARGE(lslr, count);
1371         if (lslr == NULL) {
1372                 OBD_FREE_PTR(llu);
1373
1374                 RETURN(-ENOMEM);
1375         }
1376
1377         INIT_LIST_HEAD(&llu->llu_link);
1378         llu->llu_lfsck = lfsck;
1379         llu->llu_obj = lfsck_object_get(obj);
1380         llmv = &llu->llu_lmv;
1381         llmv->ll_lmv_master = 1;
1382         llmv->ll_inline = 1;
1383         atomic_set(&llmv->ll_ref, 1);
1384         llmv->ll_stripes_allocated = count;
1385         llmv->ll_hash_type = LMV_HASH_TYPE_UNKNOWN;
1386         llmv->ll_lslr = lslr;
1387         llmv->ll_lmv = *lmv4;
1388
1389         down_write(&com->lc_sem);
1390         if (ns->ln_status != LS_SCANNING_PHASE1 &&
1391             ns->ln_status != LS_SCANNING_PHASE2) {
1392                 ns->ln_striped_dirs_skipped++;
1393                 up_write(&com->lc_sem);
1394                 lfsck_lmv_put(env, llmv);
1395         } else {
1396                 ns->ln_striped_dirs_repaired++;
1397                 llmv->ll_counted = 1;
1398                 spin_lock(&lfsck->li_lock);
1399                 list_add_tail(&llu->llu_link, &lfsck->li_list_lmv);
1400                 spin_unlock(&lfsck->li_lock);
1401                 up_write(&com->lc_sem);
1402         }
1403
1404         RETURN(0);
1405 }
1406
1407 /**
1408  * Set master LMV EA for the specified striped directory.
1409  *
1410  * First, if the master MDT-object of a striped directory lost its LMV EA,
1411  * then there may be some users have created some files under the master
1412  * MDT-object directly. Under such case, the LFSCK cannot re-generate LMV
1413  * EA for the master MDT-object, because we should keep the existing files
1414  * to be visible to client. Then the LFSCK will mark the striped directory
1415  * as read-only and keep it there to be handled by administrator manually.
1416  *
1417  * If nobody has created files under the master MDT-object of the striped
1418  * directory, then we will set the master LMV EA and generate a new rescan
1419  * (the striped directory) request that will be handled later by the LFSCK
1420  * instance on the MDT later.
1421  *
1422  * \param[in] env       pointer to the thread context
1423  * \param[in] com       pointer to the lfsck component
1424  * \param[in] obj       pointer to the object on which the LMV EA will be set
1425  * \param[in] lmv       pointer to the buffer holding the new LMV EA
1426  * \param[in] cfid      the shard's FID used for verification
1427  * \param[in] cidx      the shard's index used for verification
1428  * \param[in] flags     to indicate which element(s) in the LMV EA will be set
1429  *
1430  * \retval              positive number if nothing to be done
1431  * \retval              zero for success
1432  * \retval              negative error number on failure
1433  */
1434 static int lfsck_namespace_set_lmv_master(const struct lu_env *env,
1435                                           struct lfsck_component *com,
1436                                           struct dt_object *obj,
1437                                           struct lmv_mds_md_v1 *lmv,
1438                                           const struct lu_fid *cfid,
1439                                           __u32 cidx, __u32 flags)
1440 {
1441         struct lfsck_thread_info *info = lfsck_env_info(env);
1442         struct lmv_mds_md_v1 *lmv3 = &info->lti_lmv3;
1443         struct lu_seq_range *range = &info->lti_range;
1444         struct lfsck_instance *lfsck = com->lc_lfsck;
1445         struct seq_server_site *ss = lfsck_dev_site(lfsck);
1446         struct lustre_handle lh = { 0 };
1447         int pidx = -1;
1448         int rc = 0;
1449
1450         ENTRY;
1451         fld_range_set_mdt(range);
1452         rc = fld_server_lookup(env, ss->ss_server_fld,
1453                                fid_seq(lfsck_dto2fid(obj)), range);
1454         if (rc != 0)
1455                 GOTO(log, rc);
1456
1457         pidx = range->lsr_index;
1458         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
1459                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
1460                               LCK_EX);
1461         if (rc != 0)
1462                 GOTO(log, rc);
1463
1464         rc = lfsck_read_stripe_lmv(env, lfsck, obj, lmv3);
1465         if (rc == -ENODATA) {
1466                 if (!(flags & LEF_SET_LMV_ALL))
1467                         GOTO(log, rc);
1468
1469                 *lmv3 = *lmv;
1470         } else if (rc == 0) {
1471                 if (flags & LEF_SET_LMV_ALL)
1472                         GOTO(log, rc = 1);
1473
1474                 if (flags & LEF_SET_LMV_HASH)
1475                         lmv3->lmv_hash_type = lmv->lmv_hash_type;
1476         } else {
1477                 GOTO(log, rc);
1478         }
1479
1480         lmv3->lmv_magic = LMV_MAGIC;
1481         lmv3->lmv_master_mdt_index = pidx;
1482         lmv3->lmv_layout_version++;
1483
1484         if (flags & LEF_SET_LMV_ALL) {
1485                 rc = lfsck_allow_regenerate_master_lmv(env, com, obj,
1486                                                        cfid, cidx);
1487                 if (rc > 0) {
1488                         rc = lfsck_disable_master_lmv(env, com, obj, false);
1489
1490                         GOTO(log, rc = (rc == 0 ? 1 : rc));
1491                 }
1492
1493                 if (rc < 0)
1494                         GOTO(log, rc);
1495
1496                 /* To indicate that the master has ever lost LMV EA. */
1497                 lmv3->lmv_hash_type |= LMV_HASH_FLAG_LOST_LMV;
1498         }
1499
1500         rc = lfsck_namespace_update_lmv(env, com, obj, lmv3, true);
1501         if (rc == 0 && flags & LEF_SET_LMV_ALL) {
1502                 if (dt_object_remote(obj))
1503                         rc = lfsck_namespace_notify_lmv_remote(env, com, obj,
1504                                                 LE_SET_LMV_MASTER, 0, pidx);
1505                 else
1506                         rc = lfsck_namespace_notify_lmv_master_local(env, com,
1507                                                                      obj);
1508         }
1509
1510         GOTO(log, rc);
1511
1512 log:
1513         lfsck_ibits_unlock(&lh, LCK_EX);
1514         CDEBUG(D_LFSCK, "%s: namespace LFSCK set master LMV EA for the object "
1515                DFID" on the %s MDT %d, flags %x: rc = %d\n",
1516                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)),
1517                dt_object_remote(obj) ? "remote" : "local", pidx, flags, rc);
1518
1519         if (rc <= 0) {
1520                 struct lfsck_namespace *ns = com->lc_file_ram;
1521
1522                 ns->ln_flags |= LF_INCONSISTENT;
1523         }
1524
1525         return rc;
1526 }
1527
1528 /**
1529  * Repair the bad name hash.
1530  *
1531  * If the name hash of some name entry under the striped directory does not
1532  * match the shard of the striped directory, then the LFSCK will repair the
1533  * inconsistency. Ideally, the LFSCK should migrate the name entry from the
1534  * current MDT to the right MDT (another one), but before the async commit
1535  * finished, the LFSCK will change the striped directory's hash type as
1536  * LMV_HASH_TYPE_UNKNOWN and mark the lmv flags as LMV_HASH_FLAG_BAD_TYPE.
1537  *
1538  * \param[in] env       pointer to the thread context
1539  * \param[in] com       pointer to the lfsck component
1540  * \param[in] shard     pointer to the shard of the striped directory that
1541  *                      contains the bad name entry
1542  * \param[in] llmv      pointer to lfsck LMV EA structure
1543  * \param[in] name      the name of the bad name hash
1544  *
1545  * \retval              positive number if nothing to be done
1546  * \retval              zero for success
1547  * \retval              negative error number on failure
1548  */
1549 int lfsck_namespace_repair_bad_name_hash(const struct lu_env *env,
1550                                          struct lfsck_component *com,
1551                                          struct dt_object *shard,
1552                                          struct lfsck_lmv *llmv,
1553                                          const char *name)
1554 {
1555         struct lfsck_thread_info *info = lfsck_env_info(env);
1556         struct lu_fid *pfid = &info->lti_fid3;
1557         struct lmv_mds_md_v1 *lmv2 = &info->lti_lmv2;
1558         struct lfsck_instance *lfsck = com->lc_lfsck;
1559         struct dt_object *parent = NULL;
1560         int rc = 0;
1561
1562         ENTRY;
1563         rc = dt_lookup_dir(env, shard, dotdot, pfid);
1564         if (rc != 0 || !fid_is_sane(pfid))
1565                 GOTO(log, rc);
1566
1567         parent = lfsck_object_find_bottom(env, lfsck, pfid);
1568         if (IS_ERR(parent))
1569                 GOTO(log, rc = PTR_ERR(parent));
1570
1571         if (unlikely(!dt_object_exists(parent)))
1572                 /* The parent object was previously accessed when verifying
1573                  * the slave LMV EA.  If this condition is true it is because
1574                  * the striped directory is being removed.
1575                  */
1576                 GOTO(log, rc = 1);
1577
1578         *lmv2 = llmv->ll_lmv;
1579         /* only set BAD_TYPE here, do not clear hash type or MIGRATION flag,
1580          * so that user can resume dir migration if this is caused by dir
1581          * migration failure.
1582          */
1583         lmv2->lmv_hash_type |= LMV_HASH_FLAG_BAD_TYPE;
1584         rc = lfsck_namespace_set_lmv_master(env, com, parent, lmv2,
1585                                             lfsck_dto2fid(shard),
1586                                             llmv->ll_lmv.lmv_master_mdt_index,
1587                                             LEF_SET_LMV_HASH);
1588
1589         GOTO(log, rc);
1590
1591 log:
1592         CDEBUG(D_LFSCK,
1593                "%s: namespace LFSCK assistant found bad name hash on the MDT %x, parent "DFID", name %s, shard_%x "DFID": rc = %d\n",
1594                lfsck_lfsck2name(lfsck), lfsck_dev_idx(lfsck),
1595                PFID(pfid), name, llmv->ll_lmv.lmv_master_mdt_index,
1596                PFID(lfsck_dto2fid(shard)), rc);
1597
1598         if (parent != NULL && !IS_ERR(parent))
1599                 lfsck_object_put(env, parent);
1600
1601         return rc;
1602 }
1603
1604 /**
1605  * Scan the shard of a striped directory for name hash verification.
1606  *
1607  * During the first-stage scanning, if the LFSCK cannot make sure whether
1608  * the shard of a stripe directory contains valid slave LMV EA or not, then
1609  * it will skip the name hash verification for this shard temporarily, and
1610  * record the shard's FID in the LFSCK tracing file. As the LFSCK processing,
1611  * the slave LMV EA may has been verified/fixed by LFSCK instance on master.
1612  * Then in the second-stage scanning, the shard will be re-scanned, and for
1613  * every name entry under the shard, the name hash will be verified, and for
1614  * unmatched name entry, the LFSCK will try to fix it.
1615  *
1616  * \param[in] env       pointer to the thread context
1617  * \param[in] com       pointer to the lfsck component
1618  * \param[in] child     pointer to the directory object to be handled
1619  *
1620  * \retval              positive number for scanning successfully
1621  * \retval              zero for the scanning is paused
1622  * \retval              negative error number on failure
1623  */
1624 int lfsck_namespace_scan_shard(const struct lu_env *env,
1625                                struct lfsck_component *com,
1626                                struct dt_object *child)
1627 {
1628         struct lfsck_thread_info *info = lfsck_env_info(env);
1629         struct lmv_mds_md_v1 *lmv = &info->lti_lmv;
1630         struct lfsck_instance *lfsck = com->lc_lfsck;
1631         struct lfsck_namespace *ns = com->lc_file_ram;
1632         struct ptlrpc_thread *thread = &lfsck->li_thread;
1633         struct lu_dirent *ent = (struct lu_dirent *)info->lti_key;
1634         struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
1635         struct lfsck_lmv *llmv = NULL;
1636         const struct dt_it_ops *iops;
1637         struct dt_it *di;
1638         __u64 cookie;
1639         __u32 args;
1640         int rc;
1641         __u16 type;
1642
1643         ENTRY;
1644         rc = lfsck_read_stripe_lmv(env, lfsck, child, lmv);
1645         if (rc != 0)
1646                 RETURN(rc == -ENODATA ? 1 : rc);
1647
1648         if (lmv->lmv_magic != LMV_MAGIC_STRIPE)
1649                 RETURN(1);
1650
1651         if (unlikely(!dt_try_as_dir(env, child, true)))
1652                 RETURN(-ENOTDIR);
1653
1654         OBD_ALLOC_PTR(llmv);
1655         if (llmv == NULL)
1656                 RETURN(-ENOMEM);
1657
1658         llmv->ll_lmv_slave = 1;
1659         llmv->ll_lmv_verified = 1;
1660         llmv->ll_lmv = *lmv;
1661         atomic_set(&llmv->ll_ref, 1);
1662
1663         args = lfsck->li_args_dir & ~(LUDA_VERIFY | LUDA_VERIFY_DRYRUN);
1664         iops = &child->do_index_ops->dio_it;
1665         di = iops->init(env, child, args);
1666         if (IS_ERR(di))
1667                 GOTO(out, rc = PTR_ERR(di));
1668
1669         rc = iops->load(env, di, 0);
1670         if (rc == 0)
1671                 rc = iops->next(env, di);
1672         else if (rc > 0)
1673                 rc = 0;
1674
1675         while (rc == 0) {
1676                 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
1677                     unlikely(!thread_is_running(thread)))
1678                         GOTO(out, rc = 0);
1679
1680                 rc = iops->rec(env, di, (struct dt_rec *)ent, args);
1681                 if (rc == 0)
1682                         rc = lfsck_unpack_ent(ent, &cookie, &type);
1683
1684                 if (rc != 0) {
1685                         if (bk->lb_param & LPF_FAILOUT)
1686                                 GOTO(out, rc);
1687
1688                         goto next;
1689                 }
1690
1691                 /* skip dot and dotdot entries */
1692                 if (name_is_dot_or_dotdot(ent->lde_name, ent->lde_namelen))
1693                         goto next;
1694
1695                 if (!lfsck_is_valid_slave_name_entry(env, llmv, ent->lde_name,
1696                                                      ent->lde_namelen)) {
1697                         ns->ln_flags |= LF_INCONSISTENT;
1698                         rc = lfsck_namespace_repair_bad_name_hash(env, com,
1699                                                 child, llmv, ent->lde_name);
1700                         if (rc == 0)
1701                                 ns->ln_name_hash_repaired++;
1702                 }
1703
1704                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
1705                         GOTO(out, rc);
1706
1707                 /* Rate control. */
1708                 lfsck_control_speed(lfsck);
1709                 if (unlikely(!thread_is_running(thread)))
1710                         GOTO(out, rc = 0);
1711
1712                 if (CFS_FAIL_CHECK(OBD_FAIL_LFSCK_FATAL2)) {
1713                         spin_lock(&lfsck->li_lock);
1714                         thread_set_flags(thread, SVC_STOPPING);
1715                         spin_unlock(&lfsck->li_lock);
1716
1717                         GOTO(out, rc = -EINVAL);
1718                 }
1719
1720 next:
1721                 rc = iops->next(env, di);
1722         }
1723
1724         GOTO(out, rc);
1725
1726 out:
1727         iops->put(env, di);
1728         iops->fini(env, di);
1729         lfsck_lmv_put(env, llmv);
1730
1731         return rc;
1732 }
1733
1734 /**
1735  * Verify the slave object's (of striped directory) LMV EA.
1736  *
1737  * For the slave object of a striped directory, before traversing the shard
1738  * the LFSCK will verify whether its slave LMV EA matches its parent's master
1739  * LMV EA or not.
1740  *
1741  * \param[in] env       pointer to the thread context
1742  * \param[in] com       pointer to the lfsck component
1743  * \param[in] obj       pointer to the object which LMV EA will be checked
1744  * \param[in] llmv      pointer to buffer holding the slave LMV EA
1745  *
1746  * \retval              positive number if nothing to be done
1747  * \retval              zero for success
1748  * \retval              negative error number on failure
1749  */
1750 int lfsck_namespace_verify_stripe_slave(const struct lu_env *env,
1751                                         struct lfsck_component *com,
1752                                         struct dt_object *obj,
1753                                         struct lfsck_lmv *llmv)
1754 {
1755         struct lfsck_thread_info *info = lfsck_env_info(env);
1756         char *name = info->lti_key;
1757         char *name2;
1758         struct lu_fid *pfid = &info->lti_fid3;
1759         const struct lu_fid *cfid = lfsck_dto2fid(obj);
1760         struct lu_fid tfid;
1761         struct lfsck_instance *lfsck = com->lc_lfsck;
1762         struct lmv_mds_md_v1 *clmv = &llmv->ll_lmv;
1763         struct lmv_mds_md_v1 *plmv = &info->lti_lmv;
1764         struct dt_object *parent = NULL;
1765         int rc = 0;
1766
1767         ENTRY;
1768         if (!lfsck_is_valid_slave_lmv(clmv)) {
1769                 rc = lfsck_namespace_trace_update(env, com, cfid,
1770                                         LNTF_UNCERTAIN_LMV, true);
1771
1772                 GOTO(out, rc);
1773         }
1774
1775         rc = dt_lookup_dir(env, obj, dotdot, pfid);
1776         if (rc != 0 || !fid_is_sane(pfid)) {
1777                 rc = lfsck_namespace_trace_update(env, com, cfid,
1778                                         LNTF_UNCERTAIN_LMV, true);
1779
1780                 GOTO(out, rc);
1781         }
1782
1783         CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_ENGINE_DELAY, cfs_fail_val);
1784
1785         parent = lfsck_object_find_bottom(env, lfsck, pfid);
1786         if (IS_ERR(parent)) {
1787                 rc = lfsck_namespace_trace_update(env, com, cfid,
1788                                         LNTF_UNCERTAIN_LMV, true);
1789
1790                 GOTO(out, rc);
1791         }
1792
1793         if (unlikely(!dt_object_exists(parent)))
1794                 GOTO(out, rc = 1);
1795
1796         if (unlikely(!dt_try_as_dir(env, parent, true)))
1797                 GOTO(out, rc = -ENOTDIR);
1798
1799         rc = lfsck_read_stripe_lmv(env, lfsck, parent, plmv);
1800         if (rc != 0) {
1801                 int rc1;
1802
1803                 /* If the parent has no LMV EA, then it maybe because:
1804                  * 1) The parent lost the LMV EA.
1805                  * 2) The child claims a wrong (slave) LMV EA.
1806                  */
1807                 if (rc == -ENODATA)
1808                         rc = lfsck_namespace_set_lmv_master(env, com, parent,
1809                                         clmv, cfid, clmv->lmv_master_mdt_index,
1810                                         LEF_SET_LMV_ALL);
1811                 else
1812                         rc = 0;
1813
1814                 rc1 = lfsck_namespace_trace_update(env, com, cfid,
1815                                                    LNTF_UNCERTAIN_LMV, true);
1816
1817                 GOTO(out, rc = (rc < 0 ? rc : rc1));
1818         }
1819
1820         /* Unmatched magic or stripe count. */
1821         if (unlikely(plmv->lmv_magic != LMV_MAGIC ||
1822                      plmv->lmv_stripe_count != clmv->lmv_stripe_count)) {
1823                 rc = lfsck_namespace_trace_update(env, com, cfid,
1824                                                   LNTF_UNCERTAIN_LMV, true);
1825
1826                 GOTO(out, rc);
1827         }
1828
1829         /* If the master hash type has been set as LMV_HASH_TYPE_UNKNOWN,
1830          * then the slave hash type is not important.
1831          */
1832         if ((plmv->lmv_hash_type & LMV_HASH_TYPE_MASK) ==
1833             LMV_HASH_TYPE_UNKNOWN &&
1834             plmv->lmv_hash_type & LMV_HASH_FLAG_BAD_TYPE)
1835                 GOTO(out, rc = 0);
1836
1837         /* Unmatched hash type. */
1838         if (unlikely((plmv->lmv_hash_type & LMV_HASH_TYPE_MASK) !=
1839                      (clmv->lmv_hash_type & LMV_HASH_TYPE_MASK))) {
1840                 rc = lfsck_namespace_trace_update(env, com, cfid,
1841                                                   LNTF_UNCERTAIN_LMV, true);
1842
1843                 GOTO(out, rc);
1844         }
1845
1846         snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2), DFID":%u",
1847                  PFID(cfid), clmv->lmv_master_mdt_index);
1848         name2 = info->lti_tmpbuf2;
1849
1850         rc = lfsck_links_get_first(env, obj, name, &tfid);
1851         if (rc == 0 && strcmp(name, name2) == 0 && lu_fid_eq(pfid, &tfid)) {
1852                 llmv->ll_lmv_verified = 1;
1853
1854                 GOTO(out, rc);
1855         }
1856
1857         rc = dt_lookup_dir(env, parent, name2, &tfid);
1858         if (rc != 0 || !lu_fid_eq(cfid, &tfid))
1859                 rc = lfsck_namespace_trace_update(env, com, cfid,
1860                                                   LNTF_UNCERTAIN_LMV, true);
1861         else
1862                 llmv->ll_lmv_verified = 1;
1863
1864         GOTO(out, rc);
1865
1866 out:
1867         if (parent != NULL && !IS_ERR(parent))
1868                 lfsck_object_put(env, parent);
1869
1870         return rc;
1871 }
1872
1873 /**
1874  * Double scan the striped directory or the shard.
1875  *
1876  * All the shards' under the given striped directory or its shard have
1877  * been scanned, the LFSCK has got the global knownledge about the LMV
1878  * EA consistency.
1879  *
1880  * If the target is one shard of a striped directory, then only needs to
1881  * update related tracing file.
1882  *
1883  * If the target is the master MDT-object of a striped directory, then the
1884  * LFSCK will make the decision about whether the master LMV EA is invalid
1885  * or not, and repair it if inconsistenct; for every shard of the striped
1886  * directory, whether the slave LMV EA is invalid or not, and repair it if
1887  * inconsistent.
1888  *
1889  * \param[in] env       pointer to the thread context
1890  * \param[in] com       pointer to the lfsck component
1891  * \param[in] lnr       pointer to the namespace request that contains the
1892  *                      striped directory or the shard
1893  *
1894  * \retval              zero for success
1895  * \retval              negative error number on failure
1896  */
1897 int lfsck_namespace_striped_dir_rescan(const struct lu_env *env,
1898                                        struct lfsck_component *com,
1899                                        struct lfsck_namespace_req *lnr)
1900 {
1901         struct lfsck_thread_info *info = lfsck_env_info(env);
1902         struct lfsck_instance *lfsck = com->lc_lfsck;
1903         struct lfsck_namespace *ns = com->lc_file_ram;
1904         struct lfsck_lmv *llmv = lnr->lnr_lmv;
1905         struct lmv_mds_md_v1 *lmv = &llmv->ll_lmv;
1906         struct lmv_mds_md_v1 *lmv2 = &info->lti_lmv2;
1907         struct lfsck_assistant_object *lso = lnr->lnr_lar.lar_parent;
1908         const struct lu_fid *pfid = &lso->lso_fid;
1909         struct dt_object *dir = NULL;
1910         struct dt_object *obj = NULL;
1911         struct lu_seq_range *range = &info->lti_range;
1912         struct seq_server_site *ss = lfsck_dev_site(lfsck);
1913         __u32 stripe_count;
1914         __u32 hash_type;
1915         int rc = 0;
1916         int i;
1917
1918         ENTRY;
1919         if (llmv->ll_lmv_slave) {
1920                 if (llmv->ll_lmv_verified) {
1921                         ns->ln_striped_shards_scanned++;
1922                         lfsck_namespace_trace_update(env, com, pfid,
1923                                         LNTF_UNCERTAIN_LMV |
1924                                         LNTF_RECHECK_NAME_HASH, false);
1925                 }
1926
1927                 RETURN(0);
1928         }
1929
1930         /* Either the striped directory has been disabled or only part of
1931          * the striped directory have been scanned. The LFSCK cannot repair
1932          * something based on incompleted knowledge. So skip it.
1933          */
1934         if (llmv->ll_ignore || llmv->ll_exit_value <= 0)
1935                 RETURN(0);
1936
1937         /* There ever been some failure, as to the LFSCK cannot know whether
1938          * it has got the global knowledge about the LMV EA consistency or not,
1939          * so it cannot make reparation about the incompleted knowledge.
1940          */
1941         if (llmv->ll_failed) {
1942                 ns->ln_striped_dirs_scanned++;
1943                 ns->ln_striped_dirs_failed++;
1944
1945                 RETURN(0);
1946         }
1947
1948         if (lmv->lmv_stripe_count > LFSCK_LMV_MAX_STRIPES)
1949                 stripe_count = max(llmv->ll_max_filled_off + 1,
1950                                    llmv->ll_max_stripe_count);
1951         else
1952                 stripe_count = max(llmv->ll_max_filled_off + 1,
1953                                    lmv->lmv_stripe_count);
1954
1955         if (lmv->lmv_stripe_count != stripe_count) {
1956                 lmv->lmv_stripe_count = stripe_count;
1957                 llmv->ll_lmv_updated = 1;
1958         }
1959
1960         if (!lmv_is_known_hash_type(lmv->lmv_hash_type) &&
1961             !(lmv->lmv_hash_type & LMV_HASH_FLAG_BAD_TYPE) &&
1962             lmv_is_known_hash_type(llmv->ll_hash_type)) {
1963                 hash_type = llmv->ll_hash_type & LMV_HASH_TYPE_MASK;
1964                 lmv->lmv_hash_type = llmv->ll_hash_type;
1965                 llmv->ll_lmv_updated = 1;
1966         } else {
1967                 hash_type = lmv->lmv_hash_type & LMV_HASH_TYPE_MASK;
1968                 if (!lmv_is_known_hash_type(hash_type))
1969                         hash_type = LMV_HASH_TYPE_UNKNOWN;
1970         }
1971
1972         if (llmv->ll_lmv_updated) {
1973                 if (dir == NULL) {
1974                         dir = lfsck_assistant_object_load(env, lfsck, lso);
1975                         if (IS_ERR(dir)) {
1976                                 rc = PTR_ERR(dir);
1977
1978                                 RETURN(rc == -ENOENT ? 0 : rc);
1979                         }
1980                 }
1981
1982                 lmv->lmv_layout_version++;
1983                 rc = lfsck_namespace_update_lmv(env, com, dir, lmv, false);
1984                 if (rc != 0)
1985                         RETURN(rc);
1986
1987                 ns->ln_striped_dirs_scanned++;
1988                 if (!llmv->ll_counted)
1989                         ns->ln_striped_dirs_repaired++;
1990         }
1991
1992         fld_range_set_mdt(range);
1993         for (i = 0; i <= llmv->ll_max_filled_off; i++) {
1994                 struct lfsck_slave_lmv_rec *lslr = llmv->ll_lslr + i;
1995                 const struct lu_fid *cfid = &lslr->lslr_fid;
1996                 const struct lu_name *cname;
1997                 struct linkea_data ldata = { NULL };
1998                 int rc1 = 0;
1999                 bool repair_linkea = false;
2000                 bool repair_lmvea = false;
2001                 bool rename = false;
2002                 bool create = false;
2003                 bool linkea_repaired = false;
2004                 bool lmvea_repaired = false;
2005                 bool rename_repaired = false;
2006                 bool create_repaired = false;
2007
2008                 /* LMV EA hole. */
2009                 if (fid_is_zero(cfid))
2010                         continue;
2011
2012                 lnr->lnr_fid = *cfid;
2013                 lnr->lnr_namelen = scnprintf(lnr->lnr_name,
2014                                              lnr->lnr_size - sizeof(*lnr),
2015                                              DFID":%u", PFID(cfid), i);
2016                 cname = lfsck_name_get_const(env, lnr->lnr_name,
2017                                              lnr->lnr_namelen);
2018                 obj = lfsck_object_find_bottom(env, lfsck, cfid);
2019                 if (IS_ERR(obj)) {
2020                         if (dir == NULL) {
2021                                 dir = lfsck_assistant_object_load(env, lfsck,
2022                                                                   lso);
2023                                 if (IS_ERR(dir)) {
2024                                         if (PTR_ERR(dir) == -ENOENT)
2025                                                 RETURN(0);
2026
2027                                         dir = NULL;
2028                                 }
2029                         } else if (lfsck_is_dead_obj(dir)) {
2030                                 GOTO(out, rc = 0);
2031                         }
2032
2033                         rc1 = PTR_ERR(obj);
2034                         goto next;
2035                 }
2036
2037                 switch (lslr->lslr_flags) {
2038                 case LSLF_NONE:
2039                         if (llmv->ll_inline ||
2040                             lslr->lslr_stripe_count != stripe_count ||
2041                             (lslr->lslr_hash_type & LMV_HASH_TYPE_MASK) !=
2042                              hash_type)
2043                                 repair_lmvea = true;
2044                         break;
2045                 case LSLF_BAD_INDEX2:
2046                         /* The index in the slave LMV EA is right,
2047                          * the name entry should be updated.
2048                          */
2049                         rename = true;
2050                         snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2),
2051                                  DFID":%u", PFID(cfid), lslr->lslr_index);
2052                         if (llmv->ll_inline ||
2053                             lslr->lslr_stripe_count != stripe_count ||
2054                             (lslr->lslr_hash_type & LMV_HASH_TYPE_MASK) !=
2055                              hash_type)
2056                                 repair_lmvea = true;
2057                         break;
2058                 case LSLF_BAD_INDEX1:
2059                         /* The index in the name entry is right,
2060                          * the slave LMV EA should be updated.
2061                          */
2062                 case LSLF_NO_LMVEA:
2063                         repair_lmvea = true;
2064                         break;
2065                 case LSLF_DANGLING:
2066                         create = true;
2067                         goto repair;
2068                 default:
2069                         break;
2070                 }
2071
2072                 rc1 = lfsck_links_read_with_rec(env, obj, &ldata);
2073                 if (rc1 == -ENOENT) {
2074                         create = true;
2075                         goto repair;
2076                 }
2077
2078                 if (rc1 == -EINVAL || rc1 == -ENODATA) {
2079                         repair_linkea = true;
2080                         goto repair;
2081                 }
2082
2083                 if (rc1 != 0)
2084                         goto next;
2085
2086                 if (ldata.ld_leh->leh_reccount != 1) {
2087                         repair_linkea = true;
2088                         goto repair;
2089                 }
2090
2091                 rc1 = linkea_links_find(&ldata, cname, pfid);
2092                 if (rc1 != 0)
2093                         repair_linkea = true;
2094
2095 repair:
2096                 if (create) {
2097                         if (dir == NULL) {
2098                                 dir = lfsck_assistant_object_load(env, lfsck,
2099                                                                   lso);
2100                                 if (IS_ERR(dir)) {
2101                                         rc1 = PTR_ERR(dir);
2102
2103                                         if (rc1 == -ENOENT)
2104                                                 GOTO(out, rc = 0);
2105
2106                                         dir = NULL;
2107                                         goto next;
2108                                 }
2109                         }
2110
2111                         rc1 = lfsck_namespace_repair_dangling(env, com, dir,
2112                                                               obj, lnr);
2113                         if (rc1 >= 0) {
2114                                 create_repaired = true;
2115                                 if (rc == 0)
2116                                         ns->ln_dangling_repaired++;
2117                         }
2118                 }
2119
2120                 if (repair_lmvea) {
2121                         *lmv2 = *lmv;
2122                         lmv2->lmv_magic = LMV_MAGIC_STRIPE;
2123                         lmv2->lmv_stripe_count = stripe_count;
2124                         lmv2->lmv_master_mdt_index = i;
2125                         lmv2->lmv_hash_type = hash_type;
2126
2127                         rc1 = lfsck_namespace_update_lmv(env, com, obj,
2128                                                          lmv2, false);
2129                         if (rc1 < 0)
2130                                 goto next;
2131
2132                         if (dt_object_remote(obj)) {
2133                                 rc1 = fld_server_lookup(env, ss->ss_server_fld,
2134                                         fid_seq(lfsck_dto2fid(obj)), range);
2135                                 if (rc1 != 0)
2136                                         goto next;
2137
2138                                 rc1 = lfsck_namespace_notify_lmv_remote(env,
2139                                                 com, obj, LE_SET_LMV_SLAVE, 0,
2140                                                 range->lsr_index);
2141                         } else {
2142                                 ns->ln_striped_shards_repaired++;
2143                                 rc1 = lfsck_namespace_trace_update(env, com,
2144                                         cfid, LNTF_RECHECK_NAME_HASH, true);
2145                         }
2146
2147                         if (rc1 < 0)
2148                                 goto next;
2149
2150                         if (rc1 >= 0)
2151                                 lmvea_repaired = true;
2152                 } else if (llmv->ll_inline) {
2153                         if (dt_object_remote(obj)) {
2154                                 rc1 = fld_server_lookup(env, ss->ss_server_fld,
2155                                         fid_seq(lfsck_dto2fid(obj)), range);
2156                                 if (rc1 != 0)
2157                                         goto next;
2158
2159                                 /* The slave LMV EA on the remote shard is
2160                                  * correct, just notify the LFSCK instance
2161                                  * on such MDT to re-verify the name_hash.
2162                                  */
2163                                 rc1 = lfsck_namespace_notify_lmv_remote(env,
2164                                                 com, obj, LE_SET_LMV_SLAVE,
2165                                                 LEF_RECHECK_NAME_HASH,
2166                                                 range->lsr_index);
2167                         } else {
2168                                 rc1 = lfsck_namespace_trace_update(env, com,
2169                                         cfid, LNTF_RECHECK_NAME_HASH, true);
2170                         }
2171
2172                         if (rc1 < 0)
2173                                 goto next;
2174                 }
2175
2176                 if (rename) {
2177                         if (dir == NULL) {
2178                                 dir = lfsck_assistant_object_load(env, lfsck,
2179                                                                   lso);
2180                                 if (IS_ERR(dir)) {
2181                                         rc1 = PTR_ERR(dir);
2182
2183                                         if (rc1 == -ENOENT)
2184                                                 GOTO(out, rc = 0);
2185
2186                                         dir = NULL;
2187                                         goto next;
2188                                 }
2189                         }
2190
2191                         rc1 = lfsck_namespace_repair_dirent(env, com, dir, obj,
2192                                         info->lti_tmpbuf2, lnr->lnr_name,
2193                                         lnr->lnr_type, true, false);
2194                         if (rc1 >= 0) {
2195                                 rename_repaired = true;
2196                                 if (rc1 > 0) {
2197                                         ns->ln_dirent_repaired++;
2198                                         rc1 = lfsck_namespace_trace_update(env,
2199                                                 com, cfid,
2200                                                 LNTF_RECHECK_NAME_HASH, true);
2201                                 }
2202                         }
2203
2204                         if (rc1 < 0)
2205                                 goto next;
2206                 }
2207
2208                 if (repair_linkea) {
2209                         struct lustre_handle lh = { 0 };
2210
2211                         if (dir == NULL) {
2212                                 dir = lfsck_assistant_object_load(env, lfsck,
2213                                                                   lso);
2214                                 if (IS_ERR(dir)) {
2215                                         rc1 = PTR_ERR(dir);
2216
2217                                         if (rc1 == -ENOENT)
2218                                                 GOTO(out, rc = 0);
2219
2220                                         dir = NULL;
2221                                         goto next;
2222                                 }
2223                         }
2224
2225                         rc1 = linkea_links_new(&ldata, &info->lti_big_buf,
2226                                                cname, lfsck_dto2fid(dir));
2227                         if (rc1 != 0)
2228                                 goto next;
2229
2230                         rc1 = lfsck_ibits_lock(env, lfsck, obj, &lh,
2231                                                MDS_INODELOCK_UPDATE |
2232                                                MDS_INODELOCK_XATTR, LCK_EX);
2233                         if (rc1 != 0)
2234                                 goto next;
2235
2236                         rc1 = lfsck_namespace_rebuild_linkea(env, com, obj,
2237                                                              &ldata);
2238                         lfsck_ibits_unlock(&lh, LCK_EX);
2239                         if (rc1 >= 0) {
2240                                 linkea_repaired = true;
2241                                 if (rc1 > 0)
2242                                         ns->ln_linkea_repaired++;
2243                         }
2244                 }
2245
2246 next:
2247                 if (create || rename || repair_linkea || repair_lmvea) {
2248                         CDEBUG(D_LFSCK,
2249                                "%s: namespace LFSCK repair the shard %d "DFID" of the striped directory "DFID" with dangling %s/%s, rename %s/%s, llinkea %s/%s, repair_lmvea %s/%s: rc = %d\n",
2250                                lfsck_lfsck2name(lfsck),
2251                                i, PFID(cfid), PFID(pfid),
2252                                create ? "yes" : "no",
2253                                create_repaired ? "yes" : "no",
2254                                rename ? "yes" : "no",
2255                                rename_repaired ? "yes" : "no",
2256                                repair_linkea ? "yes" : "no",
2257                                linkea_repaired ? "yes" : "no",
2258                                repair_lmvea ? "yes" : "no",
2259                                lmvea_repaired ? "yes" : "no", rc1);
2260                 }
2261
2262                 if (obj != NULL && !IS_ERR(obj)) {
2263                         lfsck_object_put(env, obj);
2264                         obj = NULL;
2265                 }
2266
2267                 if (rc1 < 0) {
2268                         rc = rc1;
2269                         ns->ln_striped_shards_failed++;
2270                 }
2271         }
2272
2273         GOTO(out, rc);
2274
2275 out:
2276         if (obj != NULL && !IS_ERR(obj))
2277                 lfsck_object_put(env, obj);
2278
2279         if (dir != NULL && !IS_ERR(dir))
2280                 lfsck_object_put(env, dir);
2281
2282         return rc;
2283 }
2284
2285 /**
2286  * Verify the shard's name entry under the striped directory.
2287  *
2288  * Before all shards of the striped directory scanned, the LFSCK cannot
2289  * know whether the master LMV EA is valid or not, and also cannot know
2290  * how to repair an invalid shard exactly. For example, the stripe index
2291  * stored in the shard's name does not match the stripe index stored in
2292  * the slave LMV EA, then the LFSCK cannot know which one is correct.
2293  * If the LFSCK just assumed one is correct, and fixed the other, then
2294  * as the LFSCK processing, it may find that the former reparation is
2295  * wrong and have to roll back. Unfortunately, if some applications saw
2296  * the changes and made further modification based on such changes, then
2297  * the roll back is almost impossible.
2298  *
2299  * To avoid above trouble, the LFSCK will scan the master object of the
2300  * striped directory twice, that is NOT the same as normal two-stages
2301  * scanning, the double scanning the striped directory will happen both
2302  * during the first-stage scanning:
2303  *
2304  * 1) When the striped directory is opened for scanning, the LFSCK will
2305  *    iterate each shard in turn, and records its slave LMV EA in the
2306  *    lfsck_lmv::ll_lslr. In this step, if the 'shard' (may be fake
2307  *    shard) name does not match the shard naming rule, for example, it
2308  *    does not contains the shard's FID, or not contains index, then we
2309  *    can remove the bad name entry directly. But if the name is valid,
2310  *    but the shard has no slave LMV EA or the slave LMV EA does not
2311  *    match its name, then we just record related information in the
2312  *    lfsck_lmv::ll_lslr in RAM.
2313  *
2314  * 2) When all the known shards have been scanned, then the engine will
2315  *    generate a dummy request (via lfsck_namespace_close_dir) to tell
2316  *    the assistant thread that all the known shards have been scanned.
2317  *    Since the assistant has got the global knowledge about the index
2318  *    conflict, stripe count, hash type, and so on. Then the assistant
2319  *    thread will scan the lfsck_lmv::ll_lslr, and for every shard in
2320  *    the record, check and repair inconsistency.
2321  *
2322  * Generally, the stripe directory has only several shards, and there
2323  * will NOT be a lof of striped directory. So double scanning striped
2324  * directory will not much affect the LFSCK performance.
2325  *
2326  * \param[in] env       pointer to the thread context
2327  * \param[in] com       pointer to the lfsck component
2328  * \param[in] lnr       pointer to the namespace request that contains the
2329  *                      shard's name, parent object, parent's LMV, and ect.
2330  *
2331  * \retval              zero for success
2332  * \retval              negative error number on failure
2333  */
2334 int lfsck_namespace_handle_striped_master(const struct lu_env *env,
2335                                           struct lfsck_component *com,
2336                                           struct lfsck_namespace_req *lnr)
2337 {
2338         struct lfsck_thread_info *info = lfsck_env_info(env);
2339         struct lmv_mds_md_v1 *lmv = &info->lti_lmv;
2340         struct lfsck_instance *lfsck = com->lc_lfsck;
2341         struct lfsck_namespace *ns = com->lc_file_ram;
2342         struct lfsck_lmv *llmv = lnr->lnr_lmv;
2343         struct lfsck_assistant_object *lso = lnr->lnr_lar.lar_parent;
2344         const struct lu_fid *pfid = &lso->lso_fid;
2345         struct dt_object *dir;
2346         struct dt_object *obj = NULL;
2347         struct dt_device *dev = NULL;
2348         int shard_idx = 0;
2349         int stripe = 0;
2350         int rc = 0;
2351         int depth = 0;
2352         bool repaired = false;
2353         enum lfsck_namespace_inconsistency_type type = LNIT_NONE;
2354
2355         ENTRY;
2356         if (unlikely(llmv->ll_ignore))
2357                 RETURN(0);
2358
2359         dir = lfsck_assistant_object_load(env, lfsck, lso);
2360         if (IS_ERR(dir)) {
2361                 rc = PTR_ERR(dir);
2362
2363                 RETURN(rc == -ENOENT ? 0 : rc);
2364         }
2365
2366         shard_idx = lfsck_find_mdt_idx_by_fid(env, lfsck, &lnr->lnr_fid);
2367         if (shard_idx < 0)
2368                 GOTO(fail_lmv, rc = shard_idx);
2369
2370         if (shard_idx == lfsck_dev_idx(lfsck)) {
2371                 if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
2372                         GOTO(out, rc = 0);
2373
2374                 dev = lfsck->li_bottom;
2375         } else {
2376                 struct lfsck_tgt_desc *ltd;
2377
2378                 /* Usually, some local filesystem consistency verification
2379                  * tools can guarantee the local namespace tree consistenct.
2380                  * So the LFSCK will only verify the remote directory.
2381                  */
2382                 if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0)) {
2383                         rc = lfsck_namespace_trace_update(env, com, pfid,
2384                                                 LNTF_CHECK_PARENT, true);
2385
2386                         GOTO(out, rc);
2387                 }
2388
2389                 ltd = lfsck_ltd2tgt(&lfsck->li_mdt_descs, shard_idx);
2390                 if (unlikely(ltd == NULL)) {
2391                         CDEBUG(D_LFSCK,
2392                                "%s: cannot talk with MDT %x which did not join the namespace LFSCK\n",
2393                                lfsck_lfsck2name(lfsck), shard_idx);
2394                         lfsck_lad_set_bitmap(env, com, shard_idx);
2395
2396                         GOTO(fail_lmv, rc = -ENODEV);
2397                 }
2398
2399                 dev = ltd->ltd_tgt;
2400         }
2401
2402         obj = lfsck_object_find_by_dev(env, dev, &lnr->lnr_fid);
2403         if (IS_ERR(obj)) {
2404                 if (lfsck_is_dead_obj(dir))
2405                         RETURN(0);
2406
2407                 GOTO(fail_lmv, rc = PTR_ERR(obj));
2408         }
2409
2410         if (!dt_object_exists(obj)) {
2411                 stripe = lfsck_shard_name_to_index(env, lnr->lnr_name,
2412                                 lnr->lnr_namelen, lnr->lnr_type, &lnr->lnr_fid);
2413                 if (stripe < 0) {
2414                         type = LNIT_BAD_DIRENT;
2415
2416                         GOTO(out, rc = 0);
2417                 }
2418
2419 dangling:
2420                 rc = lfsck_namespace_check_exist(env, dir, obj, lnr->lnr_name);
2421                 if (rc == 0) {
2422                         memset(lmv, 0, sizeof(*lmv));
2423                         lmv->lmv_magic = LMV_MAGIC;
2424                         rc = lfsck_record_lmv(env, com, dir, lnr, lmv, stripe,
2425                                               LSLF_DANGLING, LSLF_NONE, &depth);
2426                 }
2427
2428                 GOTO(out, rc);
2429         }
2430
2431         stripe = lfsck_shard_name_to_index(env, lnr->lnr_name, lnr->lnr_namelen,
2432                                            lfsck_object_type(obj),
2433                                            &lnr->lnr_fid);
2434         if (stripe < 0) {
2435                 type = LNIT_BAD_DIRENT;
2436
2437                 GOTO(out, rc = 0);
2438         }
2439
2440         rc = lfsck_read_stripe_lmv(env, lfsck, obj, lmv);
2441         if (unlikely(rc == -ENOENT))
2442                 /* It may happen when the remote object has been removed,
2443                  * but the local MDT does not aware of that.
2444                  */
2445                 goto dangling;
2446
2447         if (rc == -ENODATA)
2448                 rc = lfsck_record_lmv(env, com, dir, lnr, lmv, stripe,
2449                                       LSLF_NO_LMVEA, LSLF_NONE, &depth);
2450         else if (rc == 0)
2451                 rc = lfsck_record_lmv(env, com, dir, lnr, lmv, stripe,
2452                                       lmv->lmv_master_mdt_index != stripe ?
2453                                       LSLF_BAD_INDEX1 : LSLF_NONE, LSLF_NONE,
2454                                       &depth);
2455
2456         GOTO(out, rc);
2457
2458 fail_lmv:
2459         llmv->ll_failed = 1;
2460
2461 out:
2462         if (rc >= 0 && type == LNIT_NONE && !S_ISDIR(lnr->lnr_type))
2463                 type = LNIT_BAD_TYPE;
2464
2465         switch (type) {
2466         case LNIT_BAD_TYPE:
2467                 rc = lfsck_namespace_repair_dirent(env, com, dir, obj,
2468                                                    lnr->lnr_name, lnr->lnr_name,
2469                                                    lnr->lnr_type, true, false);
2470                 if (rc > 0)
2471                         repaired = true;
2472                 break;
2473         case LNIT_BAD_DIRENT:
2474                 rc = lfsck_namespace_repair_dirent(env, com, dir, obj,
2475                                                    lnr->lnr_name, lnr->lnr_name,
2476                                                    lnr->lnr_type, false, false);
2477                 if (rc > 0)
2478                         repaired = true;
2479                 break;
2480         default:
2481                 break;
2482         }
2483
2484         if (rc < 0) {
2485                 CDEBUG(D_LFSCK,
2486                        "%s: namespace LFSCK assistant fail to handle the shard: "DFID", parent "DFID", name %.*s: rc = %d\n",
2487                        lfsck_lfsck2name(lfsck), PFID(&lnr->lnr_fid),
2488                        PFID(pfid), lnr->lnr_namelen, lnr->lnr_name, rc);
2489
2490                 if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
2491                      rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
2492                      rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&
2493                     dev != NULL && dev != lfsck->li_bottom)
2494                         lfsck_lad_set_bitmap(env, com, shard_idx);
2495
2496                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT))
2497                         rc = 0;
2498         } else {
2499                 if (repaired) {
2500                         ns->ln_items_repaired++;
2501
2502                         switch (type) {
2503                         case LNIT_BAD_TYPE:
2504                                 ns->ln_bad_type_repaired++;
2505                                 break;
2506                         case LNIT_BAD_DIRENT:
2507                                 ns->ln_dirent_repaired++;
2508                                 break;
2509                         default:
2510                                 break;
2511                         }
2512                 }
2513
2514                 rc = 0;
2515         }
2516
2517         if (obj != NULL && !IS_ERR(obj))
2518                 lfsck_object_put(env, obj);
2519
2520         lfsck_object_put(env, dir);
2521
2522         return rc;
2523 }