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