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