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