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