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