Whamcloud - gitweb
LU-9019 lfsck: migrate to 64 bit time
[fs/lustre-release.git] / lustre / lfsck / lfsck_namespace.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) 2013, 2016, Intel Corporation.
24  */
25 /*
26  * lustre/lfsck/lfsck_namespace.c
27  *
28  * Author: Fan, Yong <fan.yong@intel.com>
29  */
30
31 #define DEBUG_SUBSYSTEM S_LFSCK
32
33 #include <lustre/lustre_idl.h>
34 #include <lu_object.h>
35 #include <dt_object.h>
36 #include <md_object.h>
37 #include <lustre_fid.h>
38 #include <lustre_lib.h>
39 #include <lustre_net.h>
40 #include <lustre/lustre_user.h>
41
42 #include "lfsck_internal.h"
43
44 #define LFSCK_NAMESPACE_MAGIC_V1        0xA0629D03
45 #define LFSCK_NAMESPACE_MAGIC_V2        0xA0621A0B
46 #define LFSCK_NAMESPACE_MAGIC_V3        0xA06249FF
47
48 /* For Lustre-2.x (x <= 6), the namespace LFSCK used LFSCK_NAMESPACE_MAGIC_V1
49  * as the trace file magic. When downgrade to such old release, the old LFSCK
50  * will not recognize the new LFSCK_NAMESPACE_MAGIC_V2 in the new trace file,
51  * then it will reset the whole LFSCK, and will not cause start failure. The
52  * similar case will happen when upgrade from such old release. */
53 #define LFSCK_NAMESPACE_MAGIC           LFSCK_NAMESPACE_MAGIC_V3
54
55 enum lfsck_nameentry_check {
56         LFSCK_NAMEENTRY_DEAD            = 1, /* The object has been unlinked. */
57         LFSCK_NAMEENTRY_REMOVED         = 2, /* The entry has been removed. */
58         LFSCK_NAMEENTRY_RECREATED       = 3, /* The entry has been recreated. */
59 };
60
61 static struct lfsck_namespace_req *
62 lfsck_namespace_assistant_req_init(struct lfsck_instance *lfsck,
63                                    struct lfsck_assistant_object *lso,
64                                    struct lu_dirent *ent, __u16 type)
65 {
66         struct lfsck_namespace_req *lnr;
67         int                         size;
68
69         size = sizeof(*lnr) + (ent->lde_namelen & ~3) + 4;
70         OBD_ALLOC(lnr, size);
71         if (lnr == NULL)
72                 return ERR_PTR(-ENOMEM);
73
74         INIT_LIST_HEAD(&lnr->lnr_lar.lar_list);
75         lnr->lnr_lar.lar_parent = lfsck_assistant_object_get(lso);
76         lnr->lnr_lmv = lfsck_lmv_get(lfsck->li_lmv);
77         lnr->lnr_fid = ent->lde_fid;
78         lnr->lnr_dir_cookie = ent->lde_hash;
79         lnr->lnr_attr = ent->lde_attrs;
80         lnr->lnr_size = size;
81         lnr->lnr_type = type;
82         lnr->lnr_namelen = ent->lde_namelen;
83         memcpy(lnr->lnr_name, ent->lde_name, ent->lde_namelen);
84
85         return lnr;
86 }
87
88 static void lfsck_namespace_assistant_req_fini(const struct lu_env *env,
89                                                struct lfsck_assistant_req *lar)
90 {
91         struct lfsck_namespace_req *lnr =
92                         container_of0(lar, struct lfsck_namespace_req, lnr_lar);
93
94         if (lnr->lnr_lmv != NULL)
95                 lfsck_lmv_put(env, lnr->lnr_lmv);
96
97         lfsck_assistant_object_put(env, lar->lar_parent);
98         OBD_FREE(lnr, lnr->lnr_size);
99 }
100
101 static void lfsck_namespace_le_to_cpu(struct lfsck_namespace *dst,
102                                       struct lfsck_namespace *src)
103 {
104         dst->ln_magic = le32_to_cpu(src->ln_magic);
105         dst->ln_status = le32_to_cpu(src->ln_status);
106         dst->ln_flags = le32_to_cpu(src->ln_flags);
107         dst->ln_success_count = le32_to_cpu(src->ln_success_count);
108         dst->ln_run_time_phase1 = le64_to_cpu(src->ln_run_time_phase1);
109         dst->ln_run_time_phase2 = le64_to_cpu(src->ln_run_time_phase2);
110         dst->ln_time_last_complete = le64_to_cpu(src->ln_time_last_complete);
111         dst->ln_time_latest_start = le64_to_cpu(src->ln_time_latest_start);
112         dst->ln_time_last_checkpoint =
113                                 le64_to_cpu(src->ln_time_last_checkpoint);
114         lfsck_position_le_to_cpu(&dst->ln_pos_latest_start,
115                                  &src->ln_pos_latest_start);
116         lfsck_position_le_to_cpu(&dst->ln_pos_last_checkpoint,
117                                  &src->ln_pos_last_checkpoint);
118         lfsck_position_le_to_cpu(&dst->ln_pos_first_inconsistent,
119                                  &src->ln_pos_first_inconsistent);
120         dst->ln_items_checked = le64_to_cpu(src->ln_items_checked);
121         dst->ln_items_repaired = le64_to_cpu(src->ln_items_repaired);
122         dst->ln_items_failed = le64_to_cpu(src->ln_items_failed);
123         dst->ln_dirs_checked = le64_to_cpu(src->ln_dirs_checked);
124         dst->ln_objs_checked_phase2 = le64_to_cpu(src->ln_objs_checked_phase2);
125         dst->ln_objs_repaired_phase2 =
126                                 le64_to_cpu(src->ln_objs_repaired_phase2);
127         dst->ln_objs_failed_phase2 = le64_to_cpu(src->ln_objs_failed_phase2);
128         dst->ln_objs_nlink_repaired = le64_to_cpu(src->ln_objs_nlink_repaired);
129         fid_le_to_cpu(&dst->ln_fid_latest_scanned_phase2,
130                       &src->ln_fid_latest_scanned_phase2);
131         dst->ln_dirent_repaired = le64_to_cpu(src->ln_dirent_repaired);
132         dst->ln_linkea_repaired = le64_to_cpu(src->ln_linkea_repaired);
133         dst->ln_mul_linked_checked = le64_to_cpu(src->ln_mul_linked_checked);
134         dst->ln_mul_linked_repaired = le64_to_cpu(src->ln_mul_linked_repaired);
135         dst->ln_unknown_inconsistency =
136                                 le64_to_cpu(src->ln_unknown_inconsistency);
137         dst->ln_unmatched_pairs_repaired =
138                                 le64_to_cpu(src->ln_unmatched_pairs_repaired);
139         dst->ln_dangling_repaired = le64_to_cpu(src->ln_dangling_repaired);
140         dst->ln_mul_ref_repaired = le64_to_cpu(src->ln_mul_ref_repaired);
141         dst->ln_bad_type_repaired = le64_to_cpu(src->ln_bad_type_repaired);
142         dst->ln_lost_dirent_repaired =
143                                 le64_to_cpu(src->ln_lost_dirent_repaired);
144         dst->ln_striped_dirs_scanned =
145                                 le64_to_cpu(src->ln_striped_dirs_scanned);
146         dst->ln_striped_dirs_repaired =
147                                 le64_to_cpu(src->ln_striped_dirs_repaired);
148         dst->ln_striped_dirs_failed =
149                                 le64_to_cpu(src->ln_striped_dirs_failed);
150         dst->ln_striped_dirs_disabled =
151                                 le64_to_cpu(src->ln_striped_dirs_disabled);
152         dst->ln_striped_dirs_skipped =
153                                 le64_to_cpu(src->ln_striped_dirs_skipped);
154         dst->ln_striped_shards_scanned =
155                                 le64_to_cpu(src->ln_striped_shards_scanned);
156         dst->ln_striped_shards_repaired =
157                                 le64_to_cpu(src->ln_striped_shards_repaired);
158         dst->ln_striped_shards_failed =
159                                 le64_to_cpu(src->ln_striped_shards_failed);
160         dst->ln_striped_shards_skipped =
161                                 le64_to_cpu(src->ln_striped_shards_skipped);
162         dst->ln_name_hash_repaired = le64_to_cpu(src->ln_name_hash_repaired);
163         dst->ln_local_lpf_scanned = le64_to_cpu(src->ln_local_lpf_scanned);
164         dst->ln_local_lpf_moved = le64_to_cpu(src->ln_local_lpf_moved);
165         dst->ln_local_lpf_skipped = le64_to_cpu(src->ln_local_lpf_skipped);
166         dst->ln_local_lpf_failed = le64_to_cpu(src->ln_local_lpf_failed);
167         dst->ln_bitmap_size = le32_to_cpu(src->ln_bitmap_size);
168         dst->ln_time_latest_reset = le64_to_cpu(src->ln_time_latest_reset);
169         dst->ln_linkea_overflow_cleared =
170                                 le64_to_cpu(src->ln_linkea_overflow_cleared);
171 }
172
173 static void lfsck_namespace_cpu_to_le(struct lfsck_namespace *dst,
174                                       struct lfsck_namespace *src)
175 {
176         dst->ln_magic = cpu_to_le32(src->ln_magic);
177         dst->ln_status = cpu_to_le32(src->ln_status);
178         dst->ln_flags = cpu_to_le32(src->ln_flags);
179         dst->ln_success_count = cpu_to_le32(src->ln_success_count);
180         dst->ln_run_time_phase1 = cpu_to_le64(src->ln_run_time_phase1);
181         dst->ln_run_time_phase2 = cpu_to_le64(src->ln_run_time_phase2);
182         dst->ln_time_last_complete = cpu_to_le64(src->ln_time_last_complete);
183         dst->ln_time_latest_start = cpu_to_le64(src->ln_time_latest_start);
184         dst->ln_time_last_checkpoint =
185                                 cpu_to_le64(src->ln_time_last_checkpoint);
186         lfsck_position_cpu_to_le(&dst->ln_pos_latest_start,
187                                  &src->ln_pos_latest_start);
188         lfsck_position_cpu_to_le(&dst->ln_pos_last_checkpoint,
189                                  &src->ln_pos_last_checkpoint);
190         lfsck_position_cpu_to_le(&dst->ln_pos_first_inconsistent,
191                                  &src->ln_pos_first_inconsistent);
192         dst->ln_items_checked = cpu_to_le64(src->ln_items_checked);
193         dst->ln_items_repaired = cpu_to_le64(src->ln_items_repaired);
194         dst->ln_items_failed = cpu_to_le64(src->ln_items_failed);
195         dst->ln_dirs_checked = cpu_to_le64(src->ln_dirs_checked);
196         dst->ln_objs_checked_phase2 = cpu_to_le64(src->ln_objs_checked_phase2);
197         dst->ln_objs_repaired_phase2 =
198                                 cpu_to_le64(src->ln_objs_repaired_phase2);
199         dst->ln_objs_failed_phase2 = cpu_to_le64(src->ln_objs_failed_phase2);
200         dst->ln_objs_nlink_repaired = cpu_to_le64(src->ln_objs_nlink_repaired);
201         fid_cpu_to_le(&dst->ln_fid_latest_scanned_phase2,
202                       &src->ln_fid_latest_scanned_phase2);
203         dst->ln_dirent_repaired = cpu_to_le64(src->ln_dirent_repaired);
204         dst->ln_linkea_repaired = cpu_to_le64(src->ln_linkea_repaired);
205         dst->ln_mul_linked_checked = cpu_to_le64(src->ln_mul_linked_checked);
206         dst->ln_mul_linked_repaired = cpu_to_le64(src->ln_mul_linked_repaired);
207         dst->ln_unknown_inconsistency =
208                                 cpu_to_le64(src->ln_unknown_inconsistency);
209         dst->ln_unmatched_pairs_repaired =
210                                 cpu_to_le64(src->ln_unmatched_pairs_repaired);
211         dst->ln_dangling_repaired = cpu_to_le64(src->ln_dangling_repaired);
212         dst->ln_mul_ref_repaired = cpu_to_le64(src->ln_mul_ref_repaired);
213         dst->ln_bad_type_repaired = cpu_to_le64(src->ln_bad_type_repaired);
214         dst->ln_lost_dirent_repaired =
215                                 cpu_to_le64(src->ln_lost_dirent_repaired);
216         dst->ln_striped_dirs_scanned =
217                                 cpu_to_le64(src->ln_striped_dirs_scanned);
218         dst->ln_striped_dirs_repaired =
219                                 cpu_to_le64(src->ln_striped_dirs_repaired);
220         dst->ln_striped_dirs_failed =
221                                 cpu_to_le64(src->ln_striped_dirs_failed);
222         dst->ln_striped_dirs_disabled =
223                                 cpu_to_le64(src->ln_striped_dirs_disabled);
224         dst->ln_striped_dirs_skipped =
225                                 cpu_to_le64(src->ln_striped_dirs_skipped);
226         dst->ln_striped_shards_scanned =
227                                 cpu_to_le64(src->ln_striped_shards_scanned);
228         dst->ln_striped_shards_repaired =
229                                 cpu_to_le64(src->ln_striped_shards_repaired);
230         dst->ln_striped_shards_failed =
231                                 cpu_to_le64(src->ln_striped_shards_failed);
232         dst->ln_striped_shards_skipped =
233                                 cpu_to_le64(src->ln_striped_shards_skipped);
234         dst->ln_name_hash_repaired = cpu_to_le64(src->ln_name_hash_repaired);
235         dst->ln_local_lpf_scanned = cpu_to_le64(src->ln_local_lpf_scanned);
236         dst->ln_local_lpf_moved = cpu_to_le64(src->ln_local_lpf_moved);
237         dst->ln_local_lpf_skipped = cpu_to_le64(src->ln_local_lpf_skipped);
238         dst->ln_local_lpf_failed = cpu_to_le64(src->ln_local_lpf_failed);
239         dst->ln_bitmap_size = cpu_to_le32(src->ln_bitmap_size);
240         dst->ln_time_latest_reset = cpu_to_le64(src->ln_time_latest_reset);
241         dst->ln_linkea_overflow_cleared =
242                                 cpu_to_le64(src->ln_linkea_overflow_cleared);
243 }
244
245 static void lfsck_namespace_record_failure(const struct lu_env *env,
246                                            struct lfsck_instance *lfsck,
247                                            struct lfsck_namespace *ns)
248 {
249         struct lfsck_position pos;
250
251         ns->ln_items_failed++;
252         lfsck_pos_fill(env, lfsck, &pos, false);
253         if (lfsck_pos_is_zero(&ns->ln_pos_first_inconsistent) ||
254             lfsck_pos_is_eq(&pos, &ns->ln_pos_first_inconsistent) < 0) {
255                 ns->ln_pos_first_inconsistent = pos;
256
257                 CDEBUG(D_LFSCK, "%s: namespace LFSCK hit first non-repaired "
258                        "inconsistency at the pos [%llu, "DFID", %#llx]\n",
259                        lfsck_lfsck2name(lfsck),
260                        ns->ln_pos_first_inconsistent.lp_oit_cookie,
261                        PFID(&ns->ln_pos_first_inconsistent.lp_dir_parent),
262                        ns->ln_pos_first_inconsistent.lp_dir_cookie);
263         }
264 }
265
266 /**
267  * Load the MDT bitmap from the lfsck_namespace trace file.
268  *
269  * \param[in] env       pointer to the thread context
270  * \param[in] com       pointer to the lfsck component
271  *
272  * \retval              0 for success
273  * \retval              negative error number on failure or data corruption
274  */
275 static int lfsck_namespace_load_bitmap(const struct lu_env *env,
276                                        struct lfsck_component *com)
277 {
278         struct dt_object                *obj    = com->lc_obj;
279         struct lfsck_assistant_data     *lad    = com->lc_data;
280         struct lfsck_namespace          *ns     = com->lc_file_ram;
281         struct cfs_bitmap                       *bitmap = lad->lad_bitmap;
282         ssize_t                          size;
283         __u32                            nbits;
284         int                              rc;
285         ENTRY;
286
287         if (com->lc_lfsck->li_mdt_descs.ltd_tgts_bitmap->size >
288             ns->ln_bitmap_size)
289                 nbits = com->lc_lfsck->li_mdt_descs.ltd_tgts_bitmap->size;
290         else
291                 nbits = ns->ln_bitmap_size;
292
293         if (unlikely(nbits < BITS_PER_LONG))
294                 nbits = BITS_PER_LONG;
295
296         if (nbits > bitmap->size) {
297                 __u32 new_bits = bitmap->size;
298                 struct cfs_bitmap *new_bitmap;
299
300                 while (new_bits < nbits)
301                         new_bits <<= 1;
302
303                 new_bitmap = CFS_ALLOCATE_BITMAP(new_bits);
304                 if (new_bitmap == NULL)
305                         RETURN(-ENOMEM);
306
307                 lad->lad_bitmap = new_bitmap;
308                 CFS_FREE_BITMAP(bitmap);
309                 bitmap = new_bitmap;
310         }
311
312         if (ns->ln_bitmap_size == 0) {
313                 lad->lad_incomplete = 0;
314                 CFS_RESET_BITMAP(bitmap);
315
316                 RETURN(0);
317         }
318
319         size = (ns->ln_bitmap_size + 7) >> 3;
320         rc = dt_xattr_get(env, obj,
321                           lfsck_buf_get(env, bitmap->data, size),
322                           XATTR_NAME_LFSCK_BITMAP);
323         if (rc != size)
324                 RETURN(rc >= 0 ? -EINVAL : rc);
325
326         if (cfs_bitmap_check_empty(bitmap))
327                 lad->lad_incomplete = 0;
328         else
329                 lad->lad_incomplete = 1;
330
331         RETURN(0);
332 }
333
334 /**
335  * Load namespace LFSCK statistics information from the trace file.
336  *
337  * \param[in] env       pointer to the thread context
338  * \param[in] com       pointer to the lfsck component
339  *
340  * \retval              0 for success
341  * \retval              negative error number on failure
342  */
343 static int lfsck_namespace_load(const struct lu_env *env,
344                                 struct lfsck_component *com)
345 {
346         int len = com->lc_file_size;
347         int rc;
348
349         rc = dt_xattr_get(env, com->lc_obj,
350                           lfsck_buf_get(env, com->lc_file_disk, len),
351                           XATTR_NAME_LFSCK_NAMESPACE);
352         if (rc == len) {
353                 struct lfsck_namespace *ns = com->lc_file_ram;
354
355                 lfsck_namespace_le_to_cpu(ns,
356                                 (struct lfsck_namespace *)com->lc_file_disk);
357                 if (ns->ln_magic != LFSCK_NAMESPACE_MAGIC) {
358                         CDEBUG(D_LFSCK, "%s: invalid lfsck_namespace magic "
359                                "%#x != %#x\n", lfsck_lfsck2name(com->lc_lfsck),
360                                ns->ln_magic, LFSCK_NAMESPACE_MAGIC);
361                         rc = -ESTALE;
362                 } else {
363                         rc = 0;
364                 }
365         } else if (rc != -ENODATA) {
366                 CDEBUG(D_LFSCK, "%s: fail to load lfsck_namespace, "
367                        "expected = %d: rc = %d\n",
368                        lfsck_lfsck2name(com->lc_lfsck), len, rc);
369                 if (rc >= 0)
370                         rc = -ESTALE;
371         }
372
373         return rc;
374 }
375
376 static int lfsck_namespace_store(const struct lu_env *env,
377                                  struct lfsck_component *com)
378 {
379         struct dt_object                *obj    = com->lc_obj;
380         struct lfsck_instance           *lfsck  = com->lc_lfsck;
381         struct lfsck_namespace          *ns     = com->lc_file_ram;
382         struct lfsck_assistant_data     *lad    = com->lc_data;
383         struct dt_device                *dev    = lfsck_obj2dev(obj);
384         struct cfs_bitmap               *bitmap = NULL;
385         struct thandle                  *handle;
386         __u32                            nbits  = 0;
387         int                              len    = com->lc_file_size;
388         int                              rc;
389         ENTRY;
390
391         if (lad != NULL) {
392                 bitmap = lad->lad_bitmap;
393                 nbits = bitmap->size;
394
395                 LASSERT(nbits > 0);
396                 LASSERTF((nbits & 7) == 0, "Invalid nbits %u\n", nbits);
397         }
398
399         ns->ln_bitmap_size = nbits;
400         lfsck_namespace_cpu_to_le((struct lfsck_namespace *)com->lc_file_disk,
401                                   ns);
402         handle = dt_trans_create(env, dev);
403         if (IS_ERR(handle))
404                 GOTO(log, rc = PTR_ERR(handle));
405
406         rc = dt_declare_xattr_set(env, obj,
407                                   lfsck_buf_get(env, com->lc_file_disk, len),
408                                   XATTR_NAME_LFSCK_NAMESPACE, 0, handle);
409         if (rc != 0)
410                 GOTO(out, rc);
411
412         if (bitmap != NULL) {
413                 rc = dt_declare_xattr_set(env, obj,
414                                 lfsck_buf_get(env, bitmap->data, nbits >> 3),
415                                 XATTR_NAME_LFSCK_BITMAP, 0, handle);
416                 if (rc != 0)
417                         GOTO(out, rc);
418         }
419
420         rc = dt_trans_start_local(env, dev, handle);
421         if (rc != 0)
422                 GOTO(out, rc);
423
424         rc = dt_xattr_set(env, obj,
425                           lfsck_buf_get(env, com->lc_file_disk, len),
426                           XATTR_NAME_LFSCK_NAMESPACE, 0, handle);
427         if (rc == 0 && bitmap != NULL)
428                 rc = dt_xattr_set(env, obj,
429                                   lfsck_buf_get(env, bitmap->data, nbits >> 3),
430                                   XATTR_NAME_LFSCK_BITMAP, 0, handle);
431
432         GOTO(out, rc);
433
434 out:
435         dt_trans_stop(env, dev, handle);
436
437 log:
438         if (rc != 0)
439                 CDEBUG(D_LFSCK, "%s: fail to store lfsck_namespace: rc = %d\n",
440                        lfsck_lfsck2name(lfsck), rc);
441         return rc;
442 }
443
444 static int lfsck_namespace_init(const struct lu_env *env,
445                                 struct lfsck_component *com)
446 {
447         struct lfsck_namespace *ns = com->lc_file_ram;
448         int rc;
449
450         memset(ns, 0, sizeof(*ns));
451         ns->ln_magic = LFSCK_NAMESPACE_MAGIC;
452         ns->ln_status = LS_INIT;
453         ns->ln_time_latest_reset = ktime_get_real_seconds();
454         down_write(&com->lc_sem);
455         rc = lfsck_namespace_store(env, com);
456         if (rc == 0)
457                 rc = lfsck_load_sub_trace_files(env, com,
458                         &dt_lfsck_namespace_features, LFSCK_NAMESPACE, true);
459         up_write(&com->lc_sem);
460
461         return rc;
462 }
463
464 /**
465  * Update the namespace LFSCK trace file for the given @fid
466  *
467  * \param[in] env       pointer to the thread context
468  * \param[in] com       pointer to the lfsck component
469  * \param[in] fid       the fid which flags to be updated in the lfsck
470  *                      trace file
471  * \param[in] add       true if add new flags, otherwise remove flags
472  *
473  * \retval              0 for success or nothing to be done
474  * \retval              negative error number on failure
475  */
476 int lfsck_namespace_trace_update(const struct lu_env *env,
477                                  struct lfsck_component *com,
478                                  const struct lu_fid *fid,
479                                  const __u8 flags, bool add)
480 {
481         struct lfsck_instance   *lfsck  = com->lc_lfsck;
482         struct dt_object        *obj;
483         struct lu_fid           *key    = &lfsck_env_info(env)->lti_fid3;
484         struct dt_device        *dev;
485         struct thandle          *th     = NULL;
486         int                      idx;
487         int                      rc     = 0;
488         __u8                     old    = 0;
489         __u8                     new    = 0;
490         ENTRY;
491
492         LASSERT(flags != 0);
493
494         if (unlikely(!fid_is_sane(fid)))
495                 RETURN(0);
496
497         idx = lfsck_sub_trace_file_fid2idx(fid);
498         mutex_lock(&com->lc_sub_trace_objs[idx].lsto_mutex);
499         obj = com->lc_sub_trace_objs[idx].lsto_obj;
500         if (unlikely(obj == NULL)) {
501                 mutex_unlock(&com->lc_sub_trace_objs[idx].lsto_mutex);
502                 RETURN(0);
503         }
504
505         lfsck_object_get(obj);
506         dev = lfsck_obj2dev(obj);
507         fid_cpu_to_be(key, fid);
508         rc = dt_lookup(env, obj, (struct dt_rec *)&old,
509                        (const struct dt_key *)key);
510         if (rc == -ENOENT) {
511                 if (!add)
512                         GOTO(unlock, rc = 0);
513
514                 old = 0;
515                 new = flags;
516         } else if (rc == 0) {
517                 if (add) {
518                         if ((old & flags) == flags)
519                                 GOTO(unlock, rc = 0);
520
521                         new = old | flags;
522                 } else {
523                         if ((old & flags) == 0)
524                                 GOTO(unlock, rc = 0);
525
526                         new = old & ~flags;
527                 }
528         } else {
529                 GOTO(log, rc);
530         }
531
532         th = dt_trans_create(env, dev);
533         if (IS_ERR(th))
534                 GOTO(log, rc = PTR_ERR(th));
535
536         if (old != 0) {
537                 rc = dt_declare_delete(env, obj,
538                                        (const struct dt_key *)key, th);
539                 if (rc != 0)
540                         GOTO(log, rc);
541         }
542
543         if (new != 0) {
544                 rc = dt_declare_insert(env, obj,
545                                        (const struct dt_rec *)&new,
546                                        (const struct dt_key *)key, th);
547                 if (rc != 0)
548                         GOTO(log, rc);
549         }
550
551         rc = dt_trans_start_local(env, dev, th);
552         if (rc != 0)
553                 GOTO(log, rc);
554
555         if (old != 0) {
556                 rc = dt_delete(env, obj, (const struct dt_key *)key, th);
557                 if (rc != 0)
558                         GOTO(log, rc);
559         }
560
561         if (new != 0) {
562                 rc = dt_insert(env, obj, (const struct dt_rec *)&new,
563                                (const struct dt_key *)key, th, 1);
564                 if (rc != 0)
565                         GOTO(log, rc);
566         }
567
568         GOTO(log, rc);
569
570 log:
571         if (th != NULL && !IS_ERR(th))
572                 dt_trans_stop(env, dev, th);
573
574         CDEBUG(D_LFSCK, "%s: namespace LFSCK %s flags for "DFID" in the "
575                "trace file, flags %x, old %x, new %x: rc = %d\n",
576                lfsck_lfsck2name(lfsck), add ? "add" : "del", PFID(fid),
577                (__u32)flags, (__u32)old, (__u32)new, rc);
578
579 unlock:
580         mutex_unlock(&com->lc_sub_trace_objs[idx].lsto_mutex);
581         lfsck_object_put(env, obj);
582
583         return rc;
584 }
585
586 int lfsck_namespace_check_exist(const struct lu_env *env,
587                                 struct dt_object *dir,
588                                 struct dt_object *obj, const char *name)
589 {
590         struct lu_fid    *fid = &lfsck_env_info(env)->lti_fid;
591         int               rc;
592         ENTRY;
593
594         if (unlikely(lfsck_is_dead_obj(obj)))
595                 RETURN(LFSCK_NAMEENTRY_DEAD);
596
597         rc = dt_lookup(env, dir, (struct dt_rec *)fid,
598                        (const struct dt_key *)name);
599         if (rc == -ENOENT)
600                 RETURN(LFSCK_NAMEENTRY_REMOVED);
601
602         if (rc < 0)
603                 RETURN(rc);
604
605         if (!lu_fid_eq(fid, lfsck_dto2fid(obj)))
606                 RETURN(LFSCK_NAMEENTRY_RECREATED);
607
608         RETURN(0);
609 }
610
611 static int lfsck_declare_namespace_exec_dir(const struct lu_env *env,
612                                             struct dt_object *obj,
613                                             struct thandle *handle)
614 {
615         int rc;
616
617         /* For remote updating LINKEA, there may be further LFSCK action
618          * on remote MDT after the updating, so update the LINKEA ASAP. */
619         if (dt_object_remote(obj))
620                 handle->th_sync = 1;
621
622         /* For destroying all invalid linkEA entries. */
623         rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, handle);
624         if (rc == 0)
625                 /* For insert new linkEA entry. */
626                 rc = dt_declare_xattr_set(env, obj,
627                         lfsck_buf_get_const(env, NULL, MAX_LINKEA_SIZE),
628                         XATTR_NAME_LINK, 0, handle);
629         return rc;
630 }
631
632 int __lfsck_links_read(const struct lu_env *env, struct dt_object *obj,
633                        struct linkea_data *ldata, bool with_rec)
634 {
635         int rc;
636
637         if (ldata->ld_buf->lb_buf == NULL)
638                 return -ENOMEM;
639
640         if (!dt_object_exists(obj))
641                 return -ENOENT;
642
643         rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK);
644         if (rc == -ERANGE) {
645                 /* Buf was too small, figure out what we need. */
646                 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LINK);
647                 if (unlikely(rc == 0))
648                         return -ENODATA;
649
650                 if (rc < 0)
651                         return rc;
652
653                 lu_buf_realloc(ldata->ld_buf, rc);
654                 if (ldata->ld_buf->lb_buf == NULL)
655                         return -ENOMEM;
656
657                 rc = dt_xattr_get(env, obj, ldata->ld_buf, XATTR_NAME_LINK);
658         }
659
660         if (unlikely(rc == 0))
661                 return -ENODATA;
662
663         if (rc > 0) {
664                 if (with_rec)
665                         rc = linkea_init_with_rec(ldata);
666                 else
667                         rc = linkea_init(ldata);
668         }
669
670         return rc;
671 }
672
673 /**
674  * Remove linkEA for the given object.
675  *
676  * The caller should take the ldlm lock before the calling.
677  *
678  * \param[in] env       pointer to the thread context
679  * \param[in] com       pointer to the lfsck component
680  * \param[in] obj       pointer to the dt_object to be handled
681  *
682  * \retval              0 for repaired cases
683  * \retval              negative error number on failure
684  */
685 static int lfsck_namespace_links_remove(const struct lu_env *env,
686                                         struct lfsck_component *com,
687                                         struct dt_object *obj)
688 {
689         struct lfsck_instance           *lfsck  = com->lc_lfsck;
690         struct dt_device                *dev    = lfsck_obj2dev(obj);
691         struct thandle                  *th     = NULL;
692         int                              rc     = 0;
693         ENTRY;
694
695         LASSERT(dt_object_remote(obj) == 0);
696
697         th = dt_trans_create(env, dev);
698         if (IS_ERR(th))
699                 GOTO(log, rc = PTR_ERR(th));
700
701         rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, th);
702         if (rc != 0)
703                 GOTO(stop, rc);
704
705         rc = dt_trans_start_local(env, dev, th);
706         if (rc != 0)
707                 GOTO(stop, rc);
708
709         dt_write_lock(env, obj, 0);
710         if (unlikely(lfsck_is_dead_obj(obj)))
711                 GOTO(unlock, rc = -ENOENT);
712
713         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
714                 GOTO(unlock, rc = 0);
715
716         rc = dt_xattr_del(env, obj, XATTR_NAME_LINK, th);
717
718         GOTO(unlock, rc);
719
720 unlock:
721         dt_write_unlock(env, obj);
722
723 stop:
724         dt_trans_stop(env, dev, th);
725
726 log:
727         CDEBUG(D_LFSCK, "%s: namespace LFSCK remove invalid linkEA "
728                "for the object "DFID": rc = %d\n",
729                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
730
731         if (rc == 0) {
732                 struct lfsck_namespace *ns = com->lc_file_ram;
733
734                 ns->ln_flags |= LF_INCONSISTENT;
735         }
736
737         return rc;
738 }
739
740 static int lfsck_links_write(const struct lu_env *env, struct dt_object *obj,
741                              struct linkea_data *ldata, struct thandle *handle)
742 {
743         struct lu_buf buf;
744         int rc;
745
746         lfsck_buf_init(&buf, ldata->ld_buf->lb_buf, ldata->ld_leh->leh_len);
747
748 again:
749         rc = dt_xattr_set(env, obj, &buf, XATTR_NAME_LINK, 0, handle);
750         if (unlikely(rc == -ENOSPC)) {
751                 rc = linkea_overflow_shrink(ldata);
752                 if (likely(rc > 0)) {
753                         buf.lb_len = rc;
754                         goto again;
755                 }
756         }
757
758         return rc;
759 }
760
761 static int lfsck_namespace_unpack_linkea_entry(struct linkea_data *ldata,
762                                                struct lu_name *cname,
763                                                struct lu_fid *pfid,
764                                                char *buf, const int buflen)
765 {
766         linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, cname, pfid);
767         if (unlikely(ldata->ld_reclen <= 0 ||
768                      ldata->ld_reclen + sizeof(struct link_ea_header) >
769                         ldata->ld_leh->leh_len ||
770                      cname->ln_namelen <= 0 ||
771                      cname->ln_namelen > NAME_MAX ||
772                      cname->ln_namelen >= buflen ||
773                      !fid_is_sane(pfid)))
774                 return -EINVAL;
775
776         /* To guarantee the 'name' is terminated with '0'. */
777         memcpy(buf, cname->ln_name, cname->ln_namelen);
778         buf[cname->ln_namelen] = 0;
779         cname->ln_name = buf;
780
781         return 0;
782 }
783
784 static void lfsck_linkea_del_buf(struct linkea_data *ldata,
785                                  const struct lu_name *lname)
786 {
787         LASSERT(ldata->ld_leh != NULL && ldata->ld_lee != NULL);
788
789         /* If current record is corrupted, all the subsequent
790          * records will be dropped. */
791         if (unlikely(ldata->ld_reclen <= 0 ||
792                      ldata->ld_reclen + sizeof(struct link_ea_header) >
793                         ldata->ld_leh->leh_len)) {
794                 void *ptr = ldata->ld_lee;
795
796                 ldata->ld_leh->leh_len = sizeof(struct link_ea_header);
797                 ldata->ld_leh->leh_reccount = 0;
798                 linkea_first_entry(ldata);
799                 while (ldata->ld_lee != NULL &&
800                        (char *)ldata->ld_lee < (char *)ptr) {
801                         int reclen = (ldata->ld_lee->lee_reclen[0] << 8) |
802                                      ldata->ld_lee->lee_reclen[1];
803
804                         ldata->ld_leh->leh_len += reclen;
805                         ldata->ld_leh->leh_reccount++;
806                         ldata->ld_lee = (struct link_ea_entry *)
807                                         ((char *)ldata->ld_lee + reclen);
808                 }
809
810                 ldata->ld_lee = NULL;
811         } else {
812                 linkea_del_buf(ldata, lname);
813         }
814 }
815
816 static int lfsck_namespace_filter_linkea_entry(struct linkea_data *ldata,
817                                                struct lu_name *cname,
818                                                struct lu_fid *pfid,
819                                                bool remove)
820 {
821         struct link_ea_entry    *oldlee;
822         int                      oldlen;
823         int                      repeated = 0;
824
825         oldlee = ldata->ld_lee;
826         oldlen = ldata->ld_reclen;
827         linkea_next_entry(ldata);
828         while (ldata->ld_lee != NULL) {
829                 ldata->ld_reclen = (ldata->ld_lee->lee_reclen[0] << 8) |
830                                    ldata->ld_lee->lee_reclen[1];
831                 if (unlikely(ldata->ld_reclen == oldlen &&
832                              memcmp(ldata->ld_lee, oldlee, oldlen) == 0)) {
833                         repeated++;
834                         if (!remove)
835                                 break;
836
837                         lfsck_linkea_del_buf(ldata, cname);
838                 } else {
839                         linkea_next_entry(ldata);
840                 }
841         }
842         ldata->ld_lee = oldlee;
843         ldata->ld_reclen = oldlen;
844
845         return repeated;
846 }
847
848 /**
849  * Insert orphan into .lustre/lost+found/MDTxxxx/ locally.
850  *
851  * Add the specified orphan MDT-object to the .lustre/lost+found/MDTxxxx/
852  * with the given type to generate the name, the detailed rules for name
853  * have been described as following.
854  *
855  * The function also generates the linkEA corresponding to the name entry
856  * under the .lustre/lost+found/MDTxxxx/ for the orphan MDT-object.
857  *
858  * \param[in] env       pointer to the thread context
859  * \param[in] com       pointer to the lfsck component
860  * \param[in] orphan    pointer to the orphan MDT-object
861  * \param[in] infix     additional information for the orphan name, such as
862  *                      the FID for original
863  * \param[in] type      the type for describing why the orphan MDT-object is
864  *                      created. The rules are as following:
865  *
866  *  type "D":           The MDT-object is a directory, it may knows its parent
867  *                      but because there is no valid linkEA, the LFSCK cannot
868  *                      know where to put it back to the namespace.
869  *  type "O":           The MDT-object has no linkEA, and there is no name
870  *                      entry that references the MDT-object.
871  *
872  *  type "S":           The orphan MDT-object is a shard of a striped directory
873  *
874  * \see lfsck_layout_recreate_parent() for more types.
875  *
876  * The orphan name will be like:
877  * ${FID}-${infix}-${type}-${conflict_version}
878  *
879  * \param[out] count    if some others inserted some linkEA entries by race,
880  *                      then return the linkEA entries count.
881  *
882  * \retval              positive number for repaired cases
883  * \retval              0 if needs to repair nothing
884  * \retval              negative error number on failure
885  */
886 static int lfsck_namespace_insert_orphan(const struct lu_env *env,
887                                          struct lfsck_component *com,
888                                          struct dt_object *orphan,
889                                          const char *infix, const char *type,
890                                          int *count)
891 {
892         struct lfsck_thread_info        *info   = lfsck_env_info(env);
893         struct lu_name                  *cname  = &info->lti_name;
894         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
895         struct lu_attr                  *la     = &info->lti_la2;
896         const struct lu_fid             *cfid   = lfsck_dto2fid(orphan);
897         const struct lu_fid             *pfid;
898         struct lu_fid                    tfid;
899         struct lfsck_instance           *lfsck  = com->lc_lfsck;
900         struct dt_device                *dev    = lfsck_obj2dev(orphan);
901         struct dt_object                *parent;
902         struct thandle                  *th     = NULL;
903         struct lfsck_lock_handle        *pllh   = &info->lti_llh;
904         struct lustre_handle             clh    = { 0 };
905         struct linkea_data               ldata2 = { NULL };
906         struct lu_buf                    linkea_buf;
907         int                              namelen;
908         int                              idx    = 0;
909         int                              rc     = 0;
910         bool                             exist  = false;
911         ENTRY;
912
913         cname->ln_name = NULL;
914         if (unlikely(lfsck->li_lpf_obj == NULL))
915                 GOTO(log, rc = -ENXIO);
916
917         parent = lfsck->li_lpf_obj;
918         pfid = lfsck_dto2fid(parent);
919
920 again:
921         do {
922                 namelen = snprintf(info->lti_key, NAME_MAX, DFID"%s-%s-%d",
923                                    PFID(cfid), infix, type, idx++);
924                 rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
925                                (const struct dt_key *)info->lti_key);
926                 if (rc != 0 && rc != -ENOENT)
927                         GOTO(log, rc);
928
929                 if (unlikely(rc == 0 && lu_fid_eq(cfid, &tfid)))
930                         exist = true;
931         } while (rc == 0 && !exist);
932
933         rc = lfsck_lock(env, lfsck, parent, info->lti_key, pllh,
934                         MDS_INODELOCK_UPDATE, LCK_PW);
935         if (rc != 0)
936                 GOTO(log, rc);
937
938         /* Re-check whether the name conflict with othrs after taken
939          * the ldlm lock. */
940         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
941                        (const struct dt_key *)info->lti_key);
942         if (rc == 0) {
943                 if (!lu_fid_eq(cfid, &tfid)) {
944                         exist = false;
945                         lfsck_unlock(pllh);
946                         goto again;
947                 }
948
949                 exist = true;
950         } else if (rc != -ENOENT) {
951                 GOTO(log, rc);
952         } else {
953                 exist = false;
954         }
955
956         cname->ln_name = info->lti_key;
957         cname->ln_namelen = namelen;
958         rc = linkea_links_new(&ldata2, &info->lti_linkea_buf2,
959                               cname, pfid);
960         if (rc != 0)
961                 GOTO(log, rc);
962
963         rc = lfsck_ibits_lock(env, lfsck, orphan, &clh,
964                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_LOOKUP |
965                               MDS_INODELOCK_XATTR, LCK_EX);
966         if (rc != 0)
967                 GOTO(log, rc);
968
969         lfsck_buf_init(&linkea_buf, ldata2.ld_buf->lb_buf,
970                        ldata2.ld_leh->leh_len);
971         th = dt_trans_create(env, dev);
972         if (IS_ERR(th))
973                 GOTO(log, rc = PTR_ERR(th));
974
975         if (S_ISDIR(lfsck_object_type(orphan))) {
976                 rc = dt_declare_delete(env, orphan,
977                                        (const struct dt_key *)dotdot, th);
978                 if (rc != 0)
979                         GOTO(stop, rc);
980
981                 rec->rec_type = S_IFDIR;
982                 rec->rec_fid = pfid;
983                 rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,
984                                        (const struct dt_key *)dotdot, th);
985                 if (rc != 0)
986                         GOTO(stop, rc);
987         }
988
989         rc = dt_declare_xattr_set(env, orphan, &linkea_buf,
990                                   XATTR_NAME_LINK, 0, th);
991         if (rc != 0)
992                 GOTO(stop, rc);
993
994         if (!exist) {
995                 rec->rec_type = lfsck_object_type(orphan) & S_IFMT;
996                 rec->rec_fid = cfid;
997                 rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
998                                        (const struct dt_key *)cname->ln_name,
999                                        th);
1000                 if (rc != 0)
1001                         GOTO(stop, rc);
1002
1003                 if (S_ISDIR(rec->rec_type)) {
1004                         rc = dt_declare_ref_add(env, parent, th);
1005                         if (rc != 0)
1006                                 GOTO(stop, rc);
1007                 }
1008         }
1009
1010         memset(la, 0, sizeof(*la));
1011         la->la_ctime = ktime_get_real_seconds();
1012         la->la_valid = LA_CTIME;
1013         rc = dt_declare_attr_set(env, orphan, la, th);
1014         if (rc != 0)
1015                 GOTO(stop, rc);
1016
1017         rc = dt_trans_start_local(env, dev, th);
1018         if (rc != 0)
1019                 GOTO(stop, rc);
1020
1021         dt_write_lock(env, orphan, 0);
1022         rc = lfsck_links_read2_with_rec(env, orphan, &ldata2);
1023         if (likely(rc == -ENODATA || rc == -EINVAL)) {
1024                 if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1025                         GOTO(unlock, rc = 1);
1026
1027                 if (S_ISDIR(lfsck_object_type(orphan))) {
1028                         rc = dt_delete(env, orphan,
1029                                        (const struct dt_key *)dotdot, th);
1030                         if (rc != 0)
1031                                 GOTO(unlock, rc);
1032
1033                         rec->rec_type = S_IFDIR;
1034                         rec->rec_fid = pfid;
1035                         rc = dt_insert(env, orphan, (const struct dt_rec *)rec,
1036                                        (const struct dt_key *)dotdot, th, 1);
1037                         if (rc != 0)
1038                                 GOTO(unlock, rc);
1039                 }
1040
1041                 rc = dt_xattr_set(env, orphan, &linkea_buf, XATTR_NAME_LINK, 0,
1042                                   th);
1043         } else {
1044                 if (rc == 0 && count != NULL)
1045                         *count = ldata2.ld_leh->leh_reccount;
1046
1047                 GOTO(unlock, rc);
1048         }
1049         dt_write_unlock(env, orphan);
1050
1051         if (rc == 0 && !exist) {
1052                 rec->rec_type = lfsck_object_type(orphan) & S_IFMT;
1053                 rec->rec_fid = cfid;
1054                 rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1055                                (const struct dt_key *)cname->ln_name, th, 1);
1056                 if (rc == 0 && S_ISDIR(rec->rec_type)) {
1057                         dt_write_lock(env, parent, 0);
1058                         rc = dt_ref_add(env, parent, th);
1059                         dt_write_unlock(env, parent);
1060                 }
1061         }
1062
1063         if (rc == 0)
1064                 rc = dt_attr_set(env, orphan, la, th);
1065
1066         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1067
1068 unlock:
1069         dt_write_unlock(env, orphan);
1070
1071 stop:
1072         dt_trans_stop(env, dev, th);
1073
1074 log:
1075         lfsck_ibits_unlock(&clh, LCK_EX);
1076         lfsck_unlock(pllh);
1077         CDEBUG(D_LFSCK, "%s: namespace LFSCK insert orphan for the "
1078                "object "DFID", name = %s: rc = %d\n",
1079                lfsck_lfsck2name(lfsck), PFID(cfid),
1080                cname->ln_name != NULL ? cname->ln_name : "<NULL>", rc);
1081
1082         if (rc != 0) {
1083                 struct lfsck_namespace *ns = com->lc_file_ram;
1084
1085                 ns->ln_flags |= LF_INCONSISTENT;
1086         }
1087
1088         return rc;
1089 }
1090
1091 /**
1092  * Add the specified name entry back to namespace.
1093  *
1094  * If there is a linkEA entry that back references a name entry under
1095  * some parent directory, but such parent directory does not have the
1096  * claimed name entry. On the other hand, the linkEA entries count is
1097  * not larger than the MDT-object's hard link count. Under such case,
1098  * it is quite possible that the name entry is lost. Then the LFSCK
1099  * should add the name entry back to the namespace.
1100  *
1101  * \param[in] env       pointer to the thread context
1102  * \param[in] com       pointer to the lfsck component
1103  * \param[in] parent    pointer to the directory under which the name entry
1104  *                      will be inserted into
1105  * \param[in] child     pointer to the object referenced by the name entry
1106  *                      that to be inserted into the parent
1107  * \param[in] name      the name for the child in the parent directory
1108  *
1109  * \retval              positive number for repaired cases
1110  * \retval              0 if nothing to be repaired
1111  * \retval              negative error number on failure
1112  */
1113 static int lfsck_namespace_insert_normal(const struct lu_env *env,
1114                                          struct lfsck_component *com,
1115                                          struct dt_object *parent,
1116                                          struct dt_object *child,
1117                                          const char *name)
1118 {
1119         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1120         struct lu_attr                  *la     = &info->lti_la;
1121         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
1122         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1123         /* The child and its name may be on different MDTs. */
1124         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
1125         const struct lu_fid             *cfid   = lfsck_dto2fid(child);
1126         struct dt_device                *dev    = lfsck->li_next;
1127         struct thandle                  *th     = NULL;
1128         struct lfsck_lock_handle        *llh    = &info->lti_llh;
1129         int                              rc     = 0;
1130         ENTRY;
1131
1132         /* @parent/@child may be based on lfsck->li_bottom,
1133          * but here we need the object based on the lfsck->li_next. */
1134
1135         parent = lfsck_object_locate(dev, parent);
1136         if (IS_ERR(parent))
1137                 GOTO(log, rc = PTR_ERR(parent));
1138
1139         if (unlikely(!dt_try_as_dir(env, parent)))
1140                 GOTO(log, rc = -ENOTDIR);
1141
1142         child = lfsck_object_locate(dev, child);
1143         if (IS_ERR(child))
1144                 GOTO(log, rc = PTR_ERR(child));
1145
1146         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1147                 GOTO(log, rc = 1);
1148
1149         rc = lfsck_lock(env, lfsck, parent, name, llh,
1150                         MDS_INODELOCK_UPDATE, LCK_PW);
1151         if (rc != 0)
1152                 GOTO(log, rc);
1153
1154         th = dt_trans_create(env, dev);
1155         if (IS_ERR(th))
1156                 GOTO(unlock, rc = PTR_ERR(th));
1157
1158         rec->rec_type = lfsck_object_type(child) & S_IFMT;
1159         rec->rec_fid = cfid;
1160         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
1161                                (const struct dt_key *)name, th);
1162         if (rc != 0)
1163                 GOTO(stop, rc);
1164
1165         if (S_ISDIR(rec->rec_type)) {
1166                 rc = dt_declare_ref_add(env, parent, th);
1167                 if (rc != 0)
1168                         GOTO(stop, rc);
1169         }
1170
1171         memset(la, 0, sizeof(*la));
1172         la->la_ctime = ktime_get_real_seconds();
1173         la->la_valid = LA_CTIME;
1174         rc = dt_declare_attr_set(env, parent, la, th);
1175         if (rc != 0)
1176                 GOTO(stop, rc);
1177
1178         rc = dt_declare_attr_set(env, child, la, th);
1179         if (rc != 0)
1180                 GOTO(stop, rc);
1181
1182         rc = dt_trans_start_local(env, dev, th);
1183         if (rc != 0)
1184                 GOTO(stop, rc);
1185
1186         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1187                        (const struct dt_key *)name, th, 1);
1188         if (rc != 0)
1189                 GOTO(stop, rc);
1190
1191         if (S_ISDIR(rec->rec_type)) {
1192                 dt_write_lock(env, parent, 0);
1193                 rc = dt_ref_add(env, parent, th);
1194                 dt_write_unlock(env, parent);
1195                 if (rc != 0)
1196                         GOTO(stop, rc);
1197         }
1198
1199         la->la_ctime = ktime_get_real_seconds();
1200         rc = dt_attr_set(env, parent, la, th);
1201         if (rc != 0)
1202                 GOTO(stop, rc);
1203
1204         rc = dt_attr_set(env, child, la, th);
1205
1206         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1207
1208 stop:
1209         dt_trans_stop(env, dev, th);
1210
1211 unlock:
1212         lfsck_unlock(llh);
1213
1214 log:
1215         CDEBUG(D_LFSCK, "%s: namespace LFSCK insert object "DFID" with "
1216                "the name %s and type %o to the parent "DFID": rc = %d\n",
1217                lfsck_lfsck2name(lfsck), PFID(cfid), name,
1218                lfsck_object_type(child) & S_IFMT, PFID(pfid), rc);
1219
1220         if (rc != 0) {
1221                 struct lfsck_namespace *ns = com->lc_file_ram;
1222
1223                 ns->ln_flags |= LF_INCONSISTENT;
1224                 if (rc > 0)
1225                         ns->ln_lost_dirent_repaired++;
1226         }
1227
1228         return rc;
1229 }
1230
1231 /**
1232  * Create the specified orphan directory.
1233  *
1234  * For the case that the parent MDT-object stored in some MDT-object's
1235  * linkEA entry is lost, the LFSCK will re-create the parent object as
1236  * an orphan and insert it into .lustre/lost+found/MDTxxxx/ directory
1237  * with the name ${FID}-P-${conflict_version}.
1238  *
1239  * \param[in] env       pointer to the thread context
1240  * \param[in] com       pointer to the lfsck component
1241  * \param[in] orphan    pointer to the orphan MDT-object to be created
1242  * \param[in] lmv       pointer to master LMV EA that will be set to the orphan
1243  *
1244  * \retval              positive number for repaired cases
1245  * \retval              negative error number on failure
1246  */
1247 static int lfsck_namespace_create_orphan_dir(const struct lu_env *env,
1248                                              struct lfsck_component *com,
1249                                              struct dt_object *orphan,
1250                                              struct lmv_mds_md_v1 *lmv)
1251 {
1252         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1253         struct lu_attr                  *la     = &info->lti_la;
1254         struct dt_allocation_hint       *hint   = &info->lti_hint;
1255         struct dt_object_format         *dof    = &info->lti_dof;
1256         struct lu_name                  *cname  = &info->lti_name2;
1257         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
1258         struct lmv_mds_md_v1            *lmv2   = &info->lti_lmv2;
1259         const struct lu_fid             *cfid   = lfsck_dto2fid(orphan);
1260         struct lu_fid                    tfid;
1261         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1262         struct lfsck_namespace          *ns     = com->lc_file_ram;
1263         struct dt_device                *dev    = lfsck_obj2dev(orphan);
1264         struct dt_object                *parent = NULL;
1265         struct thandle                  *th     = NULL;
1266         struct lfsck_lock_handle        *llh    = &info->lti_llh;
1267         struct linkea_data               ldata  = { NULL };
1268         struct lu_buf                    linkea_buf;
1269         struct lu_buf                    lmv_buf;
1270         char                             name[32];
1271         int                              namelen;
1272         int                              idx    = 0;
1273         int                              rc     = 0;
1274         int                              rc1    = 0;
1275         ENTRY;
1276
1277         LASSERT(!dt_object_exists(orphan));
1278
1279         cname->ln_name = NULL;
1280         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1281                 GOTO(log, rc = 1);
1282
1283         if (dt_object_remote(orphan)) {
1284                 LASSERT(lfsck->li_lpf_root_obj != NULL);
1285
1286                 idx = lfsck_find_mdt_idx_by_fid(env, lfsck, cfid);
1287                 if (idx < 0)
1288                         GOTO(log, rc = idx);
1289
1290                 snprintf(name, 8, "MDT%04x", idx);
1291                 rc = dt_lookup(env, lfsck->li_lpf_root_obj,
1292                                (struct dt_rec *)&tfid,
1293                                (const struct dt_key *)name);
1294                 if (rc != 0)
1295                         GOTO(log, rc = (rc == -ENOENT ? -ENXIO : rc));
1296
1297                 parent = lfsck_object_find_bottom(env, lfsck, &tfid);
1298                 if (IS_ERR(parent))
1299                         GOTO(log, rc = PTR_ERR(parent));
1300
1301                 if (unlikely(!dt_try_as_dir(env, parent)))
1302                         GOTO(log, rc = -ENOTDIR);
1303         } else {
1304                 if (unlikely(lfsck->li_lpf_obj == NULL))
1305                         GOTO(log, rc = -ENXIO);
1306
1307                 parent = lfsck->li_lpf_obj;
1308         }
1309
1310         dev = lfsck_find_dev_by_fid(env, lfsck, cfid);
1311         if (IS_ERR(dev))
1312                 GOTO(log, rc = PTR_ERR(dev));
1313
1314         idx = 0;
1315
1316 again:
1317         do {
1318                 namelen = snprintf(name, 31, DFID"-P-%d",
1319                                    PFID(cfid), idx++);
1320                 rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1321                                (const struct dt_key *)name);
1322                 if (rc != 0 && rc != -ENOENT)
1323                         GOTO(log, rc);
1324         } while (rc == 0);
1325
1326         rc = lfsck_lock(env, lfsck, parent, name, llh,
1327                         MDS_INODELOCK_UPDATE, LCK_PW);
1328         if (rc != 0)
1329                 GOTO(log, rc);
1330
1331         /* Re-check whether the name conflict with othrs after taken
1332          * the ldlm lock. */
1333         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1334                        (const struct dt_key *)name);
1335         if (unlikely(rc == 0)) {
1336                 lfsck_unlock(llh);
1337                 goto again;
1338         }
1339
1340         if (rc != -ENOENT)
1341                 GOTO(unlock1, rc);
1342
1343         cname->ln_name = name;
1344         cname->ln_namelen = namelen;
1345
1346         memset(la, 0, sizeof(*la));
1347         la->la_mode = S_IFDIR | 0700;
1348         la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
1349                        LA_ATIME | LA_MTIME | LA_CTIME;
1350
1351         orphan->do_ops->do_ah_init(env, hint, parent, orphan,
1352                                    la->la_mode & S_IFMT);
1353
1354         memset(dof, 0, sizeof(*dof));
1355         dof->dof_type = dt_mode_to_dft(S_IFDIR);
1356
1357         rc = linkea_links_new(&ldata, &info->lti_linkea_buf2,
1358                               cname, lfsck_dto2fid(parent));
1359         if (rc != 0)
1360                 GOTO(unlock1, rc);
1361
1362         th = dt_trans_create(env, dev);
1363         if (IS_ERR(th))
1364                 GOTO(unlock1, rc = PTR_ERR(th));
1365
1366         /* Sync the remote transaction to guarantee that the subsequent
1367          * lock against the @orphan can find the @orphan in time. */
1368         if (dt_object_remote(orphan))
1369                 th->th_sync = 1;
1370
1371         rc = dt_declare_create(env, orphan, la, hint, dof, th);
1372         if (rc != 0)
1373                 GOTO(stop, rc);
1374
1375         if (unlikely(!dt_try_as_dir(env, orphan)))
1376                 GOTO(stop, rc = -ENOTDIR);
1377
1378         rc = dt_declare_ref_add(env, orphan, th);
1379         if (rc != 0)
1380                 GOTO(stop, rc);
1381
1382         rec->rec_type = S_IFDIR;
1383         rec->rec_fid = cfid;
1384         rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,
1385                                (const struct dt_key *)dot, th);
1386         if (rc != 0)
1387                 GOTO(stop, rc);
1388
1389         rec->rec_fid = lfsck_dto2fid(parent);
1390         rc = dt_declare_insert(env, orphan, (const struct dt_rec *)rec,
1391                                (const struct dt_key *)dotdot, th);
1392         if (rc != 0)
1393                 GOTO(stop, rc);
1394
1395         if (lmv != NULL) {
1396                 lmv->lmv_magic = LMV_MAGIC;
1397                 lmv->lmv_master_mdt_index = lfsck_dev_idx(lfsck);
1398                 lfsck_lmv_header_cpu_to_le(lmv2, lmv);
1399                 lfsck_buf_init(&lmv_buf, lmv2, sizeof(*lmv2));
1400                 rc = dt_declare_xattr_set(env, orphan, &lmv_buf,
1401                                           XATTR_NAME_LMV, 0, th);
1402                 if (rc != 0)
1403                         GOTO(stop, rc);
1404         }
1405
1406         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
1407                        ldata.ld_leh->leh_len);
1408         rc = dt_declare_xattr_set(env, orphan, &linkea_buf,
1409                                   XATTR_NAME_LINK, 0, th);
1410         if (rc != 0)
1411                 GOTO(stop, rc);
1412
1413         rec->rec_fid = cfid;
1414         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
1415                                (const struct dt_key *)name, th);
1416         if (rc == 0)
1417                 rc = dt_declare_ref_add(env, parent, th);
1418
1419         if (rc != 0)
1420                 GOTO(stop, rc);
1421
1422         rc = dt_trans_start_local(env, dev, th);
1423         if (rc != 0)
1424                 GOTO(stop, rc);
1425
1426         dt_write_lock(env, orphan, 0);
1427         rc = dt_create(env, orphan, la, hint, dof, th);
1428         if (rc != 0)
1429                 GOTO(unlock2, rc);
1430
1431         rc = dt_ref_add(env, orphan, th);
1432         if (rc != 0)
1433                 GOTO(unlock2, rc);
1434
1435         rec->rec_fid = cfid;
1436         rc = dt_insert(env, orphan, (const struct dt_rec *)rec,
1437                        (const struct dt_key *)dot, th, 1);
1438         if (rc != 0)
1439                 GOTO(unlock2, rc);
1440
1441         rec->rec_fid = lfsck_dto2fid(parent);
1442         rc = dt_insert(env, orphan, (const struct dt_rec *)rec,
1443                        (const struct dt_key *)dotdot, th, 1);
1444         if (rc != 0)
1445                 GOTO(unlock2, rc);
1446
1447         if (lmv != NULL) {
1448                 rc = dt_xattr_set(env, orphan, &lmv_buf, XATTR_NAME_LMV, 0, th);
1449                 if (rc != 0)
1450                         GOTO(unlock2, rc);
1451         }
1452
1453         rc = dt_xattr_set(env, orphan, &linkea_buf,
1454                           XATTR_NAME_LINK, 0, th);
1455         dt_write_unlock(env, orphan);
1456         if (rc != 0)
1457                 GOTO(stop, rc);
1458
1459         rec->rec_fid = cfid;
1460         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1461                        (const struct dt_key *)name, th, 1);
1462         if (rc == 0) {
1463                 dt_write_lock(env, parent, 0);
1464                 rc = dt_ref_add(env, parent, th);
1465                 dt_write_unlock(env, parent);
1466         }
1467
1468         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1469
1470 unlock2:
1471         dt_write_unlock(env, orphan);
1472
1473 stop:
1474         rc1 = dt_trans_stop(env, dev, th);
1475         if (rc1 != 0 && rc > 0)
1476                 rc = rc1;
1477
1478 unlock1:
1479         lfsck_unlock(llh);
1480
1481 log:
1482         CDEBUG(D_LFSCK, "%s: namespace LFSCK create orphan dir for "
1483                "the object "DFID", name = %s: rc = %d\n",
1484                lfsck_lfsck2name(lfsck), PFID(cfid),
1485                cname->ln_name != NULL ? cname->ln_name : "<NULL>", rc);
1486
1487         if (parent != NULL && !IS_ERR(parent) && parent != lfsck->li_lpf_obj)
1488                 lfsck_object_put(env, parent);
1489
1490         if (rc != 0)
1491                 ns->ln_flags |= LF_INCONSISTENT;
1492
1493         return rc;
1494 }
1495
1496 /**
1497  * Remove the specified entry from the linkEA.
1498  *
1499  * Locate the linkEA entry with the given @cname and @pfid, then
1500  * remove this entry or the other entries those are repeated with
1501  * this entry.
1502  *
1503  * \param[in] env       pointer to the thread context
1504  * \param[in] com       pointer to the lfsck component
1505  * \param[in] obj       pointer to the dt_object to be handled
1506  * \param[in,out]ldata  pointer to the buffer that holds the linkEA
1507  * \param[in] cname     the name for the child in the parent directory
1508  * \param[in] pfid      the parent directory's FID for the linkEA
1509  * \param[in] next      if true, then remove the first found linkEA
1510  *                      entry, and move the ldata->ld_lee to next entry
1511  *
1512  * \retval              positive number for repaired cases
1513  * \retval              0 if nothing to be repaired
1514  * \retval              negative error number on failure
1515  */
1516 static int lfsck_namespace_shrink_linkea(const struct lu_env *env,
1517                                          struct lfsck_component *com,
1518                                          struct dt_object *obj,
1519                                          struct linkea_data *ldata,
1520                                          struct lu_name *cname,
1521                                          struct lu_fid *pfid,
1522                                          bool next)
1523 {
1524         struct lfsck_instance           *lfsck     = com->lc_lfsck;
1525         struct dt_device                *dev       = lfsck_obj2dev(obj);
1526         struct lfsck_bookmark           *bk        = &lfsck->li_bookmark_ram;
1527         struct thandle                  *th        = NULL;
1528         struct lustre_handle             lh        = { 0 };
1529         struct linkea_data               ldata_new = { NULL };
1530         struct lu_buf                    linkea_buf;
1531         int                              buflen    = 0;
1532         int                              rc        = 0;
1533         ENTRY;
1534
1535         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
1536                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
1537                               LCK_EX);
1538         if (rc != 0)
1539                 GOTO(log, rc);
1540
1541         if (next)
1542                 lfsck_linkea_del_buf(ldata, cname);
1543         else
1544                 lfsck_namespace_filter_linkea_entry(ldata, cname, pfid,
1545                                                     true);
1546         if (ldata->ld_leh->leh_reccount > 0 ||
1547             unlikely(ldata->ld_leh->leh_overflow_time)) {
1548                 lfsck_buf_init(&linkea_buf, ldata->ld_buf->lb_buf,
1549                                ldata->ld_leh->leh_len);
1550                 buflen = linkea_buf.lb_len;
1551         }
1552
1553 again:
1554         th = dt_trans_create(env, dev);
1555         if (IS_ERR(th))
1556                 GOTO(unlock1, rc = PTR_ERR(th));
1557
1558         if (buflen != 0)
1559                 rc = dt_declare_xattr_set(env, obj, &linkea_buf,
1560                                           XATTR_NAME_LINK, 0, th);
1561         else
1562                 rc = dt_declare_xattr_del(env, obj, XATTR_NAME_LINK, th);
1563         if (rc != 0)
1564                 GOTO(stop, rc);
1565
1566         rc = dt_trans_start_local(env, dev, th);
1567         if (rc != 0)
1568                 GOTO(stop, rc);
1569
1570         dt_write_lock(env, obj, 0);
1571         if (unlikely(lfsck_is_dead_obj(obj)))
1572                 GOTO(unlock2, rc = -ENOENT);
1573
1574         rc = lfsck_links_read2_with_rec(env, obj, &ldata_new);
1575         if (rc)
1576                 GOTO(unlock2, rc = (rc == -ENODATA ? 0 : rc));
1577
1578         /* The specified linkEA entry has been removed by race. */
1579         rc = linkea_links_find(&ldata_new, cname, pfid);
1580         if (rc != 0)
1581                 GOTO(unlock2, rc = 0);
1582
1583         if (bk->lb_param & LPF_DRYRUN)
1584                 GOTO(unlock2, rc = 1);
1585
1586         if (next)
1587                 lfsck_linkea_del_buf(&ldata_new, cname);
1588         else
1589                 lfsck_namespace_filter_linkea_entry(&ldata_new, cname, pfid,
1590                                                     true);
1591
1592         if (buflen < ldata_new.ld_leh->leh_len) {
1593                 dt_write_unlock(env, obj);
1594                 dt_trans_stop(env, dev, th);
1595                 lfsck_buf_init(&linkea_buf, ldata_new.ld_buf->lb_buf,
1596                                ldata_new.ld_leh->leh_len);
1597                 goto again;
1598         }
1599
1600         if (ldata_new.ld_leh->leh_reccount > 0 ||
1601             unlikely(ldata->ld_leh->leh_overflow_time))
1602                 rc = lfsck_links_write(env, obj, &ldata_new, th);
1603         else
1604                 rc = dt_xattr_del(env, obj, XATTR_NAME_LINK, th);
1605
1606         GOTO(unlock2, rc = (rc == 0 ? 1 : rc));
1607
1608 unlock2:
1609         dt_write_unlock(env, obj);
1610
1611 stop:
1612         dt_trans_stop(env, dev, th);
1613
1614 unlock1:
1615         lfsck_ibits_unlock(&lh, LCK_EX);
1616
1617 log:
1618         CDEBUG(D_LFSCK, "%s: namespace LFSCK remove %s linkEA entry "
1619                "for the object: "DFID", parent "DFID", name %.*s\n",
1620                lfsck_lfsck2name(lfsck), next ? "invalid" : "redundant",
1621                PFID(lfsck_dto2fid(obj)), PFID(pfid), cname->ln_namelen,
1622                cname->ln_name);
1623
1624         if (rc != 0) {
1625                 struct lfsck_namespace *ns = com->lc_file_ram;
1626
1627                 ns->ln_flags |= LF_INCONSISTENT;
1628         }
1629
1630         return rc;
1631 }
1632
1633 /**
1634  * Conditionally remove the specified entry from the linkEA.
1635  *
1636  * Take the parent lock firstly, then check whether the specified
1637  * name entry exists or not: if yes, do nothing; otherwise, call
1638  * lfsck_namespace_shrink_linkea() to remove the linkea entry.
1639  *
1640  * \param[in] env       pointer to the thread context
1641  * \param[in] com       pointer to the lfsck component
1642  * \param[in] parent    pointer to the parent directory
1643  * \param[in] child     pointer to the child object that holds the linkEA
1644  * \param[in,out]ldata  pointer to the buffer that holds the linkEA
1645  * \param[in] cname     the name for the child in the parent directory
1646  * \param[in] pfid      the parent directory's FID for the linkEA
1647  *
1648  * \retval              positive number for repaired cases
1649  * \retval              0 if nothing to be repaired
1650  * \retval              negative error number on failure
1651  */
1652 static int lfsck_namespace_shrink_linkea_cond(const struct lu_env *env,
1653                                               struct lfsck_component *com,
1654                                               struct dt_object *parent,
1655                                               struct dt_object *child,
1656                                               struct linkea_data *ldata,
1657                                               struct lu_name *cname,
1658                                               struct lu_fid *pfid)
1659 {
1660         struct lfsck_thread_info *info  = lfsck_env_info(env);
1661         struct lu_fid            *cfid  = &info->lti_fid3;
1662         struct lfsck_lock_handle *llh   = &info->lti_llh;
1663         int                       rc;
1664         ENTRY;
1665
1666         rc = lfsck_lock(env, com->lc_lfsck, parent, cname->ln_name, llh,
1667                         MDS_INODELOCK_UPDATE, LCK_PR);
1668         if (rc != 0)
1669                 RETURN(rc);
1670
1671         dt_read_lock(env, parent, 0);
1672         if (unlikely(lfsck_is_dead_obj(parent))) {
1673                 dt_read_unlock(env, parent);
1674                 lfsck_unlock(llh);
1675                 rc = lfsck_namespace_shrink_linkea(env, com, child, ldata,
1676                                                    cname, pfid, true);
1677
1678                 RETURN(rc);
1679         }
1680
1681         rc = dt_lookup(env, parent, (struct dt_rec *)cfid,
1682                        (const struct dt_key *)cname->ln_name);
1683         dt_read_unlock(env, parent);
1684
1685         /* It is safe to release the ldlm lock, because when the logic come
1686          * here, we have got all the needed information above whether the
1687          * linkEA entry is valid or not. It is not important that others
1688          * may add new linkEA entry after the ldlm lock released. If other
1689          * has removed the specified linkEA entry by race, then it is OK,
1690          * because the subsequent lfsck_namespace_shrink_linkea() can handle
1691          * such case. */
1692         lfsck_unlock(llh);
1693         if (rc == -ENOENT) {
1694                 rc = lfsck_namespace_shrink_linkea(env, com, child, ldata,
1695                                                    cname, pfid, true);
1696
1697                 RETURN(rc);
1698         }
1699
1700         if (rc != 0)
1701                 RETURN(rc);
1702
1703         /* The LFSCK just found some internal status of cross-MDTs
1704          * create operation. That is normal. */
1705         if (lu_fid_eq(cfid, lfsck_dto2fid(child))) {
1706                 linkea_next_entry(ldata);
1707
1708                 RETURN(0);
1709         }
1710
1711         rc = lfsck_namespace_shrink_linkea(env, com, child, ldata, cname,
1712                                            pfid, true);
1713
1714         RETURN(rc);
1715 }
1716
1717 /**
1718  * Conditionally replace name entry in the parent.
1719  *
1720  * As required, the LFSCK may re-create the lost MDT-object for dangling
1721  * name entry, but such repairing may be wrong because of bad FID in the
1722  * name entry. As the LFSCK processing, the real MDT-object may be found,
1723  * then the LFSCK should check whether the former re-created MDT-object
1724  * has been modified or not, if not, then destroy it and update the name
1725  * entry in the parent to reference the real MDT-object.
1726  *
1727  * \param[in] env       pointer to the thread context
1728  * \param[in] com       pointer to the lfsck component
1729  * \param[in] parent    pointer to the parent directory
1730  * \param[in] child     pointer to the MDT-object that may be the real
1731  *                      MDT-object corresponding to the name entry in parent
1732  * \param[in] cfid      the current FID in the name entry
1733  * \param[in] cname     contains the name of the child in the parent directory
1734  *
1735  * \retval              positive number for repaired cases
1736  * \retval              0 if nothing to be repaired
1737  * \retval              negative error number on failure
1738  */
1739 static int lfsck_namespace_replace_cond(const struct lu_env *env,
1740                                         struct lfsck_component *com,
1741                                         struct dt_object *parent,
1742                                         struct dt_object *child,
1743                                         const struct lu_fid *cfid,
1744                                         const struct lu_name *cname)
1745 {
1746         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1747         struct lu_attr                  *la     = &info->lti_la;
1748         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
1749         struct lu_fid                    tfid;
1750         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1751         /* The child and its name may be on different MDTs. */
1752         struct dt_device                *dev    = lfsck->li_next;
1753         const char                      *name   = cname->ln_name;
1754         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
1755         struct dt_object                *cobj   = NULL;
1756         struct lfsck_lock_handle        *pllh   = &info->lti_llh;
1757         struct lustre_handle             clh    = { 0 };
1758         struct linkea_data               ldata  = { NULL };
1759         struct thandle                  *th     = NULL;
1760         bool                             exist  = true;
1761         int                              rc     = 0;
1762         ENTRY;
1763
1764         /* @parent/@child may be based on lfsck->li_bottom,
1765          * but here we need the object based on the lfsck->li_next. */
1766
1767         parent = lfsck_object_locate(dev, parent);
1768         if (IS_ERR(parent))
1769                 GOTO(log, rc = PTR_ERR(parent));
1770
1771         if (unlikely(!dt_try_as_dir(env, parent)))
1772                 GOTO(log, rc = -ENOTDIR);
1773
1774         rc = lfsck_lock(env, lfsck, parent, name, pllh,
1775                         MDS_INODELOCK_UPDATE, LCK_PW);
1776         if (rc != 0)
1777                 GOTO(log, rc);
1778
1779         if (!fid_is_sane(cfid)) {
1780                 exist = false;
1781                 goto replace;
1782         }
1783
1784         cobj = lfsck_object_find_by_dev(env, dev, cfid);
1785         if (IS_ERR(cobj)) {
1786                 rc = PTR_ERR(cobj);
1787                 if (rc == -ENOENT) {
1788                         exist = false;
1789                         goto replace;
1790                 }
1791
1792                 GOTO(log, rc);
1793         }
1794
1795         if (!dt_object_exists(cobj)) {
1796                 exist = false;
1797                 goto replace;
1798         }
1799
1800         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
1801                        (const struct dt_key *)name);
1802         if (rc == -ENOENT) {
1803                 exist = false;
1804                 goto replace;
1805         }
1806
1807         if (rc != 0)
1808                 GOTO(log, rc);
1809
1810         /* Someone changed the name entry, cannot replace it. */
1811         if (!lu_fid_eq(cfid, &tfid))
1812                 GOTO(log, rc = 0);
1813
1814         /* lock the object to be destroyed. */
1815         rc = lfsck_ibits_lock(env, lfsck, cobj, &clh,
1816                               MDS_INODELOCK_UPDATE |
1817                               MDS_INODELOCK_UPDATE | MDS_INODELOCK_XATTR,
1818                               LCK_EX);
1819         if (rc != 0)
1820                 GOTO(log, rc);
1821
1822         if (unlikely(lfsck_is_dead_obj(cobj))) {
1823                 exist = false;
1824                 goto replace;
1825         }
1826
1827         rc = dt_attr_get(env, cobj, la);
1828         if (rc != 0)
1829                 GOTO(log, rc);
1830
1831         /* The object has been modified by other(s), or it is not created by
1832          * LFSCK, the two cases are indistinguishable. So cannot replace it. */
1833         if (la->la_ctime != 0)
1834                 GOTO(log, rc);
1835
1836         if (S_ISREG(la->la_mode)) {
1837                 rc = dt_xattr_get(env, cobj, &LU_BUF_NULL, XATTR_NAME_LOV);
1838                 /* If someone has created related OST-object(s),
1839                  * then keep it. */
1840                 if ((rc > 0) || (rc < 0 && rc != -ENODATA))
1841                         GOTO(log, rc = (rc > 0 ? 0 : rc));
1842         }
1843
1844 replace:
1845         dt_read_lock(env, child, 0);
1846         rc = lfsck_links_read2_with_rec(env, child, &ldata);
1847         dt_read_unlock(env, child);
1848
1849         /* Someone changed the child, no need to replace. */
1850         if (rc == -ENODATA)
1851                 GOTO(log, rc = 0);
1852
1853         if (rc != 0)
1854                 GOTO(log, rc);
1855
1856         rc = linkea_links_find(&ldata, cname, pfid);
1857         /* Someone moved the child, no need to replace. */
1858         if (rc != 0)
1859                 GOTO(log, rc = 0);
1860
1861         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1862                 GOTO(log, rc = 1);
1863
1864         th = dt_trans_create(env, dev);
1865         if (IS_ERR(th))
1866                 GOTO(log, rc = PTR_ERR(th));
1867
1868         if (exist) {
1869                 rc = dt_declare_destroy(env, cobj, th);
1870                 if (rc != 0)
1871                         GOTO(stop, rc);
1872         }
1873
1874         rc = dt_declare_delete(env, parent, (const struct dt_key *)name, th);
1875         if (rc != 0)
1876                 GOTO(stop, rc);
1877
1878         rec->rec_type = S_IFDIR;
1879         rec->rec_fid = lfsck_dto2fid(child);
1880         rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
1881                                (const struct dt_key *)name, th);
1882         if (rc != 0)
1883                 GOTO(stop, rc);
1884
1885         rc = dt_trans_start_local(env, dev, th);
1886         if (rc != 0)
1887                 GOTO(stop, rc);
1888
1889         if (exist) {
1890                 rc = dt_destroy(env, cobj, th);
1891                 if (rc != 0)
1892                         GOTO(stop, rc);
1893         }
1894
1895         /* The old name entry maybe not exist. */
1896         rc = dt_delete(env, parent, (const struct dt_key *)name, th);
1897         if (rc != 0 && rc != -ENOENT)
1898                 GOTO(stop, rc);
1899
1900         rc = dt_insert(env, parent, (const struct dt_rec *)rec,
1901                        (const struct dt_key *)name, th, 1);
1902
1903         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1904
1905 stop:
1906         dt_trans_stop(env, dev, th);
1907
1908 log:
1909         lfsck_ibits_unlock(&clh, LCK_EX);
1910         lfsck_unlock(pllh);
1911
1912         if (cobj != NULL && !IS_ERR(cobj))
1913                 lfsck_object_put(env, cobj);
1914
1915         CDEBUG(D_LFSCK, "%s: namespace LFSCK conditionally destroy the "
1916                "object "DFID" because of conflict with the object "DFID
1917                " under the parent "DFID" with name %s: rc = %d\n",
1918                lfsck_lfsck2name(lfsck), PFID(cfid),
1919                PFID(lfsck_dto2fid(child)), PFID(pfid), name, rc);
1920
1921         return rc;
1922 }
1923
1924 /**
1925  * Overwrite the linkEA for the object with the given ldata.
1926  *
1927  * The caller should take the ldlm lock before the calling.
1928  *
1929  * \param[in] env       pointer to the thread context
1930  * \param[in] com       pointer to the lfsck component
1931  * \param[in] obj       pointer to the dt_object to be handled
1932  * \param[in] ldata     pointer to the new linkEA data
1933  *
1934  * \retval              positive number for repaired cases
1935  * \retval              0 if nothing to be repaired
1936  * \retval              negative error number on failure
1937  */
1938 int lfsck_namespace_rebuild_linkea(const struct lu_env *env,
1939                                    struct lfsck_component *com,
1940                                    struct dt_object *obj,
1941                                    struct linkea_data *ldata)
1942 {
1943         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1944         struct dt_device                *dev    = lfsck_obj2dev(obj);
1945         struct thandle                  *th     = NULL;
1946         struct lu_buf                    linkea_buf;
1947         int                              rc     = 0;
1948         ENTRY;
1949
1950         th = dt_trans_create(env, dev);
1951         if (IS_ERR(th))
1952                 GOTO(log, rc = PTR_ERR(th));
1953
1954         lfsck_buf_init(&linkea_buf, ldata->ld_buf->lb_buf,
1955                        ldata->ld_leh->leh_len);
1956         rc = dt_declare_xattr_set(env, obj, &linkea_buf,
1957                                   XATTR_NAME_LINK, 0, th);
1958         if (rc != 0)
1959                 GOTO(stop, rc);
1960
1961         rc = dt_trans_start_local(env, dev, th);
1962         if (rc != 0)
1963                 GOTO(stop, rc);
1964
1965         dt_write_lock(env, obj, 0);
1966         if (unlikely(lfsck_is_dead_obj(obj)))
1967                 GOTO(unlock, rc = 0);
1968
1969         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
1970                 GOTO(unlock, rc = 1);
1971
1972         rc = dt_xattr_set(env, obj, &linkea_buf,
1973                           XATTR_NAME_LINK, 0, th);
1974
1975         GOTO(unlock, rc = (rc == 0 ? 1 : rc));
1976
1977 unlock:
1978         dt_write_unlock(env, obj);
1979
1980 stop:
1981         dt_trans_stop(env, dev, th);
1982
1983 log:
1984         CDEBUG(D_LFSCK, "%s: namespace LFSCK rebuild linkEA for the "
1985                "object "DFID": rc = %d\n",
1986                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)), rc);
1987
1988         if (rc != 0) {
1989                 struct lfsck_namespace *ns = com->lc_file_ram;
1990
1991                 ns->ln_flags |= LF_INCONSISTENT;
1992         }
1993
1994         return rc;
1995 }
1996
1997 /**
1998  * Repair invalid name entry.
1999  *
2000  * If the name entry contains invalid information, such as bad file type
2001  * or (and) corrupted object FID, then either remove the name entry or
2002  * udpate the name entry with the given (right) information.
2003  *
2004  * \param[in] env       pointer to the thread context
2005  * \param[in] com       pointer to the lfsck component
2006  * \param[in] parent    pointer to the parent directory
2007  * \param[in] child     pointer to the object referenced by the name entry
2008  * \param[in] name      the old name of the child under the parent directory
2009  * \param[in] name2     the new name of the child under the parent directory
2010  * \param[in] type      the type claimed by the name entry
2011  * \param[in] update    update the name entry if true; otherwise, remove it
2012  * \param[in] dec       decrease the parent nlink count if true
2013  *
2014  * \retval              positive number for repaired successfully
2015  * \retval              0 if nothing to be repaired
2016  * \retval              negative error number on failure
2017  */
2018 int lfsck_namespace_repair_dirent(const struct lu_env *env,
2019                                   struct lfsck_component *com,
2020                                   struct dt_object *parent,
2021                                   struct dt_object *child,
2022                                   const char *name, const char *name2,
2023                                   __u16 type, bool update, bool dec)
2024 {
2025         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2026         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
2027         const struct lu_fid             *pfid   = lfsck_dto2fid(parent);
2028         const struct lu_fid             *cfid   = lfsck_dto2fid(child);
2029         struct lu_fid                    tfid;
2030         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2031         struct dt_device                *dev    = lfsck->li_next;
2032         struct thandle                  *th     = NULL;
2033         struct lfsck_lock_handle        *llh    = &info->lti_llh;
2034         struct lustre_handle             lh     = { 0 };
2035         int                              rc     = 0;
2036         ENTRY;
2037
2038         parent = lfsck_object_locate(dev, parent);
2039         if (IS_ERR(parent))
2040                 GOTO(log, rc = PTR_ERR(parent));
2041
2042         if (unlikely(!dt_try_as_dir(env, parent)))
2043                 GOTO(log, rc = -ENOTDIR);
2044
2045         if (!update || strcmp(name, name2) == 0)
2046                 rc = lfsck_lock(env, lfsck, parent, name, llh,
2047                                 MDS_INODELOCK_UPDATE, LCK_PW);
2048         else
2049                 rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2050                                       MDS_INODELOCK_UPDATE, LCK_PW);
2051         if (rc != 0)
2052                 GOTO(log, rc);
2053
2054         th = dt_trans_create(env, dev);
2055         if (IS_ERR(th))
2056                 GOTO(unlock1, rc = PTR_ERR(th));
2057
2058         rc = dt_declare_delete(env, parent, (const struct dt_key *)name, th);
2059         if (rc != 0)
2060                 GOTO(stop, rc);
2061
2062         if (update) {
2063                 rec->rec_type = lfsck_object_type(child) & S_IFMT;
2064                 rec->rec_fid = cfid;
2065                 rc = dt_declare_insert(env, parent,
2066                                        (const struct dt_rec *)rec,
2067                                        (const struct dt_key *)name2, th);
2068                 if (rc != 0)
2069                         GOTO(stop, rc);
2070         }
2071
2072         if (dec && S_ISDIR(type)) {
2073                 rc = dt_declare_ref_del(env, parent, th);
2074                 if (rc != 0)
2075                         GOTO(stop, rc);
2076         }
2077
2078         rc = dt_trans_start_local(env, dev, th);
2079         if (rc != 0)
2080                 GOTO(stop, rc);
2081
2082
2083         dt_write_lock(env, parent, 0);
2084         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
2085                        (const struct dt_key *)name);
2086         /* Someone has removed the bad name entry by race. */
2087         if (rc == -ENOENT)
2088                 GOTO(unlock2, rc = 0);
2089
2090         if (rc != 0)
2091                 GOTO(unlock2, rc);
2092
2093         /* Someone has removed the bad name entry and reused it for other
2094          * object by race. */
2095         if (!lu_fid_eq(&tfid, cfid))
2096                 GOTO(unlock2, rc = 0);
2097
2098         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
2099                 GOTO(unlock2, rc = 1);
2100
2101         rc = dt_delete(env, parent, (const struct dt_key *)name, th);
2102         if (rc != 0)
2103                 GOTO(unlock2, rc);
2104
2105         if (update) {
2106                 rc = dt_insert(env, parent,
2107                                (const struct dt_rec *)rec,
2108                                (const struct dt_key *)name2, th, 1);
2109                 if (rc != 0)
2110                         GOTO(unlock2, rc);
2111         }
2112
2113         if (dec && S_ISDIR(type)) {
2114                 rc = dt_ref_del(env, parent, th);
2115                 if (rc != 0)
2116                         GOTO(unlock2, rc);
2117         }
2118
2119         GOTO(unlock2, rc = (rc == 0 ? 1 : rc));
2120
2121 unlock2:
2122         dt_write_unlock(env, parent);
2123
2124 stop:
2125         dt_trans_stop(env, dev, th);
2126
2127         /* We are not sure whether the child will become orphan or not.
2128          * Record it in the LFSCK trace file for further checking in
2129          * the second-stage scanning. */
2130         if (!update && !dec && rc == 0)
2131                 lfsck_namespace_trace_update(env, com, cfid,
2132                                              LNTF_CHECK_LINKEA, true);
2133
2134 unlock1:
2135         /* It is harmless even if unlock the unused lock_handle */
2136         lfsck_ibits_unlock(&lh, LCK_PW);
2137         lfsck_unlock(llh);
2138
2139 log:
2140         CDEBUG(D_LFSCK, "%s: namespace LFSCK assistant found bad name "
2141                "entry for: parent "DFID", child "DFID", name %s, type "
2142                "in name entry %o, type claimed by child %o. repair it "
2143                "by %s with new name2 %s: rc = %d\n",
2144                lfsck_lfsck2name(lfsck), PFID(pfid), PFID(cfid),
2145                name, type, update ? lfsck_object_type(child) : 0,
2146                update ? "updating" : "removing", name2, rc);
2147
2148         if (rc != 0) {
2149                 struct lfsck_namespace *ns = com->lc_file_ram;
2150
2151                 ns->ln_flags |= LF_INCONSISTENT;
2152         }
2153
2154         return rc;
2155 }
2156
2157 /**
2158  * Update the ".." name entry for the given object.
2159  *
2160  * The object's ".." is corrupted, this function will update the ".." name
2161  * entry with the given pfid, and the linkEA with the given ldata.
2162  *
2163  * The caller should take the ldlm lock before the calling.
2164  *
2165  * \param[in] env       pointer to the thread context
2166  * \param[in] com       pointer to the lfsck component
2167  * \param[in] obj       pointer to the dt_object to be handled
2168  * \param[in] pfid      the new fid for the object's ".." name entry
2169  * \param[in] cname     the name for the @obj in the parent directory
2170  *
2171  * \retval              positive number for repaired cases
2172  * \retval              0 if nothing to be repaired
2173  * \retval              negative error number on failure
2174  */
2175 static int lfsck_namespace_repair_unmatched_pairs(const struct lu_env *env,
2176                                                   struct lfsck_component *com,
2177                                                   struct dt_object *obj,
2178                                                   const struct lu_fid *pfid,
2179                                                   struct lu_name *cname)
2180 {
2181         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2182         struct dt_insert_rec            *rec    = &info->lti_dt_rec;
2183         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2184         struct dt_device                *dev    = lfsck_obj2dev(obj);
2185         struct thandle                  *th     = NULL;
2186         struct linkea_data               ldata  = { NULL };
2187         struct lu_buf                    linkea_buf;
2188         int                              rc     = 0;
2189         ENTRY;
2190
2191         LASSERT(!dt_object_remote(obj));
2192         LASSERT(S_ISDIR(lfsck_object_type(obj)));
2193
2194         rc = linkea_links_new(&ldata, &info->lti_big_buf, cname, pfid);
2195         if (rc != 0)
2196                 GOTO(log, rc);
2197
2198         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
2199                        ldata.ld_leh->leh_len);
2200
2201         th = dt_trans_create(env, dev);
2202         if (IS_ERR(th))
2203                 GOTO(log, rc = PTR_ERR(th));
2204
2205         rc = dt_declare_delete(env, obj, (const struct dt_key *)dotdot, th);
2206         if (rc != 0)
2207                 GOTO(stop, rc);
2208
2209         rec->rec_type = S_IFDIR;
2210         rec->rec_fid = pfid;
2211         rc = dt_declare_insert(env, obj, (const struct dt_rec *)rec,
2212                                (const struct dt_key *)dotdot, th);
2213         if (rc != 0)
2214                 GOTO(stop, rc);
2215
2216         rc = dt_declare_xattr_set(env, obj, &linkea_buf,
2217                                   XATTR_NAME_LINK, 0, th);
2218         if (rc != 0)
2219                 GOTO(stop, rc);
2220
2221         rc = dt_trans_start_local(env, dev, th);
2222         if (rc != 0)
2223                 GOTO(stop, rc);
2224
2225         dt_write_lock(env, obj, 0);
2226         if (unlikely(lfsck_is_dead_obj(obj)))
2227                 GOTO(unlock, rc = 0);
2228
2229         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
2230                 GOTO(unlock, rc = 1);
2231
2232         /* The old ".." name entry maybe not exist. */
2233         dt_delete(env, obj, (const struct dt_key *)dotdot, th);
2234
2235         rc = dt_insert(env, obj, (const struct dt_rec *)rec,
2236                        (const struct dt_key *)dotdot, th, 1);
2237         if (rc != 0)
2238                 GOTO(unlock, rc);
2239
2240         rc = lfsck_links_write(env, obj, &ldata, th);
2241
2242         GOTO(unlock, rc = (rc == 0 ? 1 : rc));
2243
2244 unlock:
2245         dt_write_unlock(env, obj);
2246
2247 stop:
2248         dt_trans_stop(env, dev, th);
2249
2250 log:
2251         CDEBUG(D_LFSCK, "%s: namespace LFSCK rebuild dotdot name entry for "
2252                "the object "DFID", new parent "DFID": rc = %d\n",
2253                lfsck_lfsck2name(lfsck), PFID(lfsck_dto2fid(obj)),
2254                PFID(pfid), rc);
2255
2256         if (rc != 0) {
2257                 struct lfsck_namespace *ns = com->lc_file_ram;
2258
2259                 ns->ln_flags |= LF_INCONSISTENT;
2260         }
2261
2262         return rc;
2263 }
2264
2265 /**
2266  * Handle orphan @obj during Double Scan Directory.
2267  *
2268  * Remove the @obj's current (invalid) linkEA entries, and insert
2269  * it in the directory .lustre/lost+found/MDTxxxx/ with the name:
2270  * ${FID}-${PFID}-D-${conflict_version}
2271  *
2272  * The caller should take the ldlm lock before the calling.
2273  *
2274  * \param[in] env       pointer to the thread context
2275  * \param[in] com       pointer to the lfsck component
2276  * \param[in] obj       pointer to the orphan object to be handled
2277  * \param[in] pfid      the new fid for the object's ".." name entry
2278  * \param[in,out] lh    ldlm lock handler for the given @obj
2279  * \param[out] type     to tell the caller what the inconsistency is
2280  *
2281  * \retval              positive number for repaired cases
2282  * \retval              0 if nothing to be repaired
2283  * \retval              negative error number on failure
2284  */
2285 static int
2286 lfsck_namespace_dsd_orphan(const struct lu_env *env,
2287                            struct lfsck_component *com,
2288                            struct dt_object *obj,
2289                            const struct lu_fid *pfid,
2290                            struct lustre_handle *lh,
2291                            enum lfsck_namespace_inconsistency_type *type)
2292 {
2293         struct lfsck_thread_info *info = lfsck_env_info(env);
2294         struct lfsck_namespace   *ns   = com->lc_file_ram;
2295         int                       rc;
2296         ENTRY;
2297
2298         /* Remove the unrecognized linkEA. */
2299         rc = lfsck_namespace_links_remove(env, com, obj);
2300         lfsck_ibits_unlock(lh, LCK_EX);
2301         if (rc < 0 && rc != -ENODATA)
2302                 RETURN(rc);
2303
2304         *type = LNIT_MUL_REF;
2305
2306         /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT has
2307          * ever tried to verify some remote MDT-object that resides on this
2308          * MDT, but this MDT failed to respond such request. So means there
2309          * may be some remote name entry on other MDT that references this
2310          * object with another name, so we cannot know whether this linkEA
2311          * is valid or not. So keep it there and maybe resolved when next
2312          * LFSCK run. */
2313         if (ns->ln_flags & LF_INCOMPLETE)
2314                 RETURN(0);
2315
2316         /* The unique linkEA is invalid, even if the ".." name entry may be
2317          * valid, we still cannot know via which name entry this directory
2318          * will be referenced. Then handle it as pure orphan. */
2319         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2320                  "-"DFID, PFID(pfid));
2321         rc = lfsck_namespace_insert_orphan(env, com, obj,
2322                                            info->lti_tmpbuf, "D", NULL);
2323
2324         RETURN(rc);
2325 }
2326
2327 /**
2328  * Double Scan Directory object for single linkEA entry case.
2329  *
2330  * The given @child has unique linkEA entry. If the linkEA entry is valid,
2331  * then check whether the name is in the namespace or not, if not, add the
2332  * missing name entry back to namespace. If the linkEA entry is invalid,
2333  * then remove it and insert the @child in the .lustre/lost+found/MDTxxxx/
2334  * as an orphan.
2335  *
2336  * \param[in] env       pointer to the thread context
2337  * \param[in] com       pointer to the lfsck component
2338  * \param[in] child     pointer to the directory to be double scanned
2339  * \param[in] pfid      the FID corresponding to the ".." entry
2340  * \param[in] ldata     pointer to the linkEA data for the given @child
2341  * \param[in,out] lh    ldlm lock handler for the given @child
2342  * \param[out] type     to tell the caller what the inconsistency is
2343  * \param[in] retry     if found inconsistency, but the caller does not hold
2344  *                      ldlm lock on the @child, then set @retry as true
2345  * \param[in] unknown   set if does not know how to repair the inconsistency
2346  *
2347  * \retval              positive number for repaired cases
2348  * \retval              0 if nothing to be repaired
2349  * \retval              negative error number on failure
2350  */
2351 static int
2352 lfsck_namespace_dsd_single(const struct lu_env *env,
2353                            struct lfsck_component *com,
2354                            struct dt_object *child,
2355                            const struct lu_fid *pfid,
2356                            struct linkea_data *ldata,
2357                            struct lustre_handle *lh,
2358                            enum lfsck_namespace_inconsistency_type *type,
2359                            bool *retry, bool *unknown)
2360 {
2361         struct lfsck_thread_info *info          = lfsck_env_info(env);
2362         struct lu_name           *cname         = &info->lti_name;
2363         const struct lu_fid      *cfid          = lfsck_dto2fid(child);
2364         struct lu_fid             tfid;
2365         struct lfsck_namespace   *ns            = com->lc_file_ram;
2366         struct lfsck_instance    *lfsck         = com->lc_lfsck;
2367         struct dt_object         *parent        = NULL;
2368         struct lmv_mds_md_v1     *lmv;
2369         int                       rc            = 0;
2370         ENTRY;
2371
2372         rc = lfsck_namespace_unpack_linkea_entry(ldata, cname, &tfid,
2373                                                  info->lti_key,
2374                                                  sizeof(info->lti_key));
2375         /* The unique linkEA entry with bad parent will be handled as orphan. */
2376         if (rc != 0) {
2377                 if (!lustre_handle_is_used(lh) && retry != NULL)
2378                         *retry = true;
2379                 else
2380                         rc = lfsck_namespace_dsd_orphan(env, com, child,
2381                                                         pfid, lh, type);
2382
2383                 GOTO(out, rc);
2384         }
2385
2386         parent = lfsck_object_find_bottom(env, lfsck, &tfid);
2387         if (IS_ERR(parent))
2388                 GOTO(out, rc = PTR_ERR(parent));
2389
2390         /* We trust the unique linkEA entry in spite of whether it matches the
2391          * ".." name entry or not. Because even if the linkEA entry is wrong
2392          * and the ".." name entry is right, we still cannot know via which
2393          * name entry the child will be referenced, since all known entries
2394          * have been verified during the first-stage scanning. */
2395         if (!dt_object_exists(parent)) {
2396                 /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT
2397                  * has ever tried to verify some remote MDT-object that resides
2398                  * on this MDT, but this MDT failed to respond such request. So
2399                  * means there may be some remote name entry on other MDT that
2400                  * references this object with another name, so we cannot know
2401                  * whether this linkEA is valid or not. So keep it there and
2402                  * maybe resolved when next LFSCK run. */
2403                 if (ns->ln_flags & LF_INCOMPLETE)
2404                         GOTO(out, rc = 0);
2405
2406                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2407                         *retry = true;
2408
2409                         GOTO(out, rc = 0);
2410                 }
2411
2412                 lfsck_ibits_unlock(lh, LCK_EX);
2413
2414 lost_parent:
2415                 lmv = &info->lti_lmv;
2416                 rc = lfsck_read_stripe_lmv(env, child, lmv);
2417                 if (rc != 0 && rc != -ENODATA)
2418                         GOTO(out, rc);
2419
2420                 if (rc == -ENODATA || lmv->lmv_magic != LMV_MAGIC_STRIPE) {
2421                         lmv = NULL;
2422                 } else if (lfsck_shard_name_to_index(env,
2423                                         cname->ln_name, cname->ln_namelen,
2424                                         S_IFDIR, cfid) < 0) {
2425                         /* It is an invalid name entry, we
2426                          * cannot trust the parent also. */
2427                         rc = lfsck_namespace_shrink_linkea(env, com, child,
2428                                                 ldata, cname, &tfid, true);
2429                         if (rc < 0)
2430                                 GOTO(out, rc);
2431
2432                         snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2433                                  "-"DFID, PFID(pfid));
2434                         rc = lfsck_namespace_insert_orphan(env, com, child,
2435                                                 info->lti_tmpbuf, "S", NULL);
2436
2437                         GOTO(out, rc);
2438                 }
2439
2440                 /* Create the lost parent as an orphan. */
2441                 rc = lfsck_namespace_create_orphan_dir(env, com, parent, lmv);
2442                 if (rc >= 0) {
2443                         /* Add the missing name entry to the parent. */
2444                         rc = lfsck_namespace_insert_normal(env, com, parent,
2445                                                         child, cname->ln_name);
2446                         if (unlikely(rc == -EEXIST)) {
2447                                 /* Unfortunately, someone reused the name
2448                                  * under the parent by race. So we have
2449                                  * to remove the linkEA entry from
2450                                  * current child object. It means that the
2451                                  * LFSCK cannot recover the system
2452                                  * totally back to its original status,
2453                                  * but it is necessary to make the
2454                                  * current system to be consistent. */
2455                                 rc = lfsck_namespace_shrink_linkea(env,
2456                                                 com, child, ldata,
2457                                                 cname, &tfid, true);
2458                                 if (rc >= 0) {
2459                                         snprintf(info->lti_tmpbuf,
2460                                                  sizeof(info->lti_tmpbuf),
2461                                                  "-"DFID, PFID(pfid));
2462                                         rc = lfsck_namespace_insert_orphan(env,
2463                                                 com, child, info->lti_tmpbuf,
2464                                                 "D", NULL);
2465                                 }
2466                         }
2467                 }
2468
2469                 GOTO(out, rc);
2470         } /* !dt_object_exists(parent) */
2471
2472         /* The unique linkEA entry with bad parent will be handled as orphan. */
2473         if (unlikely(!dt_try_as_dir(env, parent))) {
2474                 if (!lustre_handle_is_used(lh) && retry != NULL)
2475                         *retry = true;
2476                 else
2477                         rc = lfsck_namespace_dsd_orphan(env, com, child,
2478                                                         pfid, lh, type);
2479
2480                 GOTO(out, rc);
2481         }
2482
2483         rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
2484                        (const struct dt_key *)cname->ln_name);
2485         if (rc == -ENOENT) {
2486                 /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT
2487                  * has ever tried to verify some remote MDT-object that resides
2488                  * on this MDT, but this MDT failed to respond such request. So
2489                  * means there may be some remote name entry on other MDT that
2490                  * references this object with another name, so we cannot know
2491                  * whether this linkEA is valid or not. So keep it there and
2492                  * maybe resolved when next LFSCK run. */
2493                 if (ns->ln_flags & LF_INCOMPLETE)
2494                         GOTO(out, rc = 0);
2495
2496                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2497                         *retry = true;
2498
2499                         GOTO(out, rc = 0);
2500                 }
2501
2502                 lfsck_ibits_unlock(lh, LCK_EX);
2503                 rc = lfsck_namespace_check_name(env, parent, child, cname);
2504                 if (rc == -ENOENT)
2505                         goto lost_parent;
2506
2507                 if (rc < 0)
2508                         GOTO(out, rc);
2509
2510                 /* It is an invalid name entry, drop it. */
2511                 if (unlikely(rc > 0)) {
2512                         rc = lfsck_namespace_shrink_linkea(env, com, child,
2513                                                 ldata, cname, &tfid, true);
2514                         if (rc >= 0) {
2515                                 snprintf(info->lti_tmpbuf,
2516                                          sizeof(info->lti_tmpbuf),
2517                                          "-"DFID, PFID(pfid));
2518                                 rc = lfsck_namespace_insert_orphan(env, com,
2519                                         child, info->lti_tmpbuf, "D", NULL);
2520                         }
2521
2522                         GOTO(out, rc);
2523                 }
2524
2525                 /* Add the missing name entry back to the namespace. */
2526                 rc = lfsck_namespace_insert_normal(env, com, parent, child,
2527                                                    cname->ln_name);
2528                 if (unlikely(rc == -ESTALE))
2529                         /* It may happen when the remote object has been
2530                          * removed, but the local MDT is not aware of that. */
2531                         goto lost_parent;
2532
2533                 if (unlikely(rc == -EEXIST)) {
2534                         /* Unfortunately, someone reused the name under the
2535                          * parent by race. So we have to remove the linkEA
2536                          * entry from current child object. It means that the
2537                          * LFSCK cannot recover the system totally back to
2538                          * its original status, but it is necessary to make
2539                          * the current system to be consistent.
2540                          *
2541                          * It also may be because of the LFSCK found some
2542                          * internal status of create operation. Under such
2543                          * case, nothing to be done. */
2544                         rc = lfsck_namespace_shrink_linkea_cond(env, com,
2545                                         parent, child, ldata, cname, &tfid);
2546                         if (rc >= 0) {
2547                                 snprintf(info->lti_tmpbuf,
2548                                          sizeof(info->lti_tmpbuf),
2549                                          "-"DFID, PFID(pfid));
2550                                 rc = lfsck_namespace_insert_orphan(env, com,
2551                                         child, info->lti_tmpbuf, "D", NULL);
2552                         }
2553                 }
2554
2555                 GOTO(out, rc);
2556         } /* rc == -ENOENT */
2557
2558         if (rc != 0)
2559                 GOTO(out, rc);
2560
2561         if (!lu_fid_eq(&tfid, cfid)) {
2562                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2563                         *retry = true;
2564
2565                         GOTO(out, rc = 0);
2566                 }
2567
2568                 lfsck_ibits_unlock(lh, LCK_EX);
2569                 /* The name entry references another MDT-object that
2570                  * may be created by the LFSCK for repairing dangling
2571                  * name entry. Try to replace it. */
2572                 rc = lfsck_namespace_replace_cond(env, com, parent, child,
2573                                                   &tfid, cname);
2574                 if (rc == 0)
2575                         rc = lfsck_namespace_dsd_orphan(env, com, child,
2576                                                         pfid, lh, type);
2577
2578                 GOTO(out, rc);
2579         }
2580
2581         /* Zero FID may because the remote directroy object has invalid linkEA,
2582          * or lost linkEA. Under such case, the LFSCK on this MDT does not know
2583          * how to repair the inconsistency, but the namespace LFSCK on the MDT
2584          * where its name entry resides may has more information (name, FID) to
2585          * repair such inconsistency. So here, keep the inconsistency to avoid
2586          * some imporper repairing. */
2587         if (fid_is_zero(pfid)) {
2588                 if (unknown)
2589                         *unknown = true;
2590
2591                 GOTO(out, rc = 0);
2592         }
2593
2594         /* The ".." name entry is wrong, update it. */
2595         if (!lu_fid_eq(pfid, lfsck_dto2fid(parent))) {
2596                 if (!lustre_handle_is_used(lh) && retry != NULL) {
2597                         *retry = true;
2598
2599                         GOTO(out, rc = 0);
2600                 }
2601
2602                 *type = LNIT_UNMATCHED_PAIRS;
2603                 rc = lfsck_namespace_repair_unmatched_pairs(env, com, child,
2604                                                 lfsck_dto2fid(parent), cname);
2605         }
2606
2607         GOTO(out, rc);
2608
2609 out:
2610         if (parent != NULL && !IS_ERR(parent))
2611                 lfsck_object_put(env, parent);
2612
2613         return rc;
2614 }
2615
2616 /**
2617  * Double Scan Directory object for multiple linkEA entries case.
2618  *
2619  * The given @child has multiple linkEA entries. There is at most one linkEA
2620  * entry will be valid, all the others will be removed. Firstly, the function
2621  * will try to find out the linkEA entry for which the name entry exists under
2622  * the given parent (@pfid). If there is no linkEA entry that matches the given
2623  * ".." name entry, then tries to find out the first linkEA entry that both the
2624  * parent and the name entry exist to rebuild a new ".." name entry.
2625  *
2626  * \param[in] env       pointer to the thread context
2627  * \param[in] com       pointer to the lfsck component
2628  * \param[in] child     pointer to the directory to be double scanned
2629  * \param[in] pfid      the FID corresponding to the ".." entry
2630  * \param[in] ldata     pointer to the linkEA data for the given @child
2631  * \param[in,out] lh    ldlm lock handler for the given @child
2632  * \param[out] type     to tell the caller what the inconsistency is
2633  * \param[in] lpf       true if the ".." entry is under lost+found/MDTxxxx/
2634  * \param[in] unknown   set if does not know how to repair the inconsistency
2635  *
2636  * \retval              positive number for repaired cases
2637  * \retval              0 if nothing to be repaired
2638  * \retval              negative error number on failure
2639  */
2640 static int
2641 lfsck_namespace_dsd_multiple(const struct lu_env *env,
2642                              struct lfsck_component *com,
2643                              struct dt_object *child,
2644                              const struct lu_fid *pfid,
2645                              struct linkea_data *ldata,
2646                              struct lustre_handle *lh,
2647                              enum lfsck_namespace_inconsistency_type *type,
2648                              bool lpf, bool *unknown)
2649 {
2650         struct lfsck_thread_info *info          = lfsck_env_info(env);
2651         struct lu_name           *cname         = &info->lti_name;
2652         const struct lu_fid      *cfid          = lfsck_dto2fid(child);
2653         struct lu_fid            *pfid2         = &info->lti_fid3;
2654         struct lu_fid             tfid;
2655         struct lfsck_namespace   *ns            = com->lc_file_ram;
2656         struct lfsck_instance    *lfsck         = com->lc_lfsck;
2657         struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
2658         struct dt_object         *parent        = NULL;
2659         struct linkea_data        ldata_new     = { NULL };
2660         int                       dirent_count  = 0;
2661         int                       rc            = 0;
2662         bool                      once          = true;
2663         ENTRY;
2664
2665 again:
2666         while (ldata->ld_lee != NULL) {
2667                 rc = lfsck_namespace_unpack_linkea_entry(ldata, cname, &tfid,
2668                                                          info->lti_key,
2669                                                          sizeof(info->lti_key));
2670                 /* Drop invalid linkEA entry. */
2671                 if (rc != 0) {
2672                         lfsck_linkea_del_buf(ldata, cname);
2673                         continue;
2674                 }
2675
2676                 /* Drop repeated linkEA entries. */
2677                 lfsck_namespace_filter_linkea_entry(ldata, cname, &tfid, true);
2678
2679                 /* If current dotdot is the .lustre/lost+found/MDTxxxx/,
2680                  * then it is possible that: the directry object has ever
2681                  * been lost, but its name entry was there. In the former
2682                  * LFSCK run, during the first-stage scanning, the LFSCK
2683                  * found the dangling name entry, but it did not recreate
2684                  * the lost object, and when moved to the second-stage
2685                  * scanning, some children objects of the lost directory
2686                  * object were found, then the LFSCK recreated such lost
2687                  * directory object as an orphan.
2688                  *
2689                  * When the LFSCK runs again, if the dangling name is still
2690                  * there, the LFSCK should move the orphan directory object
2691                  * back to the normal namespace. */
2692                 if (!lpf && !fid_is_zero(pfid) &&
2693                     !lu_fid_eq(pfid, &tfid) && once) {
2694                         linkea_next_entry(ldata);
2695                         continue;
2696                 }
2697
2698                 parent = lfsck_object_find_bottom(env, lfsck, &tfid);
2699                 if (IS_ERR(parent))
2700                         RETURN(PTR_ERR(parent));
2701
2702                 if (!dt_object_exists(parent)) {
2703                         lfsck_object_put(env, parent);
2704                         if (ldata->ld_leh->leh_reccount > 1) {
2705                                 /* If it is NOT the last linkEA entry, then
2706                                  * there is still other chance to make the
2707                                  * child to be visible via other parent, then
2708                                  * remove this linkEA entry. */
2709                                 lfsck_linkea_del_buf(ldata, cname);
2710                                 continue;
2711                         }
2712
2713                         break;
2714                 }
2715
2716                 /* The linkEA entry with bad parent will be removed. */
2717                 if (unlikely(!dt_try_as_dir(env, parent))) {
2718                         lfsck_object_put(env, parent);
2719                         lfsck_linkea_del_buf(ldata, cname);
2720                         continue;
2721                 }
2722
2723                 rc = dt_lookup(env, parent, (struct dt_rec *)&tfid,
2724                                (const struct dt_key *)cname->ln_name);
2725                 *pfid2 = *lfsck_dto2fid(parent);
2726                 if (rc == -ENOENT) {
2727                         lfsck_object_put(env, parent);
2728                         linkea_next_entry(ldata);
2729                         continue;
2730                 }
2731
2732                 if (rc != 0) {
2733                         lfsck_object_put(env, parent);
2734
2735                         RETURN(rc);
2736                 }
2737
2738                 if (lu_fid_eq(&tfid, cfid)) {
2739                         lfsck_object_put(env, parent);
2740                         /* If the parent (that is declared via linkEA entry)
2741                          * directory contains the specified child, but such
2742                          * parent does not match the dotdot name entry, then
2743                          * trust the linkEA. */
2744                         if (!fid_is_zero(pfid) && !lu_fid_eq(pfid, pfid2)) {
2745                                 *type = LNIT_UNMATCHED_PAIRS;
2746                                 rc = lfsck_namespace_repair_unmatched_pairs(env,
2747                                                 com, child, pfid2, cname);
2748
2749                                 RETURN(rc);
2750                         }
2751
2752 rebuild:
2753                         /* It is the most common case that we find the
2754                          * name entry corresponding to the linkEA entry
2755                          * that matches the ".." name entry. */
2756                         rc = linkea_links_new(&ldata_new, &info->lti_big_buf,
2757                                               cname, pfid2);
2758                         if (rc != 0)
2759                                 RETURN(rc);
2760
2761                         rc = lfsck_namespace_rebuild_linkea(env, com, child,
2762                                                             &ldata_new);
2763                         if (rc < 0)
2764                                 RETURN(rc);
2765
2766                         lfsck_linkea_del_buf(ldata, cname);
2767                         linkea_first_entry(ldata);
2768                         /* There may be some invalid dangling name entries under
2769                          * other parent directories, remove all of them. */
2770                         while (ldata->ld_lee != NULL) {
2771                                 rc = lfsck_namespace_unpack_linkea_entry(ldata,
2772                                                 cname, &tfid, info->lti_key,
2773                                                 sizeof(info->lti_key));
2774                                 if (rc != 0)
2775                                         goto next;
2776
2777                                 parent = lfsck_object_find_bottom(env, lfsck,
2778                                                                   &tfid);
2779                                 if (IS_ERR(parent)) {
2780                                         rc = PTR_ERR(parent);
2781                                         if (rc != -ENOENT &&
2782                                             bk->lb_param & LPF_FAILOUT)
2783                                                 RETURN(rc);
2784
2785                                         goto next;
2786                                 }
2787
2788                                 if (!dt_object_exists(parent)) {
2789                                         lfsck_object_put(env, parent);
2790                                         goto next;
2791                                 }
2792
2793                                 rc = lfsck_namespace_repair_dirent(env, com,
2794                                         parent, child, cname->ln_name,
2795                                         cname->ln_name, S_IFDIR, false, true);
2796                                 lfsck_object_put(env, parent);
2797                                 if (rc < 0) {
2798                                         if (bk->lb_param & LPF_FAILOUT)
2799                                                 RETURN(rc);
2800
2801                                         goto next;
2802                                 }
2803
2804                                 dirent_count += rc;
2805
2806 next:
2807                                 lfsck_linkea_del_buf(ldata, cname);
2808                         }
2809
2810                         ns->ln_dirent_repaired += dirent_count;
2811
2812                         RETURN(rc);
2813                 } /* lu_fid_eq(&tfid, lfsck_dto2fid(child)) */
2814
2815                 lfsck_ibits_unlock(lh, LCK_EX);
2816                 /* The name entry references another MDT-object that may be
2817                  * created by the LFSCK for repairing dangling name entry.
2818                  * Try to replace it. */
2819                 rc = lfsck_namespace_replace_cond(env, com, parent, child,
2820                                                   &tfid, cname);
2821                 lfsck_object_put(env, parent);
2822                 if (rc < 0)
2823                         RETURN(rc);
2824
2825                 if (rc > 0)
2826                         goto rebuild;
2827
2828                 lfsck_linkea_del_buf(ldata, cname);
2829         } /* while (ldata->ld_lee != NULL) */
2830
2831         /* If there is still linkEA overflow, return. */
2832         if (unlikely(ldata->ld_leh->leh_overflow_time))
2833                 RETURN(0);
2834
2835         linkea_first_entry(ldata);
2836         if (ldata->ld_leh->leh_reccount == 1) {
2837                 rc = lfsck_namespace_dsd_single(env, com, child, pfid, ldata,
2838                                                 lh, type, NULL, unknown);
2839
2840                 RETURN(rc);
2841         }
2842
2843         /* All linkEA entries are invalid and removed, then handle the @child
2844          * as an orphan.*/
2845         if (ldata->ld_leh->leh_reccount == 0) {
2846                 rc = lfsck_namespace_dsd_orphan(env, com, child, pfid, lh,
2847                                                 type);
2848
2849                 RETURN(rc);
2850         }
2851
2852         /* If the dangling name entry for the orphan directory object has
2853          * been remvoed, then just check whether the directory object is
2854          * still under the .lustre/lost+found/MDTxxxx/ or not. */
2855         if (lpf) {
2856                 lpf = false;
2857                 goto again;
2858         }
2859
2860         /* There is no linkEA entry that matches the ".." name entry. Find
2861          * the first linkEA entry that both parent and name entry exist to
2862          * rebuild a new ".." name entry. */
2863         if (once) {
2864                 once = false;
2865                 goto again;
2866         }
2867
2868         RETURN(rc);
2869 }
2870
2871 /**
2872  * Repair the object's nlink attribute.
2873  *
2874  * If all the known name entries have been verified, then the object's hard
2875  * link attribute should match the object's linkEA entries count unless the
2876  * object's has too many hard link to be recorded in the linkEA. Such cases
2877  * should have been marked in the LFSCK trace file. Otherwise, trust the
2878  * linkEA to update the object's nlink attribute.
2879  *
2880  * \param[in] env       pointer to the thread context
2881  * \param[in] com       pointer to the lfsck component
2882  * \param[in] obj       pointer to the dt_object to be handled
2883  * \param[in,out] la    pointer to buffer to object's attribute before
2884  *                      and after the repairing
2885  *
2886  * \retval              positive number for repaired cases
2887  * \retval              0 if nothing to be repaired
2888  * \retval              negative error number on failure
2889  */
2890 static int lfsck_namespace_repair_nlink(const struct lu_env *env,
2891                                         struct lfsck_component *com,
2892                                         struct dt_object *obj,
2893                                         struct lu_attr *la)
2894 {
2895         struct lfsck_namespace          *ns     = com->lc_file_ram;
2896         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2897         struct dt_device                *dev    = lfsck_obj2dev(obj);
2898         const struct lu_fid             *cfid   = lfsck_dto2fid(obj);
2899         struct thandle                  *th     = NULL;
2900         struct linkea_data               ldata  = { NULL };
2901         struct lustre_handle             lh     = { 0 };
2902         __u32                            old    = la->la_nlink;
2903         int                              rc     = 0;
2904         ENTRY;
2905
2906         LASSERT(!dt_object_remote(obj));
2907
2908         rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
2909                               MDS_INODELOCK_UPDATE, LCK_PW);
2910         if (rc != 0)
2911                 GOTO(log, rc);
2912
2913         th = dt_trans_create(env, dev);
2914         if (IS_ERR(th))
2915                 GOTO(log, rc = PTR_ERR(th));
2916
2917         la->la_valid = LA_NLINK;
2918         rc = dt_declare_attr_set(env, obj, la, th);
2919         if (rc != 0)
2920                 GOTO(stop, rc);
2921
2922         rc = dt_trans_start_local(env, dev, th);
2923         if (rc != 0)
2924                 GOTO(stop, rc);
2925
2926         dt_write_lock(env, obj, 0);
2927         /* If the LFSCK is marked as LF_INCOMPLETE, then means some MDT has
2928          * ever tried to verify some remote MDT-object that resides on this
2929          * MDT, but this MDT failed to respond such request. So means there
2930          * may be some remote name entry on other MDT that references this
2931          * object with another name, so we cannot know whether this linkEA
2932          * is valid or not. So keep it there and maybe resolved when next
2933          * LFSCK run. */
2934         if (ns->ln_flags & LF_INCOMPLETE)
2935                 GOTO(unlock, rc = 0);
2936
2937         rc = dt_attr_get(env, obj, la);
2938         if (rc != 0)
2939                 GOTO(unlock, rc = (rc == -ENOENT ? 0 : rc));
2940
2941         rc = lfsck_links_read2_with_rec(env, obj, &ldata);
2942         if (rc)
2943                 GOTO(unlock, rc = (rc == -ENODATA ? 0 : rc));
2944
2945         /* XXX: Currently, we only update the nlink attribute if the known
2946          *      linkEA entries is larger than the nlink attribute. That is
2947          *      safe action. */
2948         if (la->la_nlink >= ldata.ld_leh->leh_reccount ||
2949             unlikely(la->la_nlink == 0 ||
2950                      ldata.ld_leh->leh_overflow_time))
2951                 GOTO(unlock, rc = 0);
2952
2953         la->la_nlink = ldata.ld_leh->leh_reccount;
2954         if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
2955                 GOTO(unlock, rc = 1);
2956
2957         rc = dt_attr_set(env, obj, la, th);
2958
2959         GOTO(unlock, rc = (rc == 0 ? 1 : rc));
2960
2961 unlock:
2962         dt_write_unlock(env, obj);
2963
2964 stop:
2965         dt_trans_stop(env, dev, th);
2966
2967 log:
2968         lfsck_ibits_unlock(&lh, LCK_PW);
2969         CDEBUG(D_LFSCK, "%s: namespace LFSCK repaired the object "DFID"'s "
2970                "nlink count from %u to %u: rc = %d\n",
2971                lfsck_lfsck2name(lfsck), PFID(cfid), old, la->la_nlink, rc);
2972
2973         if (rc != 0)
2974                 ns->ln_flags |= LF_INCONSISTENT;
2975
2976         return rc;
2977 }
2978
2979 /**
2980  * Double scan the directory object for namespace LFSCK.
2981  *
2982  * This function will verify the <parent, child> pairs in the namespace tree:
2983  * the parent references the child via some name entry that should be in the
2984  * child's linkEA entry, the child should back references the parent via its
2985  * ".." name entry.
2986  *
2987  * The LFSCK will scan every linkEA entry in turn until find out the first
2988  * matched pairs. If found, then all other linkEA entries will be dropped.
2989  * If all the linkEA entries cannot match the ".." name entry, then there
2990  * are serveral possible cases:
2991  *
2992  * 1) If there is only one linkEA entry, then trust it as long as the PFID
2993  *    in the linkEA entry is valid.
2994  *
2995  * 2) If there are multiple linkEA entries, then try to find the linkEA
2996  *    that matches the ".." name entry. If found, then all other entries
2997  *    are invalid; otherwise, it is quite possible that the ".." name entry
2998  *    is corrupted. Under such case, the LFSCK will rebuild the ".." name
2999  *    entry according to the first valid linkEA entry (both the parent and
3000  *    the name entry should exist).
3001  *
3002  * 3) If the directory object has no (valid) linkEA entry, then the
3003  *    directory object will be handled as pure orphan and inserted
3004  *    in the .lustre/lost+found/MDTxxxx/ with the name:
3005  *    ${self_FID}-${PFID}-D-${conflict_version}
3006  *
3007  * \param[in] env       pointer to the thread context
3008  * \param[in] com       pointer to the lfsck component
3009  * \param[in] child     pointer to the directory object to be handled
3010  * \param[in] flags     to indicate the specical checking on the @child
3011  *
3012  * \retval              positive number for repaired cases
3013  * \retval              0 if nothing to be repaired
3014  * \retval              negative error number on failure
3015  */
3016 static int lfsck_namespace_double_scan_dir(const struct lu_env *env,
3017                                            struct lfsck_component *com,
3018                                            struct dt_object *child, __u8 flags)
3019 {
3020         struct lfsck_thread_info *info          = lfsck_env_info(env);
3021         const struct lu_fid      *cfid          = lfsck_dto2fid(child);
3022         struct lu_fid            *pfid          = &info->lti_fid2;
3023         struct lfsck_namespace   *ns            = com->lc_file_ram;
3024         struct lfsck_instance    *lfsck         = com->lc_lfsck;
3025         struct lustre_handle      lh            = { 0 };
3026         struct linkea_data        ldata         = { NULL };
3027         bool                      unknown       = false;
3028         bool                      lpf           = false;
3029         bool                      retry         = false;
3030         enum lfsck_namespace_inconsistency_type type = LNIT_BAD_LINKEA;
3031         int                       rc            = 0;
3032         ENTRY;
3033
3034         LASSERT(!dt_object_remote(child));