Whamcloud - gitweb
LU-6351 lfsck: check object existence before using it
[fs/lustre-release.git] / 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, 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         int rc;
845
846         dt_read_lock(env, obj, 0);
847         rc = dt_xattr_get(env, obj, lfsck_buf_get(env, lmv, sizeof(*lmv)),
848                           XATTR_NAME_LMV);
849         dt_read_unlock(env, obj);
850         if (rc != sizeof(*lmv))
851                 return rc > 0 ? -EINVAL : rc;
852
853         lfsck_lmv_header_le_to_cpu(lmv, lmv);
854         if ((lmv->lmv_magic == LMV_MAGIC &&
855              !(lmv->lmv_hash_type & LMV_HASH_FLAG_MIGRATION)) ||
856             (lmv->lmv_magic == LMV_MAGIC_STRIPE &&
857              !(lmv->lmv_hash_type & LMV_HASH_FLAG_DEAD)))
858                 return 0;
859
860         return -ENODATA;
861 }
862
863 /**
864  * Parse the shard's index from the given shard name.
865  *
866  * The valid shard name/type should be:
867  * 1) The type must be S_IFDIR
868  * 2) The name should be $FID:$index
869  * 3) the index should within valid range.
870  *
871  * \param[in] env       pointer to the thread context
872  * \param[in] name      the shard name
873  * \param[in] namelen   the name length
874  * \param[in] type      the entry's type
875  * \param[in] fid       the entry's FID
876  *
877  * \retval              zero or positive number for the index from the name
878  * \retval              negative error number on failure
879  */
880 int lfsck_shard_name_to_index(const struct lu_env *env, const char *name,
881                               int namelen, __u16 type, const struct lu_fid *fid)
882 {
883         char    *name2  = lfsck_env_info(env)->lti_tmpbuf2;
884         int      len;
885         int      idx    = 0;
886
887         if (!S_ISDIR(type))
888                 return -ENOTDIR;
889
890         LASSERT(name != name2);
891
892         len = snprintf(name2, sizeof(lfsck_env_info(env)->lti_tmpbuf2),
893                        DFID":", PFID(fid));
894         if (namelen < len + 1 || memcmp(name, name2, len) != 0)
895                 return -EINVAL;
896
897         do {
898                 if (!isdigit(name[len]))
899                         return -EINVAL;
900
901                 idx = idx * 10 + name[len++] - '0';
902         } while (len < namelen);
903
904         if (idx >= LFSCK_LMV_MAX_STRIPES)
905                 return -EINVAL;
906
907         return idx;
908 }
909
910 bool lfsck_is_valid_slave_name_entry(const struct lu_env *env,
911                                      struct lfsck_lmv *llmv,
912                                      const char *name, int namelen)
913 {
914         struct lmv_mds_md_v1    *lmv;
915         int                      idx;
916
917         if (llmv == NULL || !llmv->ll_lmv_slave || !llmv->ll_lmv_verified)
918                 return true;
919
920         lmv = &llmv->ll_lmv;
921         idx = lmv_name_to_stripe_index(lmv->lmv_hash_type,
922                                        lmv->lmv_stripe_count,
923                                        name, namelen);
924         if (unlikely(idx != lmv->lmv_master_mdt_index))
925                 return false;
926
927         return true;
928 }
929
930 /**
931  * Check whether the given name is a valid entry under the @parent.
932  *
933  * If the @parent is a striped directory then the @child should one
934  * shard of the striped directory, its name should be $FID:$index.
935  *
936  * If the @parent is a shard of a striped directory, then the name hash
937  * should match the MDT, otherwise it is invalid.
938  *
939  * \param[in] env       pointer to the thread context
940  * \param[in] parent    the parent directory
941  * \param[in] child     the child object to be checked
942  * \param[in] cname     the name for the @child in the parent directory
943  *
944  * \retval              positive number for invalid name entry
945  * \retval              0 if the name is valid or uncertain
946  * \retval              negative error number on failure
947  */
948 int lfsck_namespace_check_name(const struct lu_env *env,
949                                struct dt_object *parent,
950                                struct dt_object *child,
951                                const struct lu_name *cname)
952 {
953         struct lmv_mds_md_v1    *lmv = &lfsck_env_info(env)->lti_lmv;
954         int                      idx;
955         int                      rc;
956
957         rc = lfsck_read_stripe_lmv(env, parent, lmv);
958         if (rc != 0)
959                 RETURN(rc == -ENODATA ? 0 : rc);
960
961         if (lmv->lmv_magic == LMV_MAGIC_STRIPE) {
962                 if (!lfsck_is_valid_slave_lmv(lmv))
963                         return 0;
964
965                 idx = lmv_name_to_stripe_index(lmv->lmv_hash_type,
966                                                lmv->lmv_stripe_count,
967                                                cname->ln_name,
968                                                cname->ln_namelen);
969                 if (unlikely(idx != lmv->lmv_master_mdt_index))
970                         return 1;
971         } else if (lfsck_shard_name_to_index(env, cname->ln_name,
972                         cname->ln_namelen, lfsck_object_type(child),
973                         lfsck_dto2fid(child)) < 0) {
974                 return 1;
975         }
976
977         return 0;
978 }
979
980 /**
981  * Update the object's LMV EA with the given @lmv.
982  *
983  * \param[in] env       pointer to the thread context
984  * \param[in] com       pointer to the lfsck component
985  * \param[in] obj       pointer to the object which LMV EA will be updated
986  * \param[in] lmv       pointer to buffer holding the new LMV EA
987  * \param[in] locked    whether the caller has held ldlm lock on the @obj or not
988  *
989  * \retval              positive number for nothing to be done
990  * \retval              zero if updated successfully
991  * \retval              negative error number on failure
992  */
993 int lfsck_namespace_update_lmv(const struct lu_env *env,
994                                struct lfsck_component *com,
995                                struct dt_object *obj,
996                                struct lmv_mds_md_v1 *lmv, bool locked)
997 {
998         struct lfsck_thread_info        *info   = lfsck_env_info(env);
999         struct lmv_mds_md_v1            *lmv4   = &info->lti_lmv4;
1000         struct lu_buf                   *buf    = &info->lti_buf;
1001         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1002         struct dt_device                *dev    = lfsck_obj2dev(obj);
1003         struct thandle                  *th     = NULL;
1004         struct lustre_handle             lh     = { 0 };
1005         int                              rc     = 0;
1006         int                              rc1    = 0;
1007         ENTRY;
1008
1009         LASSERT(lmv4 != lmv);
1010
1011         lfsck_lmv_header_cpu_to_le(lmv4, lmv);
1012         lfsck_buf_init(buf, lmv4, sizeof(*lmv4));
1013
1014         if (!locked) {
1015                 rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
1016                                       MDS_INODELOCK_UPDATE |
1017                                       MDS_INODELOCK_XATTR, LCK_EX);
1018                 if (rc != 0)
1019                         GOTO(log, rc);
1020         }
1021
1022         th = dt_trans_create(env, dev);
1023         if (IS_ERR(th))
1024                 GOTO(log, rc = PTR_ERR(th));
1025
1026         /* For remote updating LMV EA, there will be further LFSCK action on
1027          * remote MDT after the updating, so update the LMV EA synchronously. */
1028         if (dt_object_remote(obj))
1029                 th->th_sync = 1;
1030
1031         rc = dt_declare_xattr_set(env, obj, buf, XATTR_NAME_LMV, 0, th);
1032         if (rc != 0)
1033                 GOTO(stop, rc);
1034
1035         rc = dt_trans_start_local(env, dev, th);
1036         if (rc != 0)
1037                 GOTO(stop, rc);
1038
1039         dt_write_lock(env, obj, 0);
1040         if (unlikely(lfsck_is_dead_obj(obj)))
1041                 GOTO(unlock, rc = 1);
1042
1043         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1044                 GOTO(unlock, rc = 0);
1045
1046         rc = dt_xattr_set(env, obj, buf, XATTR_NAME_LMV, 0, th);
1047
1048         GOTO(unlock, rc);
1049
1050 unlock:
1051         dt_write_unlock(env, obj);
1052
1053 stop:
1054         rc1 = dt_trans_stop(env, dev, th);
1055         if (rc == 0)
1056                 rc = rc1;
1057
1058 log:
1059         lfsck_ibits_unlock(&lh, LCK_EX);
1060         CDEBUG(D_LFSCK, "%s: namespace LFSCK updated the %s LMV EA "
1061                "for the object "DFID": rc = %d\n",
1062                lfsck_lfsck2name(lfsck),
1063                lmv->lmv_magic == LMV_MAGIC ? "master" : "slave",
1064                PFID(lfsck_dto2fid(obj)), rc);
1065
1066         return rc;
1067 }
1068
1069 /**
1070  * Check whether allow to re-genereate the lost master LMV EA.
1071  *
1072  * If the master MDT-object of the striped directory lost its master LMV EA,
1073  * then before the LFSCK repaired the striped directory, some ones may have
1074  * created some objects (that are not normal shards of the striped directory)
1075  * under the master MDT-object. If such case happend, then the LFSCK cannot
1076  * re-generate the lost master LMV EA to keep those objects to be visible to
1077  * client.
1078  *
1079  * \param[in] env       pointer to the thread context
1080  * \param[in] com       pointer to the lfsck component
1081  * \param[in] obj       pointer to the master MDT-object to be checked
1082  * \param[in] cfid      the shard's FID used for verification
1083  * \param[in] cidx      the shard's index used for verification
1084  *
1085  * \retval              positive number if not allow to re-generate LMV EA
1086  * \retval              zero if allow to re-generate LMV EA
1087  * \retval              negative error number on failure
1088  */
1089 static int lfsck_allow_regenerate_master_lmv(const struct lu_env *env,
1090                                              struct lfsck_component *com,
1091                                              struct dt_object *obj,
1092                                              const struct lu_fid *cfid,
1093                                              __u32 cidx)
1094 {
1095         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1096         struct lu_fid                   *tfid   = &info->lti_fid3;
1097         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1098         struct lu_dirent                *ent    =
1099                         (struct lu_dirent *)info->lti_key;
1100         const struct dt_it_ops          *iops;
1101         struct dt_it                    *di;
1102         __u64                            cookie;
1103         __u32                            args;
1104         int                              rc;
1105         __u16                            type;
1106         ENTRY;
1107
1108         if (unlikely(!dt_try_as_dir(env, obj)))
1109                 RETURN(-ENOTDIR);
1110
1111         /* Check whether the shard and the master MDT-object matches or not. */
1112         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf), DFID":%u",
1113                  PFID(cfid), cidx);
1114         rc = dt_lookup(env, obj, (struct dt_rec *)tfid,
1115                        (const struct dt_key *)info->lti_tmpbuf);
1116         if (rc != 0)
1117                 RETURN(rc);
1118
1119         if (!lu_fid_eq(tfid, cfid))
1120                 RETURN(-ENOENT);
1121
1122         args = lfsck->li_args_dir & ~(LUDA_VERIFY | LUDA_VERIFY_DRYRUN);
1123         iops = &obj->do_index_ops->dio_it;
1124         di = iops->init(env, obj, args);
1125         if (IS_ERR(di))
1126                 RETURN(PTR_ERR(di));
1127
1128         rc = iops->load(env, di, 0);
1129         if (rc == 0)
1130                 rc = iops->next(env, di);
1131         else if (rc > 0)
1132                 rc = 0;
1133
1134         if (rc != 0)
1135                 GOTO(out, rc);
1136
1137         do {
1138                 rc = iops->rec(env, di, (struct dt_rec *)ent, args);
1139                 if (rc == 0)
1140                         rc = lfsck_unpack_ent(ent, &cookie, &type);
1141
1142                 if (rc != 0)
1143                         GOTO(out, rc);
1144
1145                 /* skip dot and dotdot entries */
1146                 if (name_is_dot_or_dotdot(ent->lde_name, ent->lde_namelen))
1147                         goto next;
1148
1149                 /* If the subdir name does not match the shard name rule, then
1150                  * it is quite possible that it is NOT a shard, but created by
1151                  * someone after the master MDT-object lost the master LMV EA.
1152                  * But it is also possible that the subdir name entry crashed,
1153                  * under such double failure cases, the LFSCK cannot know how
1154                  * to repair the inconsistency. For data safe, the LFSCK will
1155                  * mark the master MDT-object as read-only. The administrator
1156                  * can fix the bad shard name manually, then run LFSCK again.
1157                  *
1158                  * XXX: If the subdir name matches the shard name rule, but it
1159                  *      is not a real shard of the striped directory, instead,
1160                  *      it was created by someone after the master MDT-object
1161                  *      lost the LMV EA, then re-generating the master LMV EA
1162                  *      will cause such subdir to be invisible to client, and
1163                  *      if its index occupies some lost shard index, then the
1164                  *      LFSCK will use it to replace the bad shard, and cause
1165                  *      the subdir (itself) to be invisible for ever. */
1166                 if (lfsck_shard_name_to_index(env, ent->lde_name,
1167                                 ent->lde_namelen, type, &ent->lde_fid) < 0)
1168                         GOTO(out, rc = 1);
1169
1170 next:
1171                 rc = iops->next(env, di);
1172         } while (rc == 0);
1173
1174         GOTO(out, rc = 0);
1175
1176 out:
1177         iops->put(env, di);
1178         iops->fini(env, di);
1179
1180         return rc;
1181 }
1182
1183 /**
1184  * Notify remote LFSCK instance that the object's LMV EA has been updated.
1185  *
1186  * \param[in] env       pointer to the thread context
1187  * \param[in] com       pointer to the lfsck component
1188  * \param[in] obj       pointer to the object on which the LMV EA will be set
1189  * \param[in] event     indicate either master or slave LMV EA has been updated
1190  * \param[in] flags     indicate which element(s) in the LMV EA has been updated
1191  * \param[in] index     the MDT index on which the LFSCK instance to be notified
1192  *
1193  * \retval              positive number if nothing to be done
1194  * \retval              zero for success
1195  * \retval              negative error number on failure
1196  */
1197 static int lfsck_namespace_notify_lmv_remote(const struct lu_env *env,
1198                                              struct lfsck_component *com,
1199                                              struct dt_object *obj,
1200                                              __u32 event, __u32 flags,
1201                                              __u32 index)
1202 {
1203         struct lfsck_request            *lr     = &lfsck_env_info(env)->lti_lr;
1204         const struct lu_fid             *fid    = lfsck_dto2fid(obj);
1205         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1206         struct lfsck_tgt_desc           *ltd    = NULL;
1207         struct ptlrpc_request           *req    = NULL;
1208         int                              rc;
1209         ENTRY;
1210
1211         ltd = lfsck_tgt_get(&lfsck->li_mdt_descs, index);
1212         if (ltd == NULL)
1213                 GOTO(out, rc = -ENODEV);
1214
1215         req = ptlrpc_request_alloc(class_exp2cliimp(ltd->ltd_exp),
1216                                    &RQF_LFSCK_NOTIFY);
1217         if (req == NULL)
1218                 GOTO(out, rc = -ENOMEM);
1219
1220         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
1221         if (rc != 0) {
1222                 ptlrpc_request_free(req);
1223
1224                 GOTO(out, rc);
1225         }
1226
1227         lr = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
1228         memset(lr, 0, sizeof(*lr));
1229         lr->lr_event = event;
1230         lr->lr_index = lfsck_dev_idx(lfsck);
1231         lr->lr_active = LFSCK_TYPE_NAMESPACE;
1232         lr->lr_fid = *fid;
1233         lr->lr_flags = flags;
1234
1235         ptlrpc_request_set_replen(req);
1236         rc = ptlrpc_queue_wait(req);
1237         ptlrpc_req_finished(req);
1238
1239         GOTO(out, rc = (rc == -ENOENT ? 1 : rc));
1240
1241 out:
1242         CDEBUG(D_LFSCK, "%s: namespace LFSCK notify LMV EA updated for the "
1243                "object "DFID" on MDT %x remotely with event %u, flags %u: "
1244                "rc = %d\n", lfsck_lfsck2name(lfsck), PFID(fid), index,
1245                event, flags, rc);
1246
1247         if (ltd != NULL)
1248                 lfsck_tgt_put(ltd);
1249
1250         return rc;
1251 }
1252
1253 /**
1254  * Generate request for local LFSCK instance to rescan the striped directory.
1255  *
1256  * \param[in] env       pointer to the thread context
1257  * \param[in] com       pointer to the lfsck component
1258  * \param[in] obj       pointer to the striped directory to be rescanned
1259  *
1260  * \retval              positive number if nothing to be done
1261  * \retval              zero for success
1262  * \retval              negative error number on failure
1263  */
1264 int lfsck_namespace_notify_lmv_master_local(const struct lu_env *env,
1265                                             struct lfsck_component *com,
1266                                             struct dt_object *obj)
1267 {
1268         struct lfsck_instance      *lfsck = com->lc_lfsck;
1269         struct lfsck_namespace     *ns    = com->lc_file_ram;
1270         struct lmv_mds_md_v1       *lmv4  = &lfsck_env_info(env)->lti_lmv4;
1271         struct lfsck_lmv_unit      *llu;
1272         struct lfsck_lmv           *llmv;
1273         struct lfsck_slave_lmv_rec *lslr;
1274         int                         count = 0;
1275         int                         rc;
1276         ENTRY;
1277
1278         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1279                 RETURN(0);
1280
1281         rc = lfsck_read_stripe_lmv(env, obj, lmv4);
1282         if (rc != 0)
1283                 RETURN(rc);
1284
1285         OBD_ALLOC_PTR(llu);
1286         if (unlikely(llu == NULL))
1287                 RETURN(-ENOMEM);
1288
1289         if (lmv4->lmv_stripe_count < 1)
1290                 count = LFSCK_LMV_DEF_STRIPES;
1291         else if (lmv4->lmv_stripe_count > LFSCK_LMV_MAX_STRIPES)
1292                 count = LFSCK_LMV_MAX_STRIPES;
1293         else
1294                 count = lmv4->lmv_stripe_count;
1295
1296         OBD_ALLOC_LARGE(lslr, sizeof(struct lfsck_slave_lmv_rec) * count);
1297         if (lslr == NULL) {
1298                 OBD_FREE_PTR(llu);
1299
1300                 RETURN(-ENOMEM);
1301         }
1302
1303         INIT_LIST_HEAD(&llu->llu_link);
1304         llu->llu_lfsck = lfsck;
1305         llu->llu_obj = lfsck_object_get(obj);
1306         llmv = &llu->llu_lmv;
1307         llmv->ll_lmv_master = 1;
1308         llmv->ll_inline = 1;
1309         atomic_set(&llmv->ll_ref, 1);
1310         llmv->ll_stripes_allocated = count;
1311         llmv->ll_hash_type = LMV_HASH_TYPE_UNKNOWN;
1312         llmv->ll_lslr = lslr;
1313         llmv->ll_lmv = *lmv4;
1314
1315         down_write(&com->lc_sem);
1316         if (ns->ln_status != LS_SCANNING_PHASE1 &&
1317             ns->ln_status != LS_SCANNING_PHASE2) {
1318                 ns->ln_striped_dirs_skipped++;
1319                 up_write(&com->lc_sem);
1320                 lfsck_lmv_put(env, llmv);
1321         } else {
1322                 ns->ln_striped_dirs_repaired++;
1323                 spin_lock(&lfsck->li_lock);
1324                 list_add_tail(&llu->llu_link, &lfsck->li_list_lmv);
1325                 spin_unlock(&lfsck->li_lock);
1326                 up_write(&com->lc_sem);
1327         }
1328
1329         RETURN(0);
1330 }
1331
1332 /**
1333  * Set master LMV EA for the specified striped directory.
1334  *
1335  * First, if the master MDT-object of a striped directory lost its LMV EA,
1336  * then there may be some users have created some files under the master
1337  * MDT-object directly. Under such case, the LFSCK cannot re-generate LMV
1338  * EA for the master MDT-object, because we should keep the existing files
1339  * to be visible to client. Then the LFSCK will mark the striped directory
1340  * as read-only and keep it there to be handled by administrator manually.
1341  *
1342  * If nobody has created files under the master MDT-object of the striped
1343  * directory, then we will set the master LMV EA and generate a new rescan
1344  * (the striped directory) request that will be handled later by the LFSCK
1345  * instance on the MDT later.
1346  *
1347  * \param[in] env       pointer to the thread context
1348  * \param[in] com       pointer to the lfsck component
1349  * \param[in] obj       pointer to the object on which the LMV EA will be set
1350  * \param[in] lmv       pointer to the buffer holding the new LMV EA
1351  * \param[in] cfid      the shard's FID used for verification
1352  * \param[in] cidx      the shard's index used for verification
1353  * \param[in] flags     to indicate which element(s) in the LMV EA will be set
1354  *
1355  * \retval              positive number if nothing to be done
1356  * \retval              zero for success
1357  * \retval              negative error number on failure
1358  */
1359 static int lfsck_namespace_set_lmv_master(const struct lu_env *env,
1360                                           struct lfsck_component *com,
1361                                           struct dt_object *obj,
1362                                           struct lmv_mds_md_v1 *lmv,
1363                                           const struct lu_fid *cfid,
1364                                           __u32 cidx, __u32 flags)
1365 {
1366         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1367         struct lmv_mds_md_v1            *lmv3   = &info->lti_lmv3;
1368         struct lu_seq_range             *range  = &info->lti_range;
1369         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1370         struct seq_server_site          *ss     = lfsck_dev_site(lfsck);
1371         struct lustre_handle             lh     = { 0 };
1372         int                              pidx   = -1;
1373         int                              rc     = 0;
1374         ENTRY;
1375
1376         fld_range_set_mdt(range);
1377         rc = fld_server_lookup(env, ss->ss_server_fld,
1378                                fid_seq(lfsck_dto2fid(obj)), range);
1379         if (rc != 0)
1380                 GOTO(log, rc);
1381
1382         pidx = range->lsr_index;
1383         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
1384                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
1385                               LCK_EX);
1386         if (rc != 0)
1387                 GOTO(log, rc);
1388
1389         rc = lfsck_read_stripe_lmv(env, obj, lmv3);
1390         if (rc == -ENODATA) {
1391                 if (!(flags & LEF_SET_LMV_ALL))
1392                         GOTO(log, rc);
1393
1394                 *lmv3 = *lmv;
1395         } else if (rc == 0) {
1396                 if (flags & LEF_SET_LMV_ALL)
1397                         GOTO(log, rc = 1);
1398
1399                 if (flags & LEF_SET_LMV_HASH)
1400                         lmv3->lmv_hash_type = lmv->lmv_hash_type;
1401         } else {
1402                 GOTO(log, rc);
1403         }
1404
1405         lmv3->lmv_magic = LMV_MAGIC;
1406         lmv3->lmv_master_mdt_index = pidx;
1407
1408         if (flags & LEF_SET_LMV_ALL) {
1409                 rc = lfsck_allow_regenerate_master_lmv(env, com, obj,
1410                                                        cfid, cidx);
1411                 if (rc > 0) {
1412                         rc = lfsck_disable_master_lmv(env, com, obj, false);
1413
1414                         GOTO(log, rc = (rc == 0 ? 1 : rc));
1415                 }
1416
1417                 if (rc < 0)
1418                         GOTO(log, rc);
1419
1420                 /* To indicate that the master has ever lost LMV EA. */
1421                 lmv3->lmv_hash_type |= LMV_HASH_FLAG_LOST_LMV;
1422         }
1423
1424         rc = lfsck_namespace_update_lmv(env, com, obj, lmv3, true);
1425         if (rc == 0 && flags & LEF_SET_LMV_ALL) {
1426                 if (dt_object_remote(obj))
1427                         rc = lfsck_namespace_notify_lmv_remote(env, com, obj,
1428                                                 LE_SET_LMV_MASTER, 0, pidx);
1429                 else
1430                         rc = lfsck_namespace_notify_lmv_master_local(env, com,
1431                                                                      obj);
1432         }
1433
1434         GOTO(log, rc);
1435
1436 log:
1437         lfsck_ibits_unlock(&lh, LCK_EX);
1438         CDEBUG(D_LFSCK, "%s: namespace LFSCK set master LMV EA for the object "
1439                DFID" on the %s MDT %d, flags %x: rc = %d\n",
1440                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)),
1441                dt_object_remote(obj) ? "remote" : "local", pidx, flags, rc);
1442
1443         if (rc <= 0) {
1444                 struct lfsck_namespace *ns = com->lc_file_ram;
1445
1446                 ns->ln_flags |= LF_INCONSISTENT;
1447         }
1448
1449         return rc;
1450 }
1451
1452 /**
1453  * Repair the bad name hash.
1454  *
1455  * If the name hash of some name entry under the striped directory does not
1456  * match the shard of the striped directory, then the LFSCK will repair the
1457  * inconsistency. Ideally, the LFSCK should migrate the name entry from the
1458  * current MDT to the right MDT (another one), but before the async commit
1459  * finished, the LFSCK will change the striped directory's hash type as
1460  * LMV_HASH_TYPE_UNKNOWN and mark the lmv flags as LMV_HASH_FLAG_BAD_TYPE.
1461  *
1462  * \param[in] env       pointer to the thread context
1463  * \param[in] com       pointer to the lfsck component
1464  * \param[in] shard     pointer to the shard of the striped directory that
1465  *                      contains the bad name entry
1466  * \param[in] llmv      pointer to lfsck LMV EA structure
1467  * \param[in] name      the name of the bad name hash
1468  *
1469  * \retval              positive number if nothing to be done
1470  * \retval              zero for success
1471  * \retval              negative error number on failure
1472  */
1473 int lfsck_namespace_repair_bad_name_hash(const struct lu_env *env,
1474                                          struct lfsck_component *com,
1475                                          struct dt_object *shard,
1476                                          struct lfsck_lmv *llmv,
1477                                          const char *name)
1478 {
1479         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1480         struct lu_fid                   *pfid   = &info->lti_fid3;
1481         struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
1482         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1483         struct dt_object                *parent = NULL;
1484         int                              rc     = 0;
1485         ENTRY;
1486
1487         rc = dt_lookup(env, shard, (struct dt_rec *)pfid,
1488                        (const struct dt_key *)dotdot);
1489         if (rc != 0 || !fid_is_sane(pfid))
1490                 GOTO(log, rc);
1491
1492         parent = lfsck_object_find_bottom(env, lfsck, pfid);
1493         if (IS_ERR(parent))
1494                 GOTO(log, rc = PTR_ERR(parent));
1495
1496         if (unlikely(!dt_object_exists(parent)))
1497                 /* The parent object was previously accessed when verifying
1498                  * the slave LMV EA.  If this condition is true it is because
1499                  * the striped directory is being removed. */
1500                 GOTO(log, rc = 1);
1501
1502         *lmv2 = llmv->ll_lmv;
1503         lmv2->lmv_hash_type = LMV_HASH_TYPE_UNKNOWN | LMV_HASH_FLAG_BAD_TYPE;
1504         rc = lfsck_namespace_set_lmv_master(env, com, parent, lmv2,
1505                                             lfsck_dto2fid(shard),
1506                                             llmv->ll_lmv.lmv_master_mdt_index,
1507                                             LEF_SET_LMV_HASH);
1508
1509         GOTO(log, rc);
1510
1511 log:
1512         CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found bad name hash "
1513                "on the MDT %x, parent "DFID", name %s, shard_%x "DFID
1514                ": rc = %d\n",
1515                lfsck_lfsck2name(lfsck), lfsck_dev_idx(lfsck),
1516                PFID(pfid), name, llmv->ll_lmv.lmv_master_mdt_index,
1517                PFID(lfsck_dto2fid(shard)), rc);
1518
1519         if (parent != NULL && !IS_ERR(parent))
1520                 lfsck_object_put(env, parent);
1521
1522         return rc;
1523 }
1524
1525 /**
1526  * Scan the shard of a striped directory for name hash verification.
1527  *
1528  * During the first-stage scanning, if the LFSCK cannot make sure whether
1529  * the shard of a stripe directory contains valid slave LMV EA or not, then
1530  * it will skip the name hash verification for this shard temporarily, and
1531  * record the shard's FID in the LFSCK tracing file. As the LFSCK processing,
1532  * the slave LMV EA may has been verified/fixed by LFSCK instance on master.
1533  * Then in the second-stage scanning, the shard will be re-scanned, and for
1534  * every name entry under the shard, the name hash will be verified, and for
1535  * unmatched name entry, the LFSCK will try to fix it.
1536  *
1537  * \param[in] env       pointer to the thread context
1538  * \param[in] com       pointer to the lfsck component
1539  * \param[in] child     pointer to the directory object to be handled
1540  *
1541  * \retval              positive number for scanning successfully
1542  * \retval              zero for the scanning is paused
1543  * \retval              negative error number on failure
1544  */
1545 int lfsck_namespace_scan_shard(const struct lu_env *env,
1546                                struct lfsck_component *com,
1547                                struct dt_object *child)
1548 {
1549         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1550         struct lmv_mds_md_v1            *lmv    = &info->lti_lmv;
1551         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1552         struct lfsck_namespace          *ns     = com->lc_file_ram;
1553         struct ptlrpc_thread            *thread = &lfsck->li_thread;
1554         struct lu_dirent                *ent    =
1555                         (struct lu_dirent *)info->lti_key;
1556         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
1557         struct lfsck_lmv                *llmv   = NULL;
1558         const struct dt_it_ops          *iops;
1559         struct dt_it                    *di;
1560         __u64                            cookie;
1561         __u32                            args;
1562         int                              rc;
1563         __u16                            type;
1564         ENTRY;
1565
1566         rc = lfsck_read_stripe_lmv(env, child, lmv);
1567         if (rc != 0)
1568                 RETURN(rc == -ENODATA ? 1 : rc);
1569
1570         if (lmv->lmv_magic != LMV_MAGIC_STRIPE)
1571                 RETURN(1);
1572
1573         if (unlikely(!dt_try_as_dir(env, child)))
1574                 RETURN(-ENOTDIR);
1575
1576         OBD_ALLOC_PTR(llmv);
1577         if (llmv == NULL)
1578                 RETURN(-ENOMEM);
1579
1580         llmv->ll_lmv_slave = 1;
1581         llmv->ll_lmv_verified = 1;
1582         llmv->ll_lmv = *lmv;
1583         atomic_set(&llmv->ll_ref, 1);
1584
1585         args = lfsck->li_args_dir & ~(LUDA_VERIFY | LUDA_VERIFY_DRYRUN);
1586         iops = &child->do_index_ops->dio_it;
1587         di = iops->init(env, child, args);
1588         if (IS_ERR(di))
1589                 GOTO(out, rc = PTR_ERR(di));
1590
1591         rc = iops->load(env, di, 0);
1592         if (rc == 0)
1593                 rc = iops->next(env, di);
1594         else if (rc > 0)
1595                 rc = 0;
1596
1597         while (rc == 0) {
1598                 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
1599                     unlikely(!thread_is_running(thread)))
1600                         GOTO(out, rc = 0);
1601
1602                 rc = iops->rec(env, di, (struct dt_rec *)ent, args);
1603                 if (rc == 0)
1604                         rc = lfsck_unpack_ent(ent, &cookie, &type);
1605
1606                 if (rc != 0) {
1607                         if (bk->lb_param & LPF_FAILOUT)
1608                                 GOTO(out, rc);
1609
1610                         goto next;
1611                 }
1612
1613                 /* skip dot and dotdot entries */
1614                 if (name_is_dot_or_dotdot(ent->lde_name, ent->lde_namelen))
1615                         goto next;
1616
1617                 if (!lfsck_is_valid_slave_name_entry(env, llmv, ent->lde_name,
1618                                                      ent->lde_namelen)) {
1619                         ns->ln_flags |= LF_INCONSISTENT;
1620                         rc = lfsck_namespace_repair_bad_name_hash(env, com,
1621                                                 child, llmv, ent->lde_name);
1622                         if (rc == 0)
1623                                 ns->ln_name_hash_repaired++;
1624                 }
1625
1626                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
1627                         GOTO(out, rc);
1628
1629                 /* Rate control. */
1630                 lfsck_control_speed(lfsck);
1631                 if (unlikely(!thread_is_running(thread)))
1632                         GOTO(out, rc = 0);
1633
1634                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_FATAL2)) {
1635                         spin_lock(&lfsck->li_lock);
1636                         thread_set_flags(thread, SVC_STOPPING);
1637                         spin_unlock(&lfsck->li_lock);
1638
1639                         GOTO(out, rc = -EINVAL);
1640                 }
1641
1642 next:
1643                 rc = iops->next(env, di);
1644         }
1645
1646         GOTO(out, rc);
1647
1648 out:
1649         iops->put(env, di);
1650         iops->fini(env, di);
1651         lfsck_lmv_put(env, llmv);
1652
1653         return rc;
1654 }
1655
1656 /**
1657  * Verify the slave object's (of striped directory) LMV EA.
1658  *
1659  * For the slave object of a striped directory, before traversing the shard
1660  * the LFSCK will verify whether its slave LMV EA matches its parent's master
1661  * LMV EA or not.
1662  *
1663  * \param[in] env       pointer to the thread context
1664  * \param[in] com       pointer to the lfsck component
1665  * \param[in] obj       pointer to the object which LMV EA will be checked
1666  * \param[in] llmv      pointer to buffer holding the slave LMV EA
1667  *
1668  * \retval              positive number if nothing to be done
1669  * \retval              zero for success
1670  * \retval              negative error number on failure
1671  */
1672 int lfsck_namespace_verify_stripe_slave(const struct lu_env *env,
1673                                         struct lfsck_component *com,
1674                                         struct dt_object *obj,
1675                                         struct lfsck_lmv *llmv)
1676 {
1677         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1678         char                            *name   = info->lti_key;
1679         char                            *name2;
1680         struct lu_fid                   *pfid   = &info->lti_fid3;
1681         const struct lu_fid             *cfid   = lfsck_dto2fid(obj);
1682         struct lu_fid                    tfid;
1683         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1684         struct lmv_mds_md_v1            *clmv   = &llmv->ll_lmv;
1685         struct lmv_mds_md_v1            *plmv   = &info->lti_lmv;
1686         struct dt_object                *parent = NULL;
1687         int                              rc     = 0;
1688         ENTRY;
1689
1690         if (!lfsck_is_valid_slave_lmv(clmv)) {
1691                 rc = lfsck_namespace_trace_update(env, com, cfid,
1692                                         LNTF_UNCERTAIN_LMV, true);
1693
1694                 GOTO(out, rc);
1695         }
1696
1697         rc = dt_lookup(env, obj, (struct dt_rec *)pfid,
1698                        (const struct dt_key *)dotdot);
1699         if (rc != 0 || !fid_is_sane(pfid)) {
1700                 rc = lfsck_namespace_trace_update(env, com, cfid,
1701                                         LNTF_UNCERTAIN_LMV, true);
1702
1703                 GOTO(out, rc);
1704         }
1705
1706         parent = lfsck_object_find_bottom(env, lfsck, pfid);
1707         if (IS_ERR(parent)) {
1708                 rc = lfsck_namespace_trace_update(env, com, cfid,
1709                                         LNTF_UNCERTAIN_LMV, true);
1710
1711                 GOTO(out, rc);
1712         }
1713
1714         if (unlikely(!dt_object_exists(parent)))
1715                 GOTO(out, rc = 1);
1716
1717         if (unlikely(!dt_try_as_dir(env, parent)))
1718                 GOTO(out, rc = -ENOTDIR);
1719
1720         rc = lfsck_read_stripe_lmv(env, parent, plmv);
1721         if (rc != 0) {
1722                 int rc1;
1723
1724                 /* If the parent has no LMV EA, then it maybe because:
1725                  * 1) The parent lost the LMV EA.
1726                  * 2) The child claims a wrong (slave) LMV EA. */
1727                 if (rc == -ENODATA)
1728                         rc = lfsck_namespace_set_lmv_master(env, com, parent,
1729                                         clmv, cfid, clmv->lmv_master_mdt_index,
1730                                         LEF_SET_LMV_ALL);
1731                 else
1732                         rc = 0;
1733
1734                 rc1 = lfsck_namespace_trace_update(env, com, cfid,
1735                                                    LNTF_UNCERTAIN_LMV, true);
1736
1737                 GOTO(out, rc = (rc < 0 ? rc : rc1));
1738         }
1739
1740         /* Unmatched magic or stripe count. */
1741         if (unlikely(plmv->lmv_magic != LMV_MAGIC ||
1742                      plmv->lmv_stripe_count != clmv->lmv_stripe_count)) {
1743                 rc = lfsck_namespace_trace_update(env, com, cfid,
1744                                                   LNTF_UNCERTAIN_LMV, true);
1745
1746                 GOTO(out, rc);
1747         }
1748
1749         /* If the master hash type has been set as LMV_HASH_TYPE_UNKNOWN,
1750          * then the slave hash type is not important. */
1751         if ((plmv->lmv_hash_type & LMV_HASH_TYPE_MASK) ==
1752             LMV_HASH_TYPE_UNKNOWN &&
1753             plmv->lmv_hash_type & LMV_HASH_FLAG_BAD_TYPE)
1754                 GOTO(out, rc = 0);
1755
1756         /* Unmatched hash type. */
1757         if (unlikely((plmv->lmv_hash_type & LMV_HASH_TYPE_MASK) !=
1758                      (clmv->lmv_hash_type & LMV_HASH_TYPE_MASK))) {
1759                 rc = lfsck_namespace_trace_update(env, com, cfid,
1760                                                   LNTF_UNCERTAIN_LMV, true);
1761
1762                 GOTO(out, rc);
1763         }
1764
1765         snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2), DFID":%u",
1766                  PFID(cfid), clmv->lmv_master_mdt_index);
1767         name2 = info->lti_tmpbuf2;
1768
1769         rc = lfsck_links_get_first(env, obj, name, &tfid);
1770         if (rc == 0 && strcmp(name, name2) == 0 && lu_fid_eq(pfid, &tfid)) {
1771                 llmv->ll_lmv_verified = 1;
1772
1773                 GOTO(out, rc);
1774         }
1775
1776         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1777                        (const struct dt_key *)name2);
1778         if (rc != 0 || !lu_fid_eq(cfid, &tfid))
1779                 rc = lfsck_namespace_trace_update(env, com, cfid,
1780                                                   LNTF_UNCERTAIN_LMV, true);
1781         else
1782                 llmv->ll_lmv_verified = 1;
1783
1784         GOTO(out, rc);
1785
1786 out:
1787         if (parent != NULL && !IS_ERR(parent))
1788                 lfsck_object_put(env, parent);
1789
1790         return rc;
1791 }
1792
1793 /**
1794  * Double scan the striped directory or the shard.
1795  *
1796  * All the shards' under the given striped directory or its shard have
1797  * been scanned, the LFSCK has got the global knownledge about the LMV
1798  * EA consistency.
1799  *
1800  * If the target is one shard of a striped directory, then only needs to
1801  * update related tracing file.
1802  *
1803  * If the target is the master MDT-object of a striped directory, then the
1804  * LFSCK will make the decision about whether the master LMV EA is invalid
1805  * or not, and repair it if inconsistenct; for every shard of the striped
1806  * directory, whether the slave LMV EA is invalid or not, and repair it if
1807  * inconsistent.
1808  *
1809  * \param[in] env       pointer to the thread context
1810  * \param[in] com       pointer to the lfsck component
1811  * \param[in] lnr       pointer to the namespace request that contains the
1812  *                      striped directory or the shard
1813  *
1814  * \retval              zero for success
1815  * \retval              negative error number on failure
1816  */
1817 int lfsck_namespace_striped_dir_rescan(const struct lu_env *env,
1818                                        struct lfsck_component *com,
1819                                        struct lfsck_namespace_req *lnr)
1820 {
1821         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1822         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1823         struct lfsck_namespace          *ns     = com->lc_file_ram;
1824         struct lfsck_lmv                *llmv   = lnr->lnr_lmv;
1825         struct lmv_mds_md_v1            *lmv    = &llmv->ll_lmv;
1826         struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
1827         struct lfsck_assistant_object   *lso    = lnr->lnr_lar.lar_parent;
1828         const struct lu_fid             *pfid   = &lso->lso_fid;
1829         struct dt_object                *dir    = NULL;
1830         struct dt_object                *obj    = NULL;
1831         struct lu_seq_range             *range  = &info->lti_range;
1832         struct seq_server_site          *ss     = lfsck_dev_site(lfsck);
1833         __u32                            stripe_count;
1834         __u32                            hash_type;
1835         int                              rc     = 0;
1836         int                              i;
1837         ENTRY;
1838
1839         if (llmv->ll_lmv_slave) {
1840                 if (llmv->ll_lmv_verified) {
1841                         ns->ln_striped_shards_scanned++;
1842                         lfsck_namespace_trace_update(env, com, pfid,
1843                                         LNTF_UNCERTAIN_LMV |
1844                                         LNTF_RECHECK_NAME_HASH, false);
1845                 }
1846
1847                 RETURN(0);
1848         }
1849
1850         /* Either the striped directory has been disabled or only part of
1851          * the striped directory have been scanned. The LFSCK cannot repair
1852          * something based on incompleted knowledge. So skip it. */
1853         if (llmv->ll_ignore || llmv->ll_exit_value <= 0)
1854                 RETURN(0);
1855
1856         /* There ever been some failure, as to the LFSCK cannot know whether
1857          * it has got the global knowledge about the LMV EA consistency or not,
1858          * so it cannot make reparation about the incompleted knowledge. */
1859         if (llmv->ll_failed) {
1860                 ns->ln_striped_dirs_scanned++;
1861                 ns->ln_striped_dirs_failed++;
1862
1863                 RETURN(0);
1864         }
1865
1866         if (lmv->lmv_stripe_count > LFSCK_LMV_MAX_STRIPES)
1867                 stripe_count = max(llmv->ll_max_filled_off + 1,
1868                                    llmv->ll_max_stripe_count);
1869         else
1870                 stripe_count = max(llmv->ll_max_filled_off + 1,
1871                                    lmv->lmv_stripe_count);
1872
1873         if (lmv->lmv_stripe_count != stripe_count) {
1874                 lmv->lmv_stripe_count = stripe_count;
1875                 llmv->ll_lmv_updated = 1;
1876         }
1877
1878         if (!lmv_is_known_hash_type(lmv->lmv_hash_type) &&
1879             !(lmv->lmv_hash_type & LMV_HASH_FLAG_BAD_TYPE) &&
1880             lmv_is_known_hash_type(llmv->ll_hash_type)) {
1881                 hash_type = llmv->ll_hash_type & LMV_HASH_TYPE_MASK;
1882                 lmv->lmv_hash_type = llmv->ll_hash_type;
1883                 llmv->ll_lmv_updated = 1;
1884         } else {
1885                 hash_type = lmv->lmv_hash_type & LMV_HASH_TYPE_MASK;
1886                 if (!lmv_is_known_hash_type(hash_type))
1887                         hash_type = LMV_HASH_TYPE_UNKNOWN;
1888         }
1889
1890         if (llmv->ll_lmv_updated) {
1891                 if (dir == NULL) {
1892                         dir = lfsck_assistant_object_load(env, lfsck, lso);
1893                         if (IS_ERR(dir)) {
1894                                 rc = PTR_ERR(dir);
1895
1896                                 RETURN(rc == -ENOENT ? 0 : rc);
1897                         }
1898                 }
1899
1900                 lmv->lmv_layout_version++;
1901                 rc = lfsck_namespace_update_lmv(env, com, dir, lmv, false);
1902                 if (rc != 0)
1903                         RETURN(rc);
1904
1905                 ns->ln_striped_dirs_scanned++;
1906                 ns->ln_striped_dirs_repaired++;
1907         }
1908
1909         fld_range_set_mdt(range);
1910         for (i = 0; i <= llmv->ll_max_filled_off; i++) {
1911                 struct lfsck_slave_lmv_rec *lslr = llmv->ll_lslr + i;
1912                 const struct lu_fid *cfid = &lslr->lslr_fid;
1913                 const struct lu_name *cname;
1914                 struct linkea_data ldata = { NULL };
1915                 int len;
1916                 int rc1 = 0;
1917                 bool repair_linkea = false;
1918                 bool repair_lmvea = false;
1919                 bool rename = false;
1920                 bool create = false;
1921                 bool linkea_repaired = false;
1922                 bool lmvea_repaired = false;
1923                 bool rename_repaired = false;
1924                 bool create_repaired = false;
1925
1926                 /* LMV EA hole. */
1927                 if (fid_is_zero(cfid))
1928                         continue;
1929
1930                 len = snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
1931                                DFID":%u", PFID(cfid), i);
1932                 cname = lfsck_name_get_const(env, info->lti_tmpbuf, len);
1933                 memcpy(lnr->lnr_name, info->lti_tmpbuf, len);
1934
1935                 obj = lfsck_object_find_bottom_nowait(env, lfsck, cfid);
1936                 if (IS_ERR(obj)) {
1937                         if (dir == NULL) {
1938                                 dir = lfsck_assistant_object_load(env, lfsck,
1939                                                                   lso);
1940                                 if (IS_ERR(dir)) {
1941                                         if (PTR_ERR(dir) == -ENOENT)
1942                                                 RETURN(0);
1943
1944                                         dir = NULL;
1945                                 }
1946                         } else if (lfsck_is_dead_obj(dir)) {
1947                                 GOTO(out, rc = 0);
1948                         }
1949
1950                         rc1 = PTR_ERR(obj);
1951                         goto next;
1952                 }
1953
1954                 switch (lslr->lslr_flags) {
1955                 case LSLF_NONE:
1956                         if (llmv->ll_inline ||
1957                             lslr->lslr_stripe_count != stripe_count ||
1958                             (lslr->lslr_hash_type & LMV_HASH_TYPE_MASK) !=
1959                              hash_type)
1960                                 repair_lmvea = true;
1961                         break;
1962                 case LSLF_BAD_INDEX2:
1963                         /* The index in the slave LMV EA is right,
1964                          * the name entry should be updated. */
1965                         rename = true;
1966                         snprintf(info->lti_tmpbuf2, sizeof(info->lti_tmpbuf2),
1967                                  DFID":%u", PFID(cfid), lslr->lslr_index);
1968                         if (llmv->ll_inline ||
1969                             lslr->lslr_stripe_count != stripe_count ||
1970                             (lslr->lslr_hash_type & LMV_HASH_TYPE_MASK) !=
1971                              hash_type)
1972                                 repair_lmvea = true;
1973                         break;
1974                 case LSLF_BAD_INDEX1:
1975                         /* The index in the name entry is right,
1976                          * the slave LMV EA should be updated. */
1977                 case LSLF_NO_LMVEA:
1978                         repair_lmvea = true;
1979                         break;
1980                 case LSLF_DANGLING:
1981                         create = true;
1982                         goto repair;
1983                 default:
1984                         break;
1985                 }
1986
1987                 rc1 = lfsck_links_read(env, obj, &ldata);
1988                 if (rc1 == -ENOENT) {
1989                         create = true;
1990                         goto repair;
1991                 }
1992
1993                 if (rc1 == -EINVAL || rc1 == -ENODATA) {
1994                         repair_linkea = true;
1995                         goto repair;
1996                 }
1997
1998                 if (rc1 != 0)
1999                         goto next;
2000
2001                 if (ldata.ld_leh->leh_reccount != 1) {
2002                         repair_linkea = true;
2003                         goto repair;
2004                 }
2005
2006                 rc1 = linkea_links_find(&ldata, cname, pfid);
2007                 if (rc1 != 0)
2008                         repair_linkea = true;
2009
2010 repair:
2011                 if (create) {
2012                         if (dir == NULL) {
2013                                 dir = lfsck_assistant_object_load(env, lfsck,
2014                                                                   lso);
2015                                 if (IS_ERR(dir)) {
2016                                         rc1 = PTR_ERR(dir);
2017
2018                                         if (rc1 == -ENOENT)
2019                                                 GOTO(out, rc = 0);
2020
2021                                         dir = NULL;
2022                                         goto next;
2023                                 }
2024                         }
2025
2026                         rc1 = lfsck_namespace_repair_dangling(env, com, dir,
2027                                                               obj, lnr);
2028                         if (rc1 >= 0) {
2029                                 create_repaired = true;
2030                                 if (rc == 0)
2031                                         ns->ln_dangling_repaired++;
2032                         }
2033                 }
2034
2035                 if (repair_lmvea) {
2036                         *lmv2 = *lmv;
2037                         lmv2->lmv_magic = LMV_MAGIC_STRIPE;
2038                         lmv2->lmv_stripe_count = stripe_count;
2039                         lmv2->lmv_master_mdt_index = i;
2040                         lmv2->lmv_hash_type = hash_type;
2041
2042                         rc1 = lfsck_namespace_update_lmv(env, com, obj,
2043                                                          lmv2, false);
2044                         if (rc1 < 0)
2045                                 goto next;
2046
2047                         if (dt_object_remote(obj)) {
2048                                 rc1 = fld_server_lookup(env, ss->ss_server_fld,
2049                                         fid_seq(lfsck_dto2fid(obj)), range);
2050                                 if (rc1 != 0)
2051                                         goto next;
2052
2053                                 rc1 = lfsck_namespace_notify_lmv_remote(env,
2054                                                 com, obj, LE_SET_LMV_SLAVE, 0,
2055                                                 range->lsr_index);
2056                         } else {
2057                                 ns->ln_striped_shards_repaired++;
2058                                 rc1 = lfsck_namespace_trace_update(env, com,
2059                                         cfid, LNTF_RECHECK_NAME_HASH, true);
2060                         }
2061
2062                         if (rc1 < 0)
2063                                 goto next;
2064
2065                         if (rc1 >= 0)
2066                                 lmvea_repaired = true;
2067                 } else if (llmv->ll_inline) {
2068                         if (dt_object_remote(obj)) {
2069                                 rc1 = fld_server_lookup(env, ss->ss_server_fld,
2070                                         fid_seq(lfsck_dto2fid(obj)), range);
2071                                 if (rc1 != 0)
2072                                         goto next;
2073
2074                                 /* The slave LMV EA on the remote shard is
2075                                  * correct, just notify the LFSCK instance
2076                                  * on such MDT to re-verify the name_hash. */
2077                                 rc1 = lfsck_namespace_notify_lmv_remote(env,
2078                                                 com, obj, LE_SET_LMV_SLAVE,
2079                                                 LEF_RECHECK_NAME_HASH,
2080                                                 range->lsr_index);
2081                         } else {
2082                                 rc1 = lfsck_namespace_trace_update(env, com,
2083                                         cfid, LNTF_RECHECK_NAME_HASH, true);
2084                         }
2085
2086                         if (rc1 < 0)
2087                                 goto next;
2088                 }
2089
2090                 if (rename) {
2091                         if (dir == NULL) {
2092                                 dir = lfsck_assistant_object_load(env, lfsck,
2093                                                                   lso);
2094                                 if (IS_ERR(dir)) {
2095                                         rc1 = PTR_ERR(dir);
2096
2097                                         if (rc1 == -ENOENT)
2098                                                 GOTO(out, rc = 0);
2099
2100                                         dir = NULL;
2101                                         goto next;
2102                                 }
2103                         }
2104
2105                         rc1 = lfsck_namespace_repair_dirent(env, com, dir, obj,
2106                                         info->lti_tmpbuf2, lnr->lnr_name,
2107                                         lnr->lnr_type, true, false);
2108                         if (rc1 >= 0) {
2109                                 rename_repaired = true;
2110                                 if (rc1 > 0) {
2111                                         ns->ln_dirent_repaired++;
2112                                         rc1 = lfsck_namespace_trace_update(env,
2113                                                 com, cfid,
2114                                                 LNTF_RECHECK_NAME_HASH, true);
2115                                 }
2116                         }
2117
2118                         if (rc1 < 0)
2119                                 goto next;
2120                 }
2121
2122                 if (repair_linkea) {
2123                         struct lustre_handle lh = { 0 };
2124
2125                         rc1 = linkea_data_new(&ldata, &info->lti_big_buf);
2126                         if (rc1 != 0)
2127                                 goto next;
2128
2129                         if (dir == NULL) {
2130                                 dir = lfsck_assistant_object_load(env, lfsck,
2131                                                                   lso);
2132                                 if (IS_ERR(dir)) {
2133                                         rc1 = PTR_ERR(dir);
2134
2135                                         if (rc1 == -ENOENT)
2136                                                 GOTO(out, rc = 0);
2137
2138                                         dir = NULL;
2139                                         goto next;
2140                                 }
2141                         }
2142
2143                         rc1 = linkea_add_buf(&ldata, cname, lfsck_dto2fid(dir));
2144                         if (rc1 != 0)
2145                                 goto next;
2146
2147                         rc1 = lfsck_ibits_lock(env, lfsck, obj, &lh,
2148                                                MDS_INODELOCK_UPDATE |
2149                                                MDS_INODELOCK_XATTR, LCK_EX);
2150                         if (rc1 != 0)
2151                                 goto next;
2152
2153                         rc1 = lfsck_namespace_rebuild_linkea(env, com, obj,
2154                                                              &ldata);
2155                         lfsck_ibits_unlock(&lh, LCK_EX);
2156                         if (rc1 >= 0) {
2157                                 linkea_repaired = true;
2158                                 if (rc1 > 0)
2159                                         ns->ln_linkea_repaired++;
2160                         }
2161                 }
2162
2163 next:
2164                 CDEBUG(D_LFSCK, "%s: namespace LFSCK repair the shard "
2165                       "%d "DFID" of the striped directory "DFID" with "
2166                       "dangling %s/%s, rename %s/%s, llinkea %s/%s, "
2167                       "repair_lmvea %s/%s: rc = %d\n", lfsck_lfsck2name(lfsck),
2168                       i, PFID(cfid), PFID(&lnr->lnr_fid),
2169                       create ? "yes" : "no", create_repaired ? "yes" : "no",
2170                       rename ? "yes" : "no", rename_repaired ? "yes" : "no",
2171                       repair_linkea ? "yes" : "no",
2172                       linkea_repaired ? "yes" : "no",
2173                       repair_lmvea ? "yes" : "no",
2174                       lmvea_repaired ? "yes" : "no", rc1);
2175
2176                 if (obj != NULL && !IS_ERR(obj)) {
2177                         lfsck_object_put(env, obj);
2178                         obj = NULL;
2179                 }
2180
2181                 if (rc1 < 0) {
2182                         rc = rc1;
2183                         ns->ln_striped_shards_failed++;
2184                 }
2185         }
2186
2187         GOTO(out, rc);
2188
2189 out:
2190         if (obj != NULL && !IS_ERR(obj))
2191                 lfsck_object_put(env, obj);
2192
2193         if (dir != NULL && !IS_ERR(dir))
2194                 lfsck_object_put(env, dir);
2195
2196         return rc;
2197 }
2198
2199 /**
2200  * Verify the shard's name entry under the striped directory.
2201  *
2202  * Before all shards of the striped directory scanned, the LFSCK cannot
2203  * know whether the master LMV EA is valid or not, and also cannot know
2204  * how to repair an invalid shard exactly. For example, the stripe index
2205  * stored in the shard's name does not match the stripe index stored in
2206  * the slave LMV EA, then the LFSCK cannot know which one is correct.
2207  * If the LFSCK just assumed one is correct, and fixed the other, then
2208  * as the LFSCK processing, it may find that the former reparation is
2209  * wrong and have to roll back. Unfortunately, if some applications saw
2210  * the changes and made further modification based on such changes, then
2211  * the roll back is almost impossible.
2212  *
2213  * To avoid above trouble, the LFSCK will scan the master object of the
2214  * striped directory twice, that is NOT the same as normal two-stages
2215  * scanning, the double scanning the striped directory will happen both
2216  * during the first-stage scanning:
2217  *
2218  * 1) When the striped directory is opened for scanning, the LFSCK will
2219  *    iterate each shard in turn, and records its slave LMV EA in the
2220  *    lfsck_lmv::ll_lslr. In this step, if the 'shard' (may be fake
2221  *    shard) name does not match the shard naming rule, for example, it
2222  *    does not contains the shard's FID, or not contains index, then we
2223  *    can remove the bad name entry directly. But if the name is valid,
2224  *    but the shard has no slave LMV EA or the slave LMV EA does not
2225  *    match its name, then we just record related information in the
2226  *    lfsck_lmv::ll_lslr in RAM.
2227  *
2228  * 2) When all the known shards have been scanned, then the engine will
2229  *    generate a dummy request (via lfsck_namespace_close_dir) to tell
2230  *    the assistant thread that all the known shards have been scanned.
2231  *    Since the assistant has got the global knowledge about the index
2232  *    conflict, stripe count, hash type, and so on. Then the assistant
2233  *    thread will scan the lfsck_lmv::ll_lslr, and for every shard in
2234  *    the record, check and repair inconsistency.
2235  *
2236  * Generally, the stripe directory has only several shards, and there
2237  * will NOT be a lof of striped directory. So double scanning striped
2238  * directory will not much affect the LFSCK performance.
2239  *
2240  * \param[in] env       pointer to the thread context
2241  * \param[in] com       pointer to the lfsck component
2242  * \param[in] lnr       pointer to the namespace request that contains the
2243  *                      shard's name, parent object, parent's LMV, and ect.
2244  *
2245  * \retval              zero for success
2246  * \retval              negative error number on failure
2247  */
2248 int lfsck_namespace_handle_striped_master(const struct lu_env *env,
2249                                           struct lfsck_component *com,
2250                                           struct lfsck_namespace_req *lnr)
2251 {
2252         struct lfsck_thread_info   *info        = lfsck_env_info(env);
2253         struct lmv_mds_md_v1       *lmv         = &info->lti_lmv;
2254         struct lfsck_instance      *lfsck       = com->lc_lfsck;
2255         struct lfsck_namespace     *ns          = com->lc_file_ram;
2256         struct lfsck_lmv           *llmv        = lnr->lnr_lmv;
2257         struct lfsck_assistant_object *lso      = lnr->lnr_lar.lar_parent;
2258         const struct lu_fid        *pfid        = &lso->lso_fid;
2259         struct dt_object           *dir;
2260         struct dt_object           *obj         = NULL;
2261         struct dt_device           *dev         = NULL;
2262         int                         shard_idx   = 0;
2263         int                         stripe      = 0;
2264         int                         rc          = 0;
2265         int                         depth       = 0;
2266         bool                        repaired    = false;
2267         enum lfsck_namespace_inconsistency_type type = LNIT_NONE;
2268         ENTRY;
2269
2270         if (unlikely(llmv->ll_ignore))
2271                 RETURN(0);
2272
2273         dir = lfsck_assistant_object_load(env, lfsck, lso);
2274         if (IS_ERR(dir)) {
2275                 rc = PTR_ERR(dir);
2276
2277                 RETURN(rc == -ENOENT ? 0 : rc);
2278         }
2279
2280         shard_idx = lfsck_find_mdt_idx_by_fid(env, lfsck, &lnr->lnr_fid);
2281         if (shard_idx < 0)
2282                 GOTO(fail_lmv, rc = shard_idx);
2283
2284         if (shard_idx == lfsck_dev_idx(lfsck)) {
2285                 if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0))
2286                         GOTO(out, rc = 0);
2287
2288                 dev = lfsck->li_bottom;
2289         } else {
2290                 struct lfsck_tgt_desc *ltd;
2291
2292                 /* Usually, some local filesystem consistency verification
2293                  * tools can guarantee the local namespace tree consistenct.
2294                  * So the LFSCK will only verify the remote directory. */
2295                 if (unlikely(strcmp(lnr->lnr_name, dotdot) == 0)) {
2296                         rc = lfsck_namespace_trace_update(env, com, pfid,
2297                                                 LNTF_CHECK_PARENT, true);
2298
2299                         GOTO(out, rc);
2300                 }
2301
2302                 ltd = LTD_TGT(&lfsck->li_mdt_descs, shard_idx);
2303                 if (unlikely(ltd == NULL)) {
2304                         CDEBUG(D_LFSCK, "%s: cannot talk with MDT %x which "
2305                                "did not join the namespace LFSCK\n",
2306                                lfsck_lfsck2name(lfsck), shard_idx);
2307                         lfsck_lad_set_bitmap(env, com, shard_idx);
2308
2309                         GOTO(fail_lmv, rc = -ENODEV);
2310                 }
2311
2312                 dev = ltd->ltd_tgt;
2313         }
2314
2315         obj = lfsck_object_find_by_dev_nowait(env, dev, &lnr->lnr_fid);
2316         if (IS_ERR(obj)) {
2317                 if (lfsck_is_dead_obj(dir))
2318                         RETURN(0);
2319
2320                 GOTO(fail_lmv, rc = PTR_ERR(obj));
2321         }
2322
2323         if (!dt_object_exists(obj)) {
2324                 stripe = lfsck_shard_name_to_index(env, lnr->lnr_name,
2325                                 lnr->lnr_namelen, lnr->lnr_type, &lnr->lnr_fid);
2326                 if (stripe < 0) {
2327                         type = LNIT_BAD_DIRENT;
2328
2329                         GOTO(out, rc = 0);
2330                 }
2331
2332 dangling:
2333                 rc = lfsck_namespace_check_exist(env, dir, obj, lnr->lnr_name);
2334                 if (rc == 0) {
2335                         memset(lmv, 0, sizeof(*lmv));
2336                         lmv->lmv_magic = LMV_MAGIC;
2337                         rc = lfsck_record_lmv(env, com, dir, lnr, lmv, stripe,
2338                                               LSLF_DANGLING, LSLF_NONE, &depth);
2339                 }
2340
2341                 GOTO(out, rc);
2342         }
2343
2344         stripe = lfsck_shard_name_to_index(env, lnr->lnr_name, lnr->lnr_namelen,
2345                                            lfsck_object_type(obj),
2346                                            &lnr->lnr_fid);
2347         if (stripe < 0) {
2348                 type = LNIT_BAD_DIRENT;
2349
2350                 GOTO(out, rc = 0);
2351         }
2352
2353         rc = lfsck_read_stripe_lmv(env, obj, lmv);
2354         if (unlikely(rc == -ENOENT))
2355                 /* It may happen when the remote object has been removed,
2356                  * but the local MDT does not aware of that. */
2357                 goto dangling;
2358
2359         if (rc == -ENODATA)
2360                 rc = lfsck_record_lmv(env, com, dir, lnr, lmv, stripe,
2361                                       LSLF_NO_LMVEA, LSLF_NONE, &depth);
2362         else if (rc == 0)
2363                 rc = lfsck_record_lmv(env, com, dir, lnr, lmv, stripe,
2364                                       lmv->lmv_master_mdt_index != stripe ?
2365                                       LSLF_BAD_INDEX1 : LSLF_NONE, LSLF_NONE,
2366                                       &depth);
2367
2368         GOTO(out, rc);
2369
2370 fail_lmv:
2371         llmv->ll_failed = 1;
2372
2373 out:
2374         if (rc >= 0 && type == LNIT_NONE && !S_ISDIR(lnr->lnr_type))
2375                 type = LNIT_BAD_TYPE;
2376
2377         switch (type) {
2378         case LNIT_BAD_TYPE:
2379                 rc = lfsck_namespace_repair_dirent(env, com, dir, obj,
2380                                                    lnr->lnr_name, lnr->lnr_name,
2381                                                    lnr->lnr_type, true, false);
2382                 if (rc > 0)
2383                         repaired = true;
2384                 break;
2385         case LNIT_BAD_DIRENT:
2386                 rc = lfsck_namespace_repair_dirent(env, com, dir, obj,
2387                                                    lnr->lnr_name, lnr->lnr_name,
2388                                                    lnr->lnr_type, false, false);
2389                 if (rc > 0)
2390                         repaired = true;
2391                 break;
2392         default:
2393                 break;
2394         }
2395
2396         if (rc < 0) {
2397                 CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant fail to handle "
2398                        "the shard: "DFID", parent "DFID", name %.*s: rc = %d\n",
2399                        lfsck_lfsck2name(lfsck), PFID(&lnr->lnr_fid),
2400                        PFID(pfid), lnr->lnr_namelen, lnr->lnr_name, rc);
2401
2402                 if ((rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -EREMCHG ||
2403                      rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
2404                      rc == -EHOSTUNREACH || rc == -EINPROGRESS) &&
2405                     dev != NULL && dev != lfsck->li_bottom)
2406                         lfsck_lad_set_bitmap(env, com, shard_idx);
2407
2408                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT))
2409                         rc = 0;
2410         } else {
2411                 if (repaired) {
2412                         ns->ln_items_repaired++;
2413
2414                         switch (type) {
2415                         case LNIT_BAD_TYPE:
2416                                 ns->ln_bad_type_repaired++;
2417                                 break;
2418                         case LNIT_BAD_DIRENT:
2419                                 ns->ln_dirent_repaired++;
2420                                 break;
2421                         default:
2422                                 break;
2423                         }
2424                 }
2425
2426                 rc = 0;
2427         }
2428
2429         if (obj != NULL && !IS_ERR(obj))
2430                 lfsck_object_put(env, obj);
2431
2432         lfsck_object_put(env, dir);
2433
2434         return rc;
2435 }