Whamcloud - gitweb
9f1ad926be3fb9b7e5799dc20b68a3acc7f1b280
[fs/lustre-release.git] / lustre / lfsck / lfsck_layout.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9
10  * This program is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  * GNU General Public License version 2 for more details.  A copy is
14  * included in the COPYING file that accompanied this code.
15
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2014, 2017, Intel Corporation.
24  */
25 /*
26  * lustre/lfsck/lfsck_layout.c
27  *
28  * Author: Fan, Yong <fan.yong@intel.com>
29  */
30
31 #ifndef EXPORT_SYMTAB
32 # define EXPORT_SYMTAB
33 #endif
34 #define DEBUG_SUBSYSTEM S_LFSCK
35
36 #include <linux/bitops.h>
37 #include <linux/rbtree.h>
38
39 #include <lu_object.h>
40 #include <dt_object.h>
41 #include <lustre_fid.h>
42 #include <lustre_lib.h>
43 #include <lustre_net.h>
44 #include <md_object.h>
45 #include <obd_class.h>
46
47 #include "lfsck_internal.h"
48
49 #define LFSCK_LAYOUT_MAGIC_V1           0xB173AE14
50 #define LFSCK_LAYOUT_MAGIC_V2           0xB1734D76
51 #define LFSCK_LAYOUT_MAGIC_V3           0xB17371B9
52 #define LFSCK_LAYOUT_MAGIC_V4           0xB1732FED
53
54 #define LFSCK_LAYOUT_MAGIC              LFSCK_LAYOUT_MAGIC_V4
55
56 struct lfsck_layout_seq {
57         struct list_head         lls_list;
58         __u64                    lls_seq;
59         __u64                    lls_lastid;
60         __u64                    lls_lastid_known;
61         struct dt_object        *lls_lastid_obj;
62         unsigned int             lls_dirty:1;
63 };
64
65 struct lfsck_layout_slave_target {
66         /* link into lfsck_layout_slave_data::llsd_master_list. */
67         struct list_head        llst_list;
68         /* The position for next record in the rbtree for iteration. */
69         struct lu_fid           llst_fid;
70         /* Dummy hash for iteration against the rbtree. */
71         __u64                   llst_hash;
72         __u64                   llst_gen;
73         atomic_t                llst_ref;
74         __u32                   llst_index;
75         /* How many times we have failed to get the master status. */
76         int                     llst_failures;
77 };
78
79 struct lfsck_layout_slave_data {
80         /* list for lfsck_layout_seq */
81         struct list_head         llsd_seq_list;
82
83         /* list for the masters involve layout verification. */
84         struct list_head         llsd_master_list;
85         spinlock_t               llsd_lock;
86         __u64                    llsd_touch_gen;
87         struct dt_object        *llsd_rb_obj;
88         struct rb_root           llsd_rb_root;
89         struct rw_semaphore      llsd_rb_rwsem;
90         unsigned int             llsd_rbtree_valid:1;
91 };
92
93 struct lfsck_layout_slave_async_args {
94         struct obd_export                *llsaa_exp;
95         struct lfsck_component           *llsaa_com;
96         struct lfsck_layout_slave_target *llsaa_llst;
97 };
98
99 static inline bool lfsck_comp_extent_aligned(__u64 size)
100 {
101          return (size & (LOV_MIN_STRIPE_SIZE - 1)) == 0;
102 }
103
104 static inline void
105 lfsck_layout_llst_put(struct lfsck_layout_slave_target *llst)
106 {
107         if (atomic_dec_and_test(&llst->llst_ref)) {
108                 LASSERT(list_empty(&llst->llst_list));
109
110                 OBD_FREE_PTR(llst);
111         }
112 }
113
114 static inline int
115 lfsck_layout_llst_add(struct lfsck_layout_slave_data *llsd, __u32 index)
116 {
117         struct lfsck_layout_slave_target *llst;
118         struct lfsck_layout_slave_target *tmp;
119         int                               rc   = 0;
120
121         OBD_ALLOC_PTR(llst);
122         if (llst == NULL)
123                 return -ENOMEM;
124
125         INIT_LIST_HEAD(&llst->llst_list);
126         llst->llst_gen = 0;
127         llst->llst_index = index;
128         atomic_set(&llst->llst_ref, 1);
129
130         spin_lock(&llsd->llsd_lock);
131         list_for_each_entry(tmp, &llsd->llsd_master_list, llst_list) {
132                 if (tmp->llst_index == index) {
133                         rc = -EALREADY;
134                         break;
135                 }
136         }
137         if (rc == 0)
138                 list_add_tail(&llst->llst_list, &llsd->llsd_master_list);
139         spin_unlock(&llsd->llsd_lock);
140
141         if (rc != 0)
142                 OBD_FREE_PTR(llst);
143
144         return rc;
145 }
146
147 static inline void
148 lfsck_layout_llst_del(struct lfsck_layout_slave_data *llsd,
149                       struct lfsck_layout_slave_target *llst)
150 {
151         bool del = false;
152
153         spin_lock(&llsd->llsd_lock);
154         if (!list_empty(&llst->llst_list)) {
155                 list_del_init(&llst->llst_list);
156                 del = true;
157         }
158         spin_unlock(&llsd->llsd_lock);
159
160         if (del)
161                 lfsck_layout_llst_put(llst);
162 }
163
164 static inline struct lfsck_layout_slave_target *
165 lfsck_layout_llst_find_and_del(struct lfsck_layout_slave_data *llsd,
166                                __u32 index, bool unlink)
167 {
168         struct lfsck_layout_slave_target *llst;
169
170         spin_lock(&llsd->llsd_lock);
171         list_for_each_entry(llst, &llsd->llsd_master_list, llst_list) {
172                 if (llst->llst_index == index) {
173                         if (unlink)
174                                 list_del_init(&llst->llst_list);
175                         else
176                                 atomic_inc(&llst->llst_ref);
177                         spin_unlock(&llsd->llsd_lock);
178
179                         return llst;
180                 }
181         }
182         spin_unlock(&llsd->llsd_lock);
183
184         return NULL;
185 }
186
187 static struct lfsck_layout_req *
188 lfsck_layout_assistant_req_init(struct lfsck_assistant_object *lso,
189                                 struct dt_object *child, __u32 comp_id,
190                                 __u32 ost_idx, __u32 lov_idx)
191 {
192         struct lfsck_layout_req *llr;
193
194         OBD_ALLOC_PTR(llr);
195         if (llr == NULL)
196                 return ERR_PTR(-ENOMEM);
197
198         INIT_LIST_HEAD(&llr->llr_lar.lar_list);
199         llr->llr_lar.lar_parent = lfsck_assistant_object_get(lso);
200         llr->llr_child = child;
201         llr->llr_comp_id = comp_id;
202         llr->llr_ost_idx = ost_idx;
203         llr->llr_lov_idx = lov_idx;
204
205         return llr;
206 }
207
208 static void lfsck_layout_assistant_req_fini(const struct lu_env *env,
209                                             struct lfsck_assistant_req *lar)
210 {
211         struct lfsck_layout_req *llr =
212                         container_of0(lar, struct lfsck_layout_req, llr_lar);
213
214         lfsck_object_put(env, llr->llr_child);
215         lfsck_assistant_object_put(env, lar->lar_parent);
216         OBD_FREE_PTR(llr);
217 }
218
219 static int
220 lfsck_layout_assistant_sync_failures_interpret(const struct lu_env *env,
221                                                struct ptlrpc_request *req,
222                                                void *args, int rc)
223 {
224         if (rc == 0) {
225                 struct lfsck_async_interpret_args *laia = args;
226                 struct lfsck_tgt_desc             *ltd  = laia->laia_ltd;
227
228                 ltd->ltd_synced_failures = 1;
229                 atomic_dec(laia->laia_count);
230         }
231
232         return 0;
233 }
234
235 /**
236  * Notify remote LFSCK instances about former failures.
237  *
238  * The local LFSCK instance has recorded which OSTs have ever failed to respond
239  * some LFSCK verification requests (maybe because of network issues or the OST
240  * itself trouble). During the respond gap, the OST may missed some OST-objects
241  * verification, then the OST cannot know whether related OST-objects have been
242  * referenced by related MDT-objects or not, then in the second-stage scanning,
243  * these OST-objects will be regarded as orphan, if the OST-object contains bad
244  * parent FID for back reference, then it will misguide the LFSCK to make wrong
245  * fixing for the fake orphan.
246  *
247  * To avoid above trouble, when layout LFSCK finishes the first-stage scanning,
248  * it will scan the bitmap for the ever failed OSTs, and notify them that they
249  * have ever missed some OST-object verification and should skip the handling
250  * for orphan OST-objects on all MDTs that are in the layout LFSCK.
251  *
252  * \param[in] env       pointer to the thread context
253  * \param[in] com       pointer to the lfsck component
254  * \param[in] lr        pointer to the lfsck request
255  */
256 static void lfsck_layout_assistant_sync_failures(const struct lu_env *env,
257                                                  struct lfsck_component *com,
258                                                  struct lfsck_request *lr)
259 {
260         struct lfsck_async_interpret_args *laia  =
261                                 &lfsck_env_info(env)->lti_laia2;
262         struct lfsck_assistant_data       *lad   = com->lc_data;
263         struct lfsck_layout               *lo    = com->lc_file_ram;
264         struct lfsck_instance             *lfsck = com->lc_lfsck;
265         struct lfsck_tgt_descs            *ltds  = &lfsck->li_ost_descs;
266         struct lfsck_tgt_desc             *ltd;
267         struct ptlrpc_request_set         *set;
268         atomic_t                           count;
269         __u32                              idx;
270         int                                rc    = 0;
271         ENTRY;
272
273         if (!test_bit(LAD_INCOMPLETE, &lad->lad_flags))
274                 RETURN_EXIT;
275
276         /* If the MDT has ever failed to verfiy some OST-objects,
277          * then sync failures with them firstly. */
278         lr->lr_flags2 = lo->ll_flags | LF_INCOMPLETE;
279
280         atomic_set(&count, 0);
281         memset(laia, 0, sizeof(*laia));
282         laia->laia_count = &count;
283         set = ptlrpc_prep_set();
284         if (set == NULL)
285                 GOTO(out, rc = -ENOMEM);
286
287         down_read(&ltds->ltd_rw_sem);
288         cfs_foreach_bit(lad->lad_bitmap, idx) {
289                 ltd = lfsck_ltd2tgt(ltds, idx);
290                 if (unlikely(!ltd))
291                         continue;
292
293                 laia->laia_ltd = ltd;
294                 rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
295                                 lfsck_layout_assistant_sync_failures_interpret,
296                                 laia, LFSCK_NOTIFY);
297                 if (rc != 0) {
298                         CDEBUG(D_LFSCK, "%s: LFSCK assistant fail to "
299                                "notify target %x for %s phase1 done: "
300                                "rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
301                                ltd->ltd_index, lad->lad_name, rc);
302
303                         break;
304                 }
305
306                 atomic_inc(&count);
307         }
308         up_read(&ltds->ltd_rw_sem);
309
310         if (rc == 0 && atomic_read(&count) > 0)
311                 rc = ptlrpc_set_wait(env, set);
312
313         ptlrpc_set_destroy(set);
314
315         if (rc == 0 && atomic_read(&count) > 0)
316                 rc = -EINVAL;
317
318         GOTO(out, rc);
319
320 out:
321         if (rc != 0)
322                 /* If failed to sync failures with the OSTs, then have to
323                  * mark the whole LFSCK as LF_INCOMPLETE to skip the whole
324                  * subsequent orphan OST-object handling. */
325                 lo->ll_flags |= LF_INCOMPLETE;
326
327         lr->lr_flags2 = lo->ll_flags;
328 }
329
330 static int lfsck_layout_verify_header_v1v3(struct dt_object *obj,
331                                            struct lov_mds_md_v1 *lmm,
332                                            __u64 start, __u32 comp_id)
333 {
334         __u32 magic;
335         __u32 pattern;
336
337         magic = le32_to_cpu(lmm->lmm_magic);
338         /* If magic crashed, keep it there. Sometime later, during OST-object
339          * orphan handling, if some OST-object(s) back-point to it, it can be
340          * verified and repaired. */
341         if (magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3) {
342                 int rc;
343
344                 if ((magic & LOV_MAGIC_MASK) == LOV_MAGIC_MAGIC)
345                         rc = -EOPNOTSUPP;
346                 else
347                         rc = -EINVAL;
348
349                 CDEBUG(D_LFSCK, "%s LOV EA magic %u for the file "DFID"\n",
350                        rc == -EINVAL ? "Unknown" : "Unsupported",
351                        magic, PFID(lfsck_dto2fid(obj)));
352
353                 return rc;
354         }
355
356         pattern = le32_to_cpu(lmm->lmm_pattern);
357
358 #if 0
359         /* XXX: DoM file verification will be supportted via LU-11081. */
360         if (lov_pattern(pattern) == LOV_PATTERN_MDT) {
361                 if (start != 0) {
362                         CDEBUG(D_LFSCK, "The DoM entry for "DFID" is not "
363                                "the first component in the mirror %x/%llu\n",
364                                PFID(lfsck_dto2fid(obj)), comp_id, start);
365
366                         return -EINVAL;
367                 }
368         }
369 #endif
370
371         if (!lov_pattern_supported_normal_comp(lov_pattern(pattern))) {
372                 CDEBUG(D_LFSCK, "Unsupported LOV EA pattern %u for the file "
373                        DFID" in the component %x\n",
374                        pattern, PFID(lfsck_dto2fid(obj)), comp_id);
375
376                 return -EOPNOTSUPP;
377         }
378
379         return 0;
380 }
381
382 static int lfsck_layout_verify_header_foreign(struct dt_object *obj,
383                                               struct lov_foreign_md *lfm,
384                                               size_t len)
385 {
386         /* magic has been verified already */
387         __u32 value_len = le32_to_cpu(lfm->lfm_length);
388         /* type and flags are not checked for instance */
389
390         CDEBUG(D_INFO, "foreign LOV EA, magic %x, len %u, type %x, flags %x, for file "DFID"\n",
391                le32_to_cpu(lfm->lfm_magic), value_len,
392                le32_to_cpu(lfm->lfm_type), le32_to_cpu(lfm->lfm_flags),
393                PFID(lfsck_dto2fid(obj)));
394
395         if (len != value_len + offsetof(typeof(*lfm), lfm_value))
396                 CDEBUG(D_LFSCK, "foreign LOV EA internal size %u does not match EA full size %zu for file "DFID"\n",
397                        value_len, len, PFID(lfsck_dto2fid(obj)));
398
399         /* nothing to repair */
400         return -ENODATA;
401 }
402
403 static int lfsck_layout_verify_header(struct dt_object *obj,
404                                       struct lov_mds_md_v1 *lmm, size_t len)
405 {
406         int rc = 0;
407
408         if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_COMP_V1) {
409                 struct lov_comp_md_v1 *lcm = (struct lov_comp_md_v1 *)lmm;
410                 int i;
411                 __u16 count = le16_to_cpu(lcm->lcm_entry_count);
412
413                 if (unlikely(count == 0)) {
414                         CDEBUG(D_LFSCK, "the PFL file "DFID" contains invalid "
415                                "components count 0\n",
416                                PFID(lfsck_dto2fid(obj)));
417
418                         return -EINVAL;
419                 }
420
421                 for (i = 0; i < count && !rc; i++) {
422                         struct lov_comp_md_entry_v1 *lcme =
423                                                 &lcm->lcm_entries[i];
424                         __u64 start = le64_to_cpu(lcme->lcme_extent.e_start);
425                         __u64 end = le64_to_cpu(lcme->lcme_extent.e_end);
426                         __u32 comp_id = le32_to_cpu(lcme->lcme_id);
427
428                         if (unlikely(comp_id == LCME_ID_INVAL ||
429                                      comp_id > LCME_ID_MAX)) {
430                                 CDEBUG(D_LFSCK, "found invalid FPL ID %u "
431                                        "for the file "DFID" at idx %d\n",
432                                        comp_id, PFID(lfsck_dto2fid(obj)), i);
433
434                                 return -EINVAL;
435                         }
436
437                         if (unlikely(start >= end ||
438                                      !lfsck_comp_extent_aligned(start) ||
439                                      (!lfsck_comp_extent_aligned(end) &&
440                                       end != LUSTRE_EOF))) {
441                                 CDEBUG(D_LFSCK, "found invalid FPL extent "
442                                        "range [%llu - %llu) for the file "
443                                        DFID" at idx %d\n",
444                                        start, end, PFID(lfsck_dto2fid(obj)), i);
445
446                                 return -EINVAL;
447                         }
448
449                         rc = lfsck_layout_verify_header_v1v3(obj,
450                                         (struct lov_mds_md_v1 *)((char *)lmm +
451                                         le32_to_cpu(lcme->lcme_offset)), start,
452                                         comp_id);
453                 }
454         } else if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_FOREIGN) {
455                 rc = lfsck_layout_verify_header_foreign(obj,
456                                                 (struct lov_foreign_md *)lmm,
457                                                 len);
458         } else {
459                 rc = lfsck_layout_verify_header_v1v3(obj, lmm, 1, 0);
460         }
461
462         return rc;
463 }
464
465 static int lfsck_layout_get_lovea(const struct lu_env *env,
466                                   struct dt_object *obj, struct lu_buf *buf)
467 {
468         int rc;
469         int rc1;
470
471 again:
472         rc = dt_xattr_get(env, obj, buf, XATTR_NAME_LOV);
473         if (rc == -ERANGE) {
474                 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LOV);
475                 if (rc <= 0)
476                         return !rc ? -ENODATA : rc;
477
478                 lu_buf_realloc(buf, rc);
479                 if (buf->lb_buf == NULL)
480                         return -ENOMEM;
481
482                 goto again;
483         }
484
485         if (rc <= 0)
486                 return !rc ? -ENODATA : rc;
487
488         if (unlikely(buf->lb_buf == NULL)) {
489                 lu_buf_alloc(buf, rc);
490                 if (buf->lb_buf == NULL)
491                         return -ENOMEM;
492
493                 goto again;
494         }
495
496         rc1 = lfsck_layout_verify_header(obj, buf->lb_buf, rc);
497
498         return rc1 ? rc1 : rc;
499 }
500
501 #define LFSCK_RBTREE_BITMAP_SIZE        PAGE_SIZE
502 #define LFSCK_RBTREE_BITMAP_WIDTH       (LFSCK_RBTREE_BITMAP_SIZE << 3)
503 #define LFSCK_RBTREE_BITMAP_MASK        (LFSCK_RBTREE_BITMAP_WIDTH - 1)
504
505 struct lfsck_rbtree_node {
506         struct rb_node   lrn_node;
507         __u64            lrn_seq;
508         __u32            lrn_first_oid;
509         atomic_t         lrn_known_count;
510         atomic_t         lrn_accessed_count;
511         void            *lrn_known_bitmap;
512         void            *lrn_accessed_bitmap;
513 };
514
515 static inline int lfsck_rbtree_cmp(struct lfsck_rbtree_node *lrn,
516                                    __u64 seq, __u32 oid)
517 {
518         if (seq < lrn->lrn_seq)
519                 return -1;
520
521         if (seq > lrn->lrn_seq)
522                 return 1;
523
524         if (oid < lrn->lrn_first_oid)
525                 return -1;
526
527         if (oid - lrn->lrn_first_oid >= LFSCK_RBTREE_BITMAP_WIDTH)
528                 return 1;
529
530         return 0;
531 }
532
533 /* The caller should hold llsd->llsd_rb_lock. */
534 static struct lfsck_rbtree_node *
535 lfsck_rbtree_search(struct lfsck_layout_slave_data *llsd,
536                     const struct lu_fid *fid, bool *exact)
537 {
538         struct rb_node           *node  = llsd->llsd_rb_root.rb_node;
539         struct rb_node           *prev  = NULL;
540         struct lfsck_rbtree_node *lrn   = NULL;
541         int                       rc    = 0;
542
543         if (exact != NULL)
544                 *exact = true;
545
546         while (node != NULL) {
547                 prev = node;
548                 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
549                 rc = lfsck_rbtree_cmp(lrn, fid_seq(fid), fid_oid(fid));
550                 if (rc < 0)
551                         node = node->rb_left;
552                 else if (rc > 0)
553                         node = node->rb_right;
554                 else
555                         return lrn;
556         }
557
558         if (exact == NULL)
559                 return NULL;
560
561         /* If there is no exactly matched one, then to the next valid one. */
562         *exact = false;
563
564         /* The rbtree is empty. */
565         if (rc == 0)
566                 return NULL;
567
568         if (rc < 0)
569                 return lrn;
570
571         node = rb_next(prev);
572
573         /* The end of the rbtree. */
574         if (node == NULL)
575                 return NULL;
576
577         lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
578
579         return lrn;
580 }
581
582 static struct lfsck_rbtree_node *lfsck_rbtree_new(const struct lu_env *env,
583                                                   const struct lu_fid *fid)
584 {
585         struct lfsck_rbtree_node *lrn;
586
587         OBD_ALLOC_PTR(lrn);
588         if (lrn == NULL)
589                 return ERR_PTR(-ENOMEM);
590
591         OBD_ALLOC(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
592         if (lrn->lrn_known_bitmap == NULL) {
593                 OBD_FREE_PTR(lrn);
594
595                 return ERR_PTR(-ENOMEM);
596         }
597
598         OBD_ALLOC(lrn->lrn_accessed_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
599         if (lrn->lrn_accessed_bitmap == NULL) {
600                 OBD_FREE(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
601                 OBD_FREE_PTR(lrn);
602
603                 return ERR_PTR(-ENOMEM);
604         }
605
606         RB_CLEAR_NODE(&lrn->lrn_node);
607         lrn->lrn_seq = fid_seq(fid);
608         lrn->lrn_first_oid = fid_oid(fid) & ~LFSCK_RBTREE_BITMAP_MASK;
609         atomic_set(&lrn->lrn_known_count, 0);
610         atomic_set(&lrn->lrn_accessed_count, 0);
611
612         return lrn;
613 }
614
615 static void lfsck_rbtree_free(struct lfsck_rbtree_node *lrn)
616 {
617         OBD_FREE(lrn->lrn_accessed_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
618         OBD_FREE(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
619         OBD_FREE_PTR(lrn);
620 }
621
622 /* The caller should hold lock. */
623 static struct lfsck_rbtree_node *
624 lfsck_rbtree_insert(struct lfsck_layout_slave_data *llsd,
625                     struct lfsck_rbtree_node *lrn)
626 {
627         struct rb_node           **pos    = &llsd->llsd_rb_root.rb_node;
628         struct rb_node            *parent = NULL;
629         struct lfsck_rbtree_node  *tmp;
630         int                        rc;
631
632         while (*pos != NULL) {
633                 parent = *pos;
634                 tmp = rb_entry(parent, struct lfsck_rbtree_node, lrn_node);
635                 rc = lfsck_rbtree_cmp(tmp, lrn->lrn_seq, lrn->lrn_first_oid);
636                 if (rc < 0)
637                         pos = &(*pos)->rb_left;
638                 else if (rc > 0)
639                         pos = &(*pos)->rb_right;
640                 else
641                         return tmp;
642         }
643
644         rb_link_node(&lrn->lrn_node, parent, pos);
645         rb_insert_color(&lrn->lrn_node, &llsd->llsd_rb_root);
646
647         return lrn;
648 }
649
650 static const struct dt_index_operations lfsck_orphan_index_ops;
651
652 static int lfsck_rbtree_setup(const struct lu_env *env,
653                               struct lfsck_component *com)
654 {
655         struct lu_fid                   *fid    = &lfsck_env_info(env)->lti_fid;
656         struct lfsck_instance           *lfsck  = com->lc_lfsck;
657         struct dt_device                *dev    = lfsck->li_bottom;
658         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
659         struct dt_object                *obj;
660
661         fid->f_seq = FID_SEQ_LAYOUT_RBTREE;
662         fid->f_oid = lfsck_dev_idx(lfsck);
663         fid->f_ver = 0;
664         obj = dt_locate(env, dev, fid);
665         if (IS_ERR(obj))
666                 RETURN(PTR_ERR(obj));
667
668         /* Generate an in-RAM object to stand for the layout rbtree.
669          * Scanning the layout rbtree will be via the iteration over
670          * the object. In the future, the rbtree may be written onto
671          * disk with the object.
672          *
673          * Mark the object to be as exist. */
674         obj->do_lu.lo_header->loh_attr |= LOHA_EXISTS;
675         obj->do_index_ops = &lfsck_orphan_index_ops;
676         llsd->llsd_rb_obj = obj;
677         llsd->llsd_rbtree_valid = 1;
678         dev->dd_record_fid_accessed = 1;
679
680         CDEBUG(D_LFSCK, "%s: layout LFSCK init OST-objects accessing bitmap\n",
681                lfsck_lfsck2name(lfsck));
682
683         return 0;
684 }
685
686 static void lfsck_rbtree_cleanup(const struct lu_env *env,
687                                  struct lfsck_component *com)
688 {
689         struct lfsck_instance           *lfsck = com->lc_lfsck;
690         struct lfsck_layout_slave_data  *llsd  = com->lc_data;
691         struct rb_node                  *node  = rb_first(&llsd->llsd_rb_root);
692         struct rb_node                  *next;
693         struct lfsck_rbtree_node        *lrn;
694
695         lfsck->li_bottom->dd_record_fid_accessed = 0;
696         /* Invalid the rbtree, then no others will use it. */
697         down_write(&llsd->llsd_rb_rwsem);
698         llsd->llsd_rbtree_valid = 0;
699         up_write(&llsd->llsd_rb_rwsem);
700
701         while (node != NULL) {
702                 next = rb_next(node);
703                 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
704                 rb_erase(node, &llsd->llsd_rb_root);
705                 lfsck_rbtree_free(lrn);
706                 node = next;
707         }
708
709         if (llsd->llsd_rb_obj != NULL) {
710                 lfsck_object_put(env, llsd->llsd_rb_obj);
711                 llsd->llsd_rb_obj = NULL;
712         }
713
714         CDEBUG(D_LFSCK, "%s: layout LFSCK fini OST-objects accessing bitmap\n",
715                lfsck_lfsck2name(lfsck));
716 }
717
718 static void lfsck_rbtree_update_bitmap(const struct lu_env *env,
719                                        struct lfsck_component *com,
720                                        const struct lu_fid *fid,
721                                        bool accessed)
722 {
723         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
724         struct lfsck_rbtree_node        *lrn;
725         bool                             insert = false;
726         int                              idx;
727         int                              rc     = 0;
728         ENTRY;
729
730         if (unlikely(!fid_is_sane(fid) || fid_is_last_id(fid)))
731                 RETURN_EXIT;
732
733         if (!fid_is_idif(fid) && !fid_is_norm(fid))
734                 RETURN_EXIT;
735
736         down_read(&llsd->llsd_rb_rwsem);
737         if (!llsd->llsd_rbtree_valid)
738                 GOTO(unlock, rc = 0);
739
740         lrn = lfsck_rbtree_search(llsd, fid, NULL);
741         if (lrn == NULL) {
742                 struct lfsck_rbtree_node *tmp;
743
744                 LASSERT(!insert);
745
746                 up_read(&llsd->llsd_rb_rwsem);
747                 tmp = lfsck_rbtree_new(env, fid);
748                 if (IS_ERR(tmp))
749                         GOTO(out, rc = PTR_ERR(tmp));
750
751                 insert = true;
752                 down_write(&llsd->llsd_rb_rwsem);
753                 if (!llsd->llsd_rbtree_valid) {
754                         lfsck_rbtree_free(tmp);
755                         GOTO(unlock, rc = 0);
756                 }
757
758                 lrn = lfsck_rbtree_insert(llsd, tmp);
759                 if (lrn != tmp)
760                         lfsck_rbtree_free(tmp);
761         }
762
763         idx = fid_oid(fid) & LFSCK_RBTREE_BITMAP_MASK;
764         /* Any accessed object must be a known object. */
765         if (!test_and_set_bit(idx, lrn->lrn_known_bitmap))
766                 atomic_inc(&lrn->lrn_known_count);
767         if (accessed && !test_and_set_bit(idx, lrn->lrn_accessed_bitmap))
768                 atomic_inc(&lrn->lrn_accessed_count);
769
770         GOTO(unlock, rc = 0);
771
772 unlock:
773         if (insert)
774                 up_write(&llsd->llsd_rb_rwsem);
775         else
776                 up_read(&llsd->llsd_rb_rwsem);
777 out:
778         if (rc != 0 && accessed) {
779                 struct lfsck_layout *lo = com->lc_file_ram;
780
781                 CDEBUG(D_LFSCK, "%s: fail to update OST-objects accessing "
782                        "bitmap, and will cause incorrect LFSCK OST-object "
783                        "handling, so disable it to cancel orphan handling "
784                        "for related device. rc = %d\n",
785                        lfsck_lfsck2name(com->lc_lfsck), rc);
786
787                 lo->ll_flags |= LF_INCOMPLETE;
788                 lfsck_rbtree_cleanup(env, com);
789         }
790 }
791
792 static inline void lldk_le_to_cpu(struct lfsck_layout_dangling_key *des,
793                                   const struct lfsck_layout_dangling_key *src)
794 {
795         fid_le_to_cpu(&des->lldk_fid, &src->lldk_fid);
796         des->lldk_comp_id = le32_to_cpu(src->lldk_comp_id);
797         des->lldk_ea_off = le32_to_cpu(src->lldk_ea_off);
798 }
799
800 static inline void lldk_cpu_to_le(struct lfsck_layout_dangling_key *des,
801                                   const struct lfsck_layout_dangling_key *src)
802 {
803         fid_cpu_to_le(&des->lldk_fid, &src->lldk_fid);
804         des->lldk_comp_id = cpu_to_le32(src->lldk_comp_id);
805         des->lldk_ea_off = cpu_to_le32(src->lldk_ea_off);
806 }
807
808 static inline void lldk_be_to_cpu(struct lfsck_layout_dangling_key *des,
809                                   const struct lfsck_layout_dangling_key *src)
810 {
811         fid_be_to_cpu(&des->lldk_fid, &src->lldk_fid);
812         des->lldk_comp_id = be32_to_cpu(src->lldk_comp_id);
813         des->lldk_ea_off = be32_to_cpu(src->lldk_ea_off);
814 }
815
816 static inline void lldk_cpu_to_be(struct lfsck_layout_dangling_key *des,
817                                   const struct lfsck_layout_dangling_key *src)
818 {
819         fid_cpu_to_be(&des->lldk_fid, &src->lldk_fid);
820         des->lldk_comp_id = cpu_to_be32(src->lldk_comp_id);
821         des->lldk_ea_off = cpu_to_be32(src->lldk_ea_off);
822 }
823
824 static void lfsck_layout_le_to_cpu(struct lfsck_layout *des,
825                                    const struct lfsck_layout *src)
826 {
827         int i;
828
829         des->ll_magic = le32_to_cpu(src->ll_magic);
830         des->ll_status = le32_to_cpu(src->ll_status);
831         des->ll_flags = le32_to_cpu(src->ll_flags);
832         des->ll_success_count = le32_to_cpu(src->ll_success_count);
833         des->ll_run_time_phase1 = le64_to_cpu(src->ll_run_time_phase1);
834         des->ll_run_time_phase2 = le64_to_cpu(src->ll_run_time_phase2);
835         des->ll_time_last_complete = le64_to_cpu(src->ll_time_last_complete);
836         des->ll_time_latest_start = le64_to_cpu(src->ll_time_latest_start);
837         des->ll_time_last_checkpoint =
838                                 le64_to_cpu(src->ll_time_last_checkpoint);
839         des->ll_pos_latest_start = le64_to_cpu(src->ll_pos_latest_start);
840         des->ll_pos_last_checkpoint = le64_to_cpu(src->ll_pos_last_checkpoint);
841         des->ll_pos_first_inconsistent =
842                         le64_to_cpu(src->ll_pos_first_inconsistent);
843         des->ll_objs_checked_phase1 = le64_to_cpu(src->ll_objs_checked_phase1);
844         des->ll_objs_failed_phase1 = le64_to_cpu(src->ll_objs_failed_phase1);
845         des->ll_objs_checked_phase2 = le64_to_cpu(src->ll_objs_checked_phase2);
846         des->ll_objs_failed_phase2 = le64_to_cpu(src->ll_objs_failed_phase2);
847         for (i = 0; i < LLIT_MAX; i++)
848                 des->ll_objs_repaired[i] =
849                                 le64_to_cpu(src->ll_objs_repaired[i]);
850         des->ll_objs_skipped = le64_to_cpu(src->ll_objs_skipped);
851         des->ll_bitmap_size = le32_to_cpu(src->ll_bitmap_size);
852         lldk_le_to_cpu(&des->ll_lldk_latest_scanned_phase2,
853                        &src->ll_lldk_latest_scanned_phase2);
854 }
855
856 static void lfsck_layout_cpu_to_le(struct lfsck_layout *des,
857                                    const struct lfsck_layout *src)
858 {
859         int i;
860
861         des->ll_magic = cpu_to_le32(src->ll_magic);
862         des->ll_status = cpu_to_le32(src->ll_status);
863         des->ll_flags = cpu_to_le32(src->ll_flags);
864         des->ll_success_count = cpu_to_le32(src->ll_success_count);
865         des->ll_run_time_phase1 = cpu_to_le64(src->ll_run_time_phase1);
866         des->ll_run_time_phase2 = cpu_to_le64(src->ll_run_time_phase2);
867         des->ll_time_last_complete = cpu_to_le64(src->ll_time_last_complete);
868         des->ll_time_latest_start = cpu_to_le64(src->ll_time_latest_start);
869         des->ll_time_last_checkpoint =
870                                 cpu_to_le64(src->ll_time_last_checkpoint);
871         des->ll_pos_latest_start = cpu_to_le64(src->ll_pos_latest_start);
872         des->ll_pos_last_checkpoint = cpu_to_le64(src->ll_pos_last_checkpoint);
873         des->ll_pos_first_inconsistent =
874                         cpu_to_le64(src->ll_pos_first_inconsistent);
875         des->ll_objs_checked_phase1 = cpu_to_le64(src->ll_objs_checked_phase1);
876         des->ll_objs_failed_phase1 = cpu_to_le64(src->ll_objs_failed_phase1);
877         des->ll_objs_checked_phase2 = cpu_to_le64(src->ll_objs_checked_phase2);
878         des->ll_objs_failed_phase2 = cpu_to_le64(src->ll_objs_failed_phase2);
879         for (i = 0; i < LLIT_MAX; i++)
880                 des->ll_objs_repaired[i] =
881                                 cpu_to_le64(src->ll_objs_repaired[i]);
882         des->ll_objs_skipped = cpu_to_le64(src->ll_objs_skipped);
883         des->ll_bitmap_size = cpu_to_le32(src->ll_bitmap_size);
884         lldk_cpu_to_le(&des->ll_lldk_latest_scanned_phase2,
885                        &src->ll_lldk_latest_scanned_phase2);
886 }
887
888 /**
889  * Load the OST bitmap from the lfsck_layout trace file.
890  *
891  * \param[in] env       pointer to the thread context
892  * \param[in] com       pointer to the lfsck component
893  *
894  * \retval              0 for success
895  * \retval              negative error number on failure or data corruption
896  */
897 static int lfsck_layout_load_bitmap(const struct lu_env *env,
898                                     struct lfsck_component *com)
899 {
900         struct dt_object                *obj    = com->lc_obj;
901         struct lfsck_assistant_data     *lad    = com->lc_data;
902         struct lfsck_layout             *lo     = com->lc_file_ram;
903         struct cfs_bitmap                       *bitmap = lad->lad_bitmap;
904         loff_t                           pos    = com->lc_file_size;
905         ssize_t                          size;
906         __u32                            nbits;
907         int                              rc;
908         ENTRY;
909
910         if (com->lc_lfsck->li_ost_descs.ltd_tgts_bitmap->size >
911             lo->ll_bitmap_size)
912                 nbits = com->lc_lfsck->li_ost_descs.ltd_tgts_bitmap->size;
913         else
914                 nbits = lo->ll_bitmap_size;
915
916         if (unlikely(nbits < BITS_PER_LONG))
917                 nbits = BITS_PER_LONG;
918
919         if (nbits > bitmap->size) {
920                 __u32 new_bits = bitmap->size;
921                 struct cfs_bitmap *new_bitmap;
922
923                 while (new_bits < nbits)
924                         new_bits <<= 1;
925
926                 new_bitmap = CFS_ALLOCATE_BITMAP(new_bits);
927                 if (new_bitmap == NULL)
928                         RETURN(-ENOMEM);
929
930                 lad->lad_bitmap = new_bitmap;
931                 CFS_FREE_BITMAP(bitmap);
932                 bitmap = new_bitmap;
933         }
934
935         if (lo->ll_bitmap_size == 0) {
936                 clear_bit(LAD_INCOMPLETE, &lad->lad_flags);
937                 CFS_RESET_BITMAP(bitmap);
938
939                 RETURN(0);
940         }
941
942         size = (lo->ll_bitmap_size + 7) >> 3;
943         rc = dt_read(env, obj, lfsck_buf_get(env, bitmap->data, size), &pos);
944         if (rc != size)
945                 RETURN(rc >= 0 ? -EINVAL : rc);
946
947         if (cfs_bitmap_check_empty(bitmap))
948                 clear_bit(LAD_INCOMPLETE, &lad->lad_flags);
949         else
950                 set_bit(LAD_INCOMPLETE, &lad->lad_flags);
951
952         RETURN(0);
953 }
954
955 /**
956  * Load the layout LFSCK trace file from disk.
957  *
958  * The layout LFSCK trace file records the layout LFSCK status information
959  * and other statistics, such as how many objects have been scanned, and how
960  * many objects have been repaired, and etc. It also contains the bitmap for
961  * failed OSTs during the layout LFSCK. All these information will be loaded
962  * from disk to RAM when the layout LFSCK component setup.
963  *
964  * \param[in] env       pointer to the thread context
965  * \param[in] com       pointer to the lfsck component
966  *
967  * \retval              positive number for file data corruption, the caller
968  *                      should reset the layout LFSCK trace file
969  * \retval              0 for success
970  * \retval              negative error number on failure
971  */
972 static int lfsck_layout_load(const struct lu_env *env,
973                              struct lfsck_component *com)
974 {
975         struct lfsck_layout             *lo     = com->lc_file_ram;
976         ssize_t                          size   = com->lc_file_size;
977         loff_t                           pos    = 0;
978         int                              rc;
979
980         rc = dt_read(env, com->lc_obj,
981                      lfsck_buf_get(env, com->lc_file_disk, size), &pos);
982         if (rc == 0) {
983                 return -ENOENT;
984         } else if (rc < 0) {
985                 CDEBUG(D_LFSCK, "%s: failed to load lfsck_layout: rc = %d\n",
986                        lfsck_lfsck2name(com->lc_lfsck), rc);
987                 return rc;
988         } else if (rc != size) {
989                 CDEBUG(D_LFSCK, "%s: lfsck_layout size %u != %u; reset it\n",
990                        lfsck_lfsck2name(com->lc_lfsck), rc, (unsigned int)size);
991                 return 1;
992         }
993
994         lfsck_layout_le_to_cpu(lo, com->lc_file_disk);
995         if (lo->ll_magic != LFSCK_LAYOUT_MAGIC) {
996                 CDEBUG(D_LFSCK, "%s: invalid lfsck_layout magic %#x != %#x, "
997                        "to be reset\n", lfsck_lfsck2name(com->lc_lfsck),
998                        lo->ll_magic, LFSCK_LAYOUT_MAGIC);
999                 return 1;
1000         }
1001
1002         return 0;
1003 }
1004
1005 /**
1006  * Store the layout LFSCK trace file on disk.
1007  *
1008  * The layout LFSCK trace file records the layout LFSCK status information
1009  * and other statistics, such as how many objects have been scanned, and how
1010  * many objects have been repaired, and etc. It also contains the bitmap for
1011  * failed OSTs during the layout LFSCK. All these information will be synced
1012  * from RAM to disk periodically.
1013  *
1014  * \param[in] env       pointer to the thread context
1015  * \param[in] com       pointer to the lfsck component
1016  *
1017  * \retval              0 for success
1018  * \retval              negative error number on failure
1019  */
1020 static int lfsck_layout_store(const struct lu_env *env,
1021                               struct lfsck_component *com)
1022 {
1023         struct dt_object        *obj    = com->lc_obj;
1024         struct lfsck_instance   *lfsck  = com->lc_lfsck;
1025         struct lfsck_layout     *lo_ram = com->lc_file_ram;
1026         struct lfsck_layout     *lo     = com->lc_file_disk;
1027         struct thandle          *th;
1028         struct dt_device        *dev    = lfsck_obj2dev(obj);
1029         struct cfs_bitmap       *bitmap = NULL;
1030         loff_t                   pos;
1031         ssize_t                  size   = com->lc_file_size;
1032         __u32                    nbits  = 0;
1033         int                      rc;
1034         ENTRY;
1035
1036         if (lfsck->li_master) {
1037                 struct lfsck_assistant_data *lad = com->lc_data;
1038
1039                 bitmap = lad->lad_bitmap;
1040                 nbits = bitmap->size;
1041
1042                 LASSERT(nbits > 0);
1043                 LASSERTF((nbits & 7) == 0, "Invalid nbits %u\n", nbits);
1044         }
1045
1046         lo_ram->ll_bitmap_size = nbits;
1047         lfsck_layout_cpu_to_le(lo, lo_ram);
1048         th = dt_trans_create(env, dev);
1049         if (IS_ERR(th))
1050                 GOTO(log, rc = PTR_ERR(th));
1051
1052         rc = dt_declare_record_write(env, obj, lfsck_buf_get(env, lo, size),
1053                                      (loff_t)0, th);
1054         if (rc != 0)
1055                 GOTO(out, rc);
1056
1057         if (bitmap != NULL) {
1058                 rc = dt_declare_record_write(env, obj,
1059                                 lfsck_buf_get(env, bitmap->data, nbits >> 3),
1060                                 (loff_t)size, th);
1061                 if (rc != 0)
1062                         GOTO(out, rc);
1063         }
1064
1065         rc = dt_trans_start_local(env, dev, th);
1066         if (rc != 0)
1067                 GOTO(out, rc);
1068
1069         pos = 0;
1070         rc = dt_record_write(env, obj, lfsck_buf_get(env, lo, size), &pos, th);
1071         if (rc != 0)
1072                 GOTO(out, rc);
1073
1074         if (bitmap != NULL) {
1075                 pos = size;
1076                 rc = dt_record_write(env, obj,
1077                                 lfsck_buf_get(env, bitmap->data, nbits >> 3),
1078                                 &pos, th);
1079         }
1080
1081         GOTO(out, rc);
1082
1083 out:
1084         dt_trans_stop(env, dev, th);
1085
1086 log:
1087         if (rc != 0)
1088                 CDEBUG(D_LFSCK, "%s: fail to store lfsck_layout: rc = %d\n",
1089                        lfsck_lfsck2name(lfsck), rc);
1090
1091         return rc;
1092 }
1093
1094 static int lfsck_layout_init(const struct lu_env *env,
1095                              struct lfsck_component *com)
1096 {
1097         struct lfsck_layout *lo = com->lc_file_ram;
1098         int rc;
1099
1100         memset(lo, 0, com->lc_file_size);
1101         lo->ll_magic = LFSCK_LAYOUT_MAGIC;
1102         lo->ll_status = LS_INIT;
1103         down_write(&com->lc_sem);
1104         rc = lfsck_layout_store(env, com);
1105         if (rc == 0 && com->lc_lfsck->li_master)
1106                 rc = lfsck_load_sub_trace_files(env, com,
1107                         &dt_lfsck_layout_dangling_features, LFSCK_LAYOUT, true);
1108         up_write(&com->lc_sem);
1109
1110         return rc;
1111 }
1112
1113 static int fid_is_for_ostobj(const struct lu_env *env,
1114                              struct lfsck_instance *lfsck,
1115                              struct dt_object *obj, const struct lu_fid *fid)
1116 {
1117         struct seq_server_site  *ss     = lfsck_dev_site(lfsck);
1118         struct lu_seq_range     *range  = &lfsck_env_info(env)->lti_range;
1119         struct lustre_ost_attrs *loa;
1120         int                      rc;
1121
1122         fld_range_set_any(range);
1123         rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), range);
1124         if (rc == 0) {
1125                 if (fld_range_is_ost(range))
1126                         return 1;
1127
1128                 return 0;
1129         }
1130
1131         loa = &lfsck_env_info(env)->lti_loa;
1132         rc = dt_xattr_get(env, obj, lfsck_buf_get(env, loa, sizeof(*loa)),
1133                           XATTR_NAME_LMA);
1134         if (rc >= (int)sizeof(struct lustre_mdt_attrs)) {
1135                 lustre_lma_swab(&loa->loa_lma);
1136
1137                 return loa->loa_lma.lma_compat & LMAC_FID_ON_OST ? 1 : 0;
1138         }
1139
1140         rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_FID);
1141
1142         return rc > 0;
1143 }
1144
1145 static struct lfsck_layout_seq *
1146 lfsck_layout_seq_lookup(struct lfsck_layout_slave_data *llsd, __u64 seq)
1147 {
1148         struct lfsck_layout_seq *lls;
1149
1150         list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
1151                 if (lls->lls_seq == seq)
1152                         return lls;
1153
1154                 if (lls->lls_seq > seq)
1155                         return NULL;
1156         }
1157
1158         return NULL;
1159 }
1160
1161 static void
1162 lfsck_layout_seq_insert(struct lfsck_layout_slave_data *llsd,
1163                         struct lfsck_layout_seq *lls)
1164 {
1165         struct lfsck_layout_seq *tmp;
1166         struct list_head        *pos = &llsd->llsd_seq_list;
1167
1168         list_for_each_entry(tmp, &llsd->llsd_seq_list, lls_list) {
1169                 if (lls->lls_seq < tmp->lls_seq) {
1170                         pos = &tmp->lls_list;
1171                         break;
1172                 }
1173         }
1174         list_add_tail(&lls->lls_list, pos);
1175 }
1176
1177 static int
1178 lfsck_layout_lastid_create(const struct lu_env *env,
1179                            struct lfsck_instance *lfsck,
1180                            struct dt_object *obj)
1181 {
1182         struct lfsck_thread_info *info   = lfsck_env_info(env);
1183         struct lu_attr           *la     = &info->lti_la;
1184         struct dt_object_format  *dof    = &info->lti_dof;
1185         struct lfsck_bookmark    *bk     = &lfsck->li_bookmark_ram;
1186         struct dt_device         *dt     = lfsck_obj2dev(obj);
1187         struct thandle           *th;
1188         __u64                     lastid = 0;
1189         loff_t                    pos    = 0;
1190         int                       rc;
1191         ENTRY;
1192
1193         if (bk->lb_param & LPF_DRYRUN)
1194                 return 0;
1195
1196         memset(la, 0, sizeof(*la));
1197         la->la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
1198         la->la_valid = LA_MODE | LA_UID | LA_GID;
1199         memset(dof, 0, sizeof(*dof));
1200         dof->dof_type = dt_mode_to_dft(S_IFREG);
1201
1202         th = dt_trans_create(env, dt);
1203         if (IS_ERR(th))
1204                 GOTO(log, rc = PTR_ERR(th));
1205
1206         rc = dt_declare_create(env, obj, la, NULL, dof, th);
1207         if (rc != 0)
1208                 GOTO(stop, rc);
1209
1210         rc = dt_declare_record_write(env, obj,
1211                                      lfsck_buf_get(env, &lastid,
1212                                                    sizeof(lastid)),
1213                                      pos, th);
1214         if (rc != 0)
1215                 GOTO(stop, rc);
1216
1217         rc = dt_trans_start_local(env, dt, th);
1218         if (rc != 0)
1219                 GOTO(stop, rc);
1220
1221         dt_write_lock(env, obj, 0);
1222         if (likely(dt_object_exists(obj) == 0)) {
1223                 rc = dt_create(env, obj, la, NULL, dof, th);
1224                 if (rc == 0)
1225                         rc = dt_record_write(env, obj,
1226                                 lfsck_buf_get(env, &lastid, sizeof(lastid)),
1227                                 &pos, th);
1228         }
1229         dt_write_unlock(env, obj);
1230
1231         GOTO(stop, rc);
1232
1233 stop:
1234         dt_trans_stop(env, dt, th);
1235
1236 log:
1237         CDEBUG(D_LFSCK, "%s: layout LFSCK will create LAST_ID for <seq> "
1238                "%#llx: rc = %d\n",
1239                lfsck_lfsck2name(lfsck), fid_seq(lfsck_dto2fid(obj)), rc);
1240
1241         return rc;
1242 }
1243
1244 static int
1245 lfsck_layout_lastid_reload(const struct lu_env *env,
1246                            struct lfsck_component *com,
1247                            struct lfsck_layout_seq *lls)
1248 {
1249         __u64   lastid;
1250         loff_t  pos     = 0;
1251         int     rc;
1252
1253         dt_read_lock(env, lls->lls_lastid_obj, 0);
1254         rc = dt_record_read(env, lls->lls_lastid_obj,
1255                             lfsck_buf_get(env, &lastid, sizeof(lastid)), &pos);
1256         dt_read_unlock(env, lls->lls_lastid_obj);
1257         if (unlikely(rc != 0))
1258                 return rc;
1259
1260         lastid = le64_to_cpu(lastid);
1261         if (lastid < lls->lls_lastid_known) {
1262                 struct lfsck_instance   *lfsck  = com->lc_lfsck;
1263                 struct lfsck_layout     *lo     = com->lc_file_ram;
1264
1265                 lls->lls_lastid = lls->lls_lastid_known;
1266                 lls->lls_dirty = 1;
1267                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
1268                         LASSERT(lfsck->li_out_notify != NULL);
1269
1270                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1271                                              LE_LASTID_REBUILDING);
1272                         lo->ll_flags |= LF_CRASHED_LASTID;
1273
1274                         CDEBUG(D_LFSCK, "%s: layout LFSCK finds crashed "
1275                                "LAST_ID file (1) for the sequence %#llx"
1276                                ", old value %llu, known value %llu\n",
1277                                lfsck_lfsck2name(lfsck), lls->lls_seq,
1278                                lastid, lls->lls_lastid);
1279                 }
1280         } else if (lastid >= lls->lls_lastid) {
1281                 lls->lls_lastid = lastid;
1282                 lls->lls_dirty = 0;
1283         }
1284
1285         return 0;
1286 }
1287
1288 static int
1289 lfsck_layout_lastid_store(const struct lu_env *env,
1290                           struct lfsck_component *com)
1291 {
1292         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1293         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
1294         struct dt_device                *dt     = lfsck->li_bottom;
1295         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
1296         struct lfsck_layout_seq         *lls;
1297         struct thandle                  *th;
1298         __u64                            lastid;
1299         int                              rc     = 0;
1300         int                              rc1    = 0;
1301
1302         list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
1303                 loff_t pos = 0;
1304
1305                 if (!lls->lls_dirty)
1306                         continue;
1307
1308                 CDEBUG(D_LFSCK, "%s: layout LFSCK will sync the LAST_ID for "
1309                        "<seq> %#llx as <oid> %llu\n",
1310                        lfsck_lfsck2name(lfsck), lls->lls_seq, lls->lls_lastid);
1311
1312                 if (bk->lb_param & LPF_DRYRUN) {
1313                         lls->lls_dirty = 0;
1314                         continue;
1315                 }
1316
1317                 th = dt_trans_create(env, dt);
1318                 if (IS_ERR(th)) {
1319                         rc1 = PTR_ERR(th);
1320                         CDEBUG(D_LFSCK, "%s: layout LFSCK failed to store "
1321                                "the LAST_ID for <seq> %#llx(1): rc = %d\n",
1322                                lfsck_lfsck2name(com->lc_lfsck),
1323                                lls->lls_seq, rc1);
1324                         continue;
1325                 }
1326
1327                 lastid = cpu_to_le64(lls->lls_lastid);
1328                 rc = dt_declare_record_write(env, lls->lls_lastid_obj,
1329                                              lfsck_buf_get(env, &lastid,
1330                                                            sizeof(lastid)),
1331                                              pos, th);
1332                 if (rc != 0)
1333                         goto stop;
1334
1335                 rc = dt_trans_start_local(env, dt, th);
1336                 if (rc != 0)
1337                         goto stop;
1338
1339                 dt_write_lock(env, lls->lls_lastid_obj, 0);
1340                 rc = dt_record_write(env, lls->lls_lastid_obj,
1341                                      lfsck_buf_get(env, &lastid,
1342                                      sizeof(lastid)), &pos, th);
1343                 dt_write_unlock(env, lls->lls_lastid_obj);
1344                 if (rc == 0)
1345                         lls->lls_dirty = 0;
1346
1347 stop:
1348                 dt_trans_stop(env, dt, th);
1349                 if (rc != 0) {
1350                         rc1 = rc;
1351                         CDEBUG(D_LFSCK, "%s: layout LFSCK failed to store "
1352                                "the LAST_ID for <seq> %#llx(2): rc = %d\n",
1353                                lfsck_lfsck2name(com->lc_lfsck),
1354                                lls->lls_seq, rc1);
1355                 }
1356         }
1357
1358         return rc1;
1359 }
1360
1361 static int
1362 lfsck_layout_lastid_load(const struct lu_env *env,
1363                          struct lfsck_component *com,
1364                          struct lfsck_layout_seq *lls)
1365 {
1366         struct lfsck_instance   *lfsck  = com->lc_lfsck;
1367         struct lfsck_layout     *lo     = com->lc_file_ram;
1368         struct lu_fid           *fid    = &lfsck_env_info(env)->lti_fid;
1369         struct dt_object        *obj;
1370         loff_t                   pos    = 0;
1371         int                      rc;
1372         ENTRY;
1373
1374         lu_last_id_fid(fid, lls->lls_seq, lfsck_dev_idx(lfsck));
1375         obj = dt_locate(env, lfsck->li_bottom, fid);
1376         if (IS_ERR(obj))
1377                 RETURN(PTR_ERR(obj));
1378
1379         /* LAST_ID crashed, to be rebuilt */
1380         if (dt_object_exists(obj) == 0) {
1381                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
1382                         LASSERT(lfsck->li_out_notify != NULL);
1383
1384                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1385                                              LE_LASTID_REBUILDING);
1386                         lo->ll_flags |= LF_CRASHED_LASTID;
1387
1388                         CDEBUG(D_LFSCK, "%s: layout LFSCK cannot find the "
1389                                "LAST_ID file for sequence %#llx\n",
1390                                lfsck_lfsck2name(lfsck), lls->lls_seq);
1391
1392                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY4) &&
1393                             cfs_fail_val > 0) {
1394                                 struct ptlrpc_thread *thread =
1395                                         &lfsck->li_thread;
1396
1397                                 up_write(&com->lc_sem);
1398                                 wait_event_idle_timeout(
1399                                         thread->t_ctl_waitq,
1400                                         !thread_is_running(thread),
1401                                         cfs_time_seconds(cfs_fail_val));
1402                                 down_write(&com->lc_sem);
1403                         }
1404                 }
1405
1406                 rc = lfsck_layout_lastid_create(env, lfsck, obj);
1407         } else {
1408                 dt_read_lock(env, obj, 0);
1409                 rc = dt_read(env, obj,
1410                         lfsck_buf_get(env, &lls->lls_lastid, sizeof(__u64)),
1411                         &pos);
1412                 dt_read_unlock(env, obj);
1413                 if (rc != 0 && rc != sizeof(__u64))
1414                         GOTO(out, rc = (rc > 0 ? -EFAULT : rc));
1415
1416                 if (rc == 0 && !(lo->ll_flags & LF_CRASHED_LASTID)) {
1417                         LASSERT(lfsck->li_out_notify != NULL);
1418
1419                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1420                                              LE_LASTID_REBUILDING);
1421                         lo->ll_flags |= LF_CRASHED_LASTID;
1422
1423                         CDEBUG(D_LFSCK, "%s: layout LFSCK finds invalid "
1424                                "LAST_ID file for the sequence %#llx"
1425                                ": rc = %d\n",
1426                                lfsck_lfsck2name(lfsck), lls->lls_seq, rc);
1427                 }
1428
1429                 lls->lls_lastid = le64_to_cpu(lls->lls_lastid);
1430                 rc = 0;
1431         }
1432
1433         GOTO(out, rc);
1434
1435 out:
1436         if (rc != 0)
1437                 lfsck_object_put(env, obj);
1438         else
1439                 lls->lls_lastid_obj = obj;
1440
1441         return rc;
1442 }
1443
1444 static void lfsck_layout_record_failure(const struct lu_env *env,
1445                                         struct lfsck_instance *lfsck,
1446                                         struct lfsck_layout *lo)
1447 {
1448         __u64 cookie;
1449
1450         lo->ll_objs_failed_phase1++;
1451         cookie = lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
1452                                                         lfsck->li_di_oit);
1453         if (lo->ll_pos_first_inconsistent == 0 ||
1454             lo->ll_pos_first_inconsistent < cookie) {
1455                 lo->ll_pos_first_inconsistent = cookie;
1456
1457                 CDEBUG(D_LFSCK, "%s: layout LFSCK hit first non-repaired "
1458                        "inconsistency at the pos [%llu]\n",
1459                        lfsck_lfsck2name(lfsck),
1460                        lo->ll_pos_first_inconsistent);
1461         }
1462 }
1463
1464 static int lfsck_layout_double_scan_result(const struct lu_env *env,
1465                                            struct lfsck_component *com,
1466                                            int rc)
1467 {
1468         struct lfsck_instance   *lfsck = com->lc_lfsck;
1469         struct lfsck_layout     *lo    = com->lc_file_ram;
1470
1471         CDEBUG(D_LFSCK, "%s: layout LFSCK double scan: rc = %d\n",
1472                lfsck_lfsck2name(lfsck), rc);
1473
1474         down_write(&com->lc_sem);
1475         lo->ll_run_time_phase2 += ktime_get_seconds() -
1476                                   com->lc_time_last_checkpoint;
1477         lo->ll_time_last_checkpoint = ktime_get_real_seconds();
1478         lo->ll_objs_checked_phase2 += com->lc_new_checked;
1479
1480         if (rc > 0) {
1481                 if (lo->ll_flags & LF_INCOMPLETE) {
1482                         lo->ll_status = LS_PARTIAL;
1483                 } else {
1484                         if (lfsck->li_master) {
1485                                 struct lfsck_assistant_data *lad = com->lc_data;
1486
1487                                 if (test_bit(LAD_INCOMPLETE, &lad->lad_flags))
1488                                         lo->ll_status = LS_PARTIAL;
1489                                 else
1490                                         lo->ll_status = LS_COMPLETED;
1491                         } else {
1492                                 lo->ll_status = LS_COMPLETED;
1493                         }
1494                 }
1495                 lo->ll_flags &= ~LF_SCANNED_ONCE;
1496                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN))
1497                         lo->ll_flags &= ~LF_INCONSISTENT;
1498                 lo->ll_time_last_complete = lo->ll_time_last_checkpoint;
1499                 lo->ll_success_count++;
1500         } else if (rc == 0) {
1501                 if (lfsck->li_status != 0)
1502                         lo->ll_status = lfsck->li_status;
1503                 else
1504                         lo->ll_status = LS_STOPPED;
1505         } else {
1506                 lo->ll_status = LS_FAILED;
1507         }
1508
1509         rc = lfsck_layout_store(env, com);
1510         up_write(&com->lc_sem);
1511
1512         CDEBUG(D_LFSCK, "%s: layout LFSCK double scan result %u: rc = %d\n",
1513                lfsck_lfsck2name(lfsck), lo->ll_status, rc);
1514
1515         return rc;
1516 }
1517
1518 static int lfsck_layout_trans_stop(const struct lu_env *env,
1519                                    struct dt_device *dev,
1520                                    struct thandle *handle, int result)
1521 {
1522         int rc;
1523
1524         /* XXX: If there is something worng or it needs to repair nothing,
1525          *      then notify the lower to stop the modification. Currently,
1526          *      we use th_result for such purpose, that may be replaced by
1527          *      some rollback mechanism in the future. */
1528         handle->th_result = result;
1529         rc = dt_trans_stop(env, dev, handle);
1530         if (result != 0)
1531                 return result > 0 ? 0 : result;
1532
1533         return rc == 0 ? 1 : rc;
1534 }
1535
1536 static int lfsck_layout_ins_dangling_rec(const struct lu_env *env,
1537                                          struct lfsck_component *com,
1538                                          const struct lu_fid *pfid,
1539                                          const struct lu_fid *cfid,
1540                                          __u32 comp_id, __u32 ea_off,
1541                                          __u32 ost_idx)
1542 {
1543         struct lfsck_layout_dangling_key *key = &lfsck_env_info(env)->lti_lldk;
1544         struct lu_fid *rec = &lfsck_env_info(env)->lti_fid3;
1545         struct dt_device *dev;
1546         struct dt_object *obj;
1547         struct thandle *th = NULL;
1548         int idx;
1549         int rc = 0;
1550         ENTRY;
1551
1552         idx = lfsck_sub_trace_file_fid2idx(pfid);
1553         obj = com->lc_sub_trace_objs[idx].lsto_obj;
1554         dev = lfsck_obj2dev(obj);
1555
1556         fid_cpu_to_be(&key->lldk_fid, pfid);
1557         key->lldk_comp_id = cpu_to_be32(comp_id);
1558         key->lldk_ea_off = cpu_to_be32(ea_off);
1559
1560         fid_cpu_to_be(rec, cfid);
1561         rec->f_ver = cpu_to_be32(ost_idx);
1562
1563         mutex_lock(&com->lc_sub_trace_objs[idx].lsto_mutex);
1564
1565         th = dt_trans_create(env, dev);
1566         if (IS_ERR(th))
1567                 GOTO(unlock, rc = PTR_ERR(th));
1568
1569         rc = dt_declare_insert(env, obj,
1570                                (const struct dt_rec *)rec,
1571                                (const struct dt_key *)key, th);
1572         if (rc)
1573                 GOTO(unlock, rc);
1574
1575         rc = dt_trans_start_local(env, dev, th);
1576         if (rc)
1577                 GOTO(unlock, rc);
1578
1579         rc = dt_insert(env, obj, (const struct dt_rec *)rec,
1580                        (const struct dt_key *)key, th);
1581
1582         GOTO(unlock, rc);
1583
1584 unlock:
1585         if (th && !IS_ERR(th))
1586                 dt_trans_stop(env, dev, th);
1587
1588         mutex_unlock(&com->lc_sub_trace_objs[idx].lsto_mutex);
1589
1590         CDEBUG(D_LFSCK, "%s: insert the paris "DFID" => "DFID", comp_id = %u, "
1591                "ea_off = %u, ost_idx = %u, into the trace file for further "
1592                "dangling check: rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
1593                PFID(pfid), PFID(cfid), comp_id, ea_off, ost_idx, rc);
1594
1595         return rc;
1596 }
1597
1598 static int lfsck_layout_del_dangling_rec(const struct lu_env *env,
1599                                          struct lfsck_component *com,
1600                                          const struct lu_fid *fid,
1601                                          __u32 comp_id, __u32 ea_off)
1602 {
1603         struct lfsck_layout_dangling_key *key = &lfsck_env_info(env)->lti_lldk;
1604         struct dt_device *dev;
1605         struct dt_object *obj;
1606         struct thandle *th = NULL;
1607         int idx;
1608         int rc = 0;
1609         ENTRY;
1610
1611         idx = lfsck_sub_trace_file_fid2idx(fid);
1612         obj = com->lc_sub_trace_objs[idx].lsto_obj;
1613         dev = lfsck_obj2dev(obj);
1614
1615         fid_cpu_to_be(&key->lldk_fid, fid);
1616         key->lldk_comp_id = cpu_to_be32(comp_id);
1617         key->lldk_ea_off = cpu_to_be32(ea_off);
1618
1619         mutex_lock(&com->lc_sub_trace_objs[idx].lsto_mutex);
1620
1621         th = dt_trans_create(env, dev);
1622         if (IS_ERR(th))
1623                 GOTO(unlock, rc = PTR_ERR(th));
1624
1625         rc = dt_declare_delete(env, obj, (const struct dt_key *)key, th);
1626         if (rc)
1627                 GOTO(unlock, rc);
1628
1629         rc = dt_trans_start_local(env, dev, th);
1630         if (rc)
1631                 GOTO(unlock, rc);
1632
1633         rc = dt_delete(env, obj, (const struct dt_key *)key, th);
1634
1635         GOTO(unlock, rc);
1636
1637 unlock:
1638         if (th && !IS_ERR(th))
1639                 dt_trans_stop(env, dev, th);
1640
1641         mutex_unlock(&com->lc_sub_trace_objs[idx].lsto_mutex);
1642
1643         CDEBUG(D_LFSCK, "%s: delete the dangling record for "DFID
1644                ", comp_id = %u, ea_off = %u from the trace file: rc = %d\n",
1645                lfsck_lfsck2name(com->lc_lfsck), PFID(fid), comp_id, ea_off, rc);
1646
1647         return rc;
1648 }
1649
1650 /**
1651  * Get the system default stripe size.
1652  *
1653  * \param[in] env       pointer to the thread context
1654  * \param[in] lfsck     pointer to the lfsck instance
1655  * \param[out] size     pointer to the default stripe size
1656  *
1657  * \retval              0 for success
1658  * \retval              negative error number on failure
1659  */
1660 static int lfsck_layout_get_def_stripesize(const struct lu_env *env,
1661                                            struct lfsck_instance *lfsck,
1662                                            __u32 *size)
1663 {
1664         struct lov_user_md      *lum = &lfsck_env_info(env)->lti_lum;
1665         struct dt_object        *root;
1666         int                      rc;
1667
1668         root = dt_locate(env, lfsck->li_next, &lfsck->li_local_root_fid);
1669         if (IS_ERR(root))
1670                 return PTR_ERR(root);
1671
1672         /* Get the default stripe size via xattr_get on the backend root. */
1673         rc = dt_xattr_get(env, root, lfsck_buf_get(env, lum, sizeof(*lum)),
1674                           XATTR_NAME_LOV);
1675         if (rc > 0) {
1676                 /* The lum->lmm_stripe_size is LE mode. The *size also
1677                  * should be LE mode. So it is unnecessary to convert. */
1678                 *size = lum->lmm_stripe_size;
1679                 rc = 0;
1680         } else if (unlikely(rc == 0)) {
1681                 rc = -EINVAL;
1682         }
1683
1684         lfsck_object_put(env, root);
1685
1686         return rc;
1687 }
1688
1689 /**
1690  * \retval       +1: repaired
1691  * \retval        0: did nothing
1692  * \retval      -ve: on error
1693  */
1694 static int lfsck_layout_refill_lovea(const struct lu_env *env,
1695                                      struct lfsck_instance *lfsck,
1696                                      struct thandle *handle,
1697                                      struct dt_object *parent,
1698                                      const struct lu_fid *cfid,
1699                                      struct lu_buf *buf,
1700                                      struct lov_mds_md_v1 *lmm,
1701                                      struct lov_ost_data_v1 *slot,
1702                                      int fl, __u32 ost_idx, int size)
1703 {
1704         struct ost_id           *oi     = &lfsck_env_info(env)->lti_oi;
1705         struct lu_buf            ea_buf;
1706         int                      rc;
1707         __u32                    magic;
1708         __u32                    pattern;
1709         __u16                    count;
1710         ENTRY;
1711
1712         magic = le32_to_cpu(lmm->lmm_magic);
1713         pattern = le32_to_cpu(lmm->lmm_pattern);
1714         count = le16_to_cpu(lmm->lmm_stripe_count);
1715
1716         fid_to_ostid(cfid, oi);
1717         ostid_cpu_to_le(oi, &slot->l_ost_oi);
1718         slot->l_ost_gen = cpu_to_le32(0);
1719         slot->l_ost_idx = cpu_to_le32(ost_idx);
1720
1721         if (pattern & LOV_PATTERN_F_HOLE) {
1722                 struct lov_ost_data_v1 *objs;
1723                 int                     i;
1724
1725                 if (magic == LOV_MAGIC_V1)
1726                         objs = &lmm->lmm_objects[0];
1727                 else
1728                         objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1729                 for (i = 0; i < count; i++, objs++) {
1730                         if (lovea_slot_is_dummy(objs))
1731                                 break;
1732                 }
1733
1734                 /* If the @slot is the last dummy slot to be refilled,
1735                  * then drop LOV_PATTERN_F_HOLE from lmm::lmm_pattern. */
1736                 if (i == count) {
1737                         lmm->lmm_pattern =
1738                                 cpu_to_le32(pattern & ~LOV_PATTERN_F_HOLE);
1739
1740                         CDEBUG(D_LFSCK, "%s: remove layout HOLE for "DFID
1741                                ": parent "DFID"\n", lfsck_lfsck2name(lfsck),
1742                                PFID(cfid), PFID(lfsck_dto2fid(parent)));
1743                 }
1744         }
1745
1746         lfsck_buf_init(&ea_buf, buf->lb_buf, size);
1747         rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV, fl, handle);
1748         if (rc == 0)
1749                 rc = 1;
1750
1751         RETURN(rc);
1752 }
1753
1754 static struct lov_ost_data_v1 *
1755 __lfsck_layout_new_v1_lovea(struct lov_mds_md_v1 *lmm,
1756                             const struct lu_fid *pfid,
1757                             __u32 stripe_size, __u32 ea_off,
1758                             __u32 pattern, __u16 count)
1759 {
1760         lmm->lmm_magic = cpu_to_le32(LOV_MAGIC_V1);
1761         lmm->lmm_pattern = cpu_to_le32(pattern);
1762         fid_to_lmm_oi(pfid, &lmm->lmm_oi);
1763         lmm_oi_cpu_to_le(&lmm->lmm_oi, &lmm->lmm_oi);
1764         lmm->lmm_stripe_size = cpu_to_le32(stripe_size);
1765         lmm->lmm_stripe_count = cpu_to_le16(count);
1766         lmm->lmm_layout_gen = cpu_to_le16(1);
1767         memset(&lmm->lmm_objects[0], 0,
1768                sizeof(struct lov_ost_data_v1) * count);
1769
1770         return &lmm->lmm_objects[ea_off];
1771 }
1772
1773 static int lfsck_layout_new_v1_lovea(const struct lu_env *env,
1774                                      struct lfsck_instance *lfsck,
1775                                      struct ost_layout *ol,
1776                                      struct dt_object *parent,
1777                                      struct lu_buf *buf, __u32 ea_off,
1778                                      struct lov_mds_md_v1 **lmm,
1779                                      struct lov_ost_data_v1 **objs)
1780 {
1781         int size;
1782         __u32 stripe_size = ol->ol_stripe_size;
1783         __u32 pattern = LOV_PATTERN_RAID0;
1784         __u16 count;
1785
1786         if (ol->ol_stripe_count != 0)
1787                 count = ol->ol_stripe_count;
1788         else
1789                 count = ea_off + 1;
1790
1791         size = lov_mds_md_size(count, LOV_MAGIC_V1);
1792         LASSERTF(buf->lb_len >= size,
1793                  "buffer len %d is less than real size %d\n",
1794                  (int)buf->lb_len, size);
1795
1796         if (stripe_size == 0) {
1797                 int rc;
1798
1799                 rc = lfsck_layout_get_def_stripesize(env, lfsck, &stripe_size);
1800                 if (rc)
1801                         return rc;
1802         }
1803
1804         *lmm = buf->lb_buf;
1805         if (ol->ol_stripe_count > 1 ||
1806             (ol->ol_stripe_count == 0 && ea_off != 0)) {
1807                 pattern |= LOV_PATTERN_F_HOLE;
1808                 memset(&(*lmm)->lmm_objects[0], 0,
1809                        count * sizeof(struct lov_ost_data_v1));
1810         }
1811
1812         *objs = __lfsck_layout_new_v1_lovea(*lmm, lfsck_dto2fid(parent),
1813                                 stripe_size, ea_off, pattern, count);
1814
1815         return size;
1816 }
1817
1818 static int lfsck_layout_new_comp_lovea(const struct lu_env *env,
1819                                        struct lu_orphan_rec_v3 *rec,
1820                                        struct dt_object *parent,
1821                                        struct lu_buf *buf, __u32 ea_off,
1822                                        struct lov_mds_md_v1 **lmm,
1823                                        struct lov_ost_data_v1 **objs)
1824 {
1825         struct ost_layout *ol = &rec->lor_layout;
1826         struct lov_comp_md_v1 *lcm;
1827         struct lov_comp_md_entry_v1 *lcme;
1828         __u32 pattern = LOV_PATTERN_RAID0;
1829         __u32 offset = sizeof(*lcm) + sizeof(*lcme);
1830         int lcme_size = lov_mds_md_size(ol->ol_stripe_count, LOV_MAGIC_V1);
1831         int size = offset + lcme_size;
1832
1833         LASSERTF(buf->lb_len >= size,
1834                  "buffer len %d is less than real size %d\n",
1835                  (int)buf->lb_len, size);
1836
1837         lcm = buf->lb_buf;
1838         lcm->lcm_magic = cpu_to_le32(LOV_MAGIC_COMP_V1);
1839         lcm->lcm_size = cpu_to_le32(size);
1840         if (rec->lor_range) {
1841                 lcm->lcm_layout_gen = cpu_to_le32(rec->lor_layout_version +
1842                                                   rec->lor_range);
1843                 lcm->lcm_flags = cpu_to_le16(LCM_FL_WRITE_PENDING);
1844         } else if (rec->lor_layout_version) {
1845                 lcm->lcm_layout_gen = cpu_to_le32(rec->lor_layout_version +
1846                                                   rec->lor_range);
1847                 lcm->lcm_flags = cpu_to_le16(LCM_FL_NONE);
1848         } else {
1849                 lcm->lcm_layout_gen = cpu_to_le32(1);
1850                 lcm->lcm_flags = cpu_to_le16(LCM_FL_NONE);
1851         }
1852         lcm->lcm_entry_count = cpu_to_le16(1);
1853         /* Currently, we do not know how many mirrors will be, set it as zero
1854          * at the beginning. It will be updated when more mirrors are found. */
1855         lcm->lcm_mirror_count = 0;
1856
1857         lcme = &lcm->lcm_entries[0];
1858         lcme->lcme_id = cpu_to_le32(ol->ol_comp_id);
1859         lcme->lcme_flags = cpu_to_le32(LCME_FL_INIT);
1860         lcme->lcme_extent.e_start = cpu_to_le64(ol->ol_comp_start);
1861         lcme->lcme_extent.e_end = cpu_to_le64(ol->ol_comp_end);
1862         lcme->lcme_offset = cpu_to_le32(offset);
1863         lcme->lcme_size = cpu_to_le32(lcme_size);
1864         lcme->lcme_layout_gen = lcm->lcm_layout_gen;
1865         if (ol->ol_stripe_count > 1)
1866                 pattern |= LOV_PATTERN_F_HOLE;
1867
1868         *lmm = buf->lb_buf + offset;
1869         *objs = __lfsck_layout_new_v1_lovea(*lmm, lfsck_dto2fid(parent),
1870                                             ol->ol_stripe_size, ea_off,
1871                                             pattern, ol->ol_stripe_count);
1872
1873         return size;
1874 }
1875
1876 static void lfsck_layout_update_lcm(struct lov_comp_md_v1 *lcm,
1877                                     struct lov_comp_md_entry_v1 *lcme,
1878                                     __u32 version, __u32 range)
1879 {
1880         struct lov_comp_md_entry_v1 *tmp;
1881         __u64 start = le64_to_cpu(lcme->lcme_extent.e_start);
1882         __u64 end = le64_to_cpu(lcme->lcme_extent.e_end);
1883         __u32 gen = version + range;
1884         __u32 tmp_gen;
1885         int i;
1886         __u16 count = le16_to_cpu(lcm->lcm_entry_count);
1887         __u16 flags = le16_to_cpu(lcm->lcm_flags);
1888
1889         if (!gen)
1890                 gen = 1;
1891         lcme->lcme_layout_gen = cpu_to_le32(gen);
1892         if (le32_to_cpu(lcm->lcm_layout_gen) < gen)
1893                 lcm->lcm_layout_gen = cpu_to_le32(gen);
1894
1895         if (range)
1896                 lcm->lcm_flags = cpu_to_le16(LCM_FL_WRITE_PENDING);
1897         else if (flags == LCM_FL_NONE && le16_to_cpu(lcm->lcm_mirror_count) > 0)
1898                 lcm->lcm_flags = cpu_to_le16(LCM_FL_RDONLY);
1899
1900         for (i = 0; i < count; i++) {
1901                 tmp = &lcm->lcm_entries[i];
1902                 if (le64_to_cpu(tmp->lcme_extent.e_end) <= start)
1903                         continue;
1904
1905                 if (le64_to_cpu(tmp->lcme_extent.e_start) >= end)
1906                         continue;
1907
1908                 if (le32_to_cpu(tmp->lcme_flags) & LCME_FL_STALE)
1909                         continue;
1910
1911                 tmp_gen = le32_to_cpu(tmp->lcme_layout_gen);
1912                 /* "lcme_layout_gen == 0" but without LCME_FL_STALE flag,
1913                  * then it should be the latest version of all mirrors. */
1914                 if (tmp_gen == 0 || tmp_gen > gen) {
1915                         lcme->lcme_flags = cpu_to_le32(
1916                                 le32_to_cpu(lcme->lcme_flags) | LCME_FL_STALE);
1917                         break;
1918                 }
1919
1920                 if (tmp_gen < gen)
1921                         tmp->lcme_flags = cpu_to_le32(
1922                                 le32_to_cpu(tmp->lcme_flags) | LCME_FL_STALE);
1923         }
1924 }
1925
1926 static int lfsck_layout_add_comp(const struct lu_env *env,
1927                                  struct lfsck_instance *lfsck,
1928                                  struct thandle *handle,
1929                                  struct lu_orphan_rec_v3 *rec,
1930                                  struct dt_object *parent,
1931                                  const struct lu_fid *cfid,
1932                                  struct lu_buf *buf, __u32 ost_idx,
1933                                  __u32 ea_off, int pos, bool new_mirror)
1934 {
1935         struct ost_layout *ol = &rec->lor_layout;
1936         struct lov_comp_md_v1 *lcm = buf->lb_buf;
1937         struct lov_comp_md_entry_v1 *lcme;
1938         struct lov_mds_md_v1 *lmm;
1939         struct lov_ost_data_v1 *objs;
1940         int added = sizeof(*lcme) +
1941                     lov_mds_md_size(ol->ol_stripe_count, LOV_MAGIC_V1);
1942         int size = le32_to_cpu(lcm->lcm_size) + added;
1943         int rc;
1944         int i;
1945         __u32 offset;
1946         __u32 pattern = LOV_PATTERN_RAID0;
1947         __u16 count = le16_to_cpu(lcm->lcm_entry_count);
1948         ENTRY;
1949
1950         lu_buf_check_and_grow(buf, size);
1951         /* set the lcm again because lu_buf_check_and_grow() may
1952          * have reallocated the buf. */
1953         lcm = buf->lb_buf;
1954         lcm->lcm_size = cpu_to_le32(size);
1955         lcm->lcm_entry_count = cpu_to_le16(count + 1);
1956         if (new_mirror)
1957                 le16_add_cpu(&lcm->lcm_mirror_count, 1);
1958
1959         /* 1. Move the component bodies from [pos, count-1] to [pos+1, count]
1960          *    with distance of 'added'. */
1961         if (pos < count) {
1962                 size = 0;
1963                 for (i = pos; i < count; i++) {
1964                         lcme = &lcm->lcm_entries[i];
1965                         size += le32_to_cpu(lcme->lcme_size);
1966                 }
1967
1968                 offset = le32_to_cpu(lcm->lcm_entries[pos].lcme_offset);
1969                 memmove(buf->lb_buf + offset + added,
1970                         buf->lb_buf + offset, size);
1971         }
1972
1973         size = 0;
1974         /* 2. Move the component header [0, pos-1] to [0, pos-1] with distance
1975          *    of 'sizeof(struct lov_comp_md_entry_v1)' */
1976         if (pos > 0) {
1977                 for (i = 0; i < pos; i++) {
1978                         lcme = &lcm->lcm_entries[i];
1979                         size += le32_to_cpu(lcme->lcme_size);
1980                 }
1981
1982                 offset = le32_to_cpu(lcm->lcm_entries[0].lcme_offset);
1983                 memmove(buf->lb_buf + offset + sizeof(*lcme),
1984                         buf->lb_buf + offset, size);
1985         }
1986
1987         /* 3. Recalculate the enter offset for the component [pos, count-1] */
1988         for (i = count - 1; i >= pos; i--) {
1989                 lcm->lcm_entries[i + 1] = lcm->lcm_entries[i];
1990                 lcm->lcm_entries[i + 1].lcme_offset =
1991                         cpu_to_le32(le32_to_cpu(lcm->lcm_entries[i + 1].
1992                                                 lcme_offset) + added);
1993         }
1994
1995         /* 4. Recalculate the enter offset for the component [0, pos) */
1996         for (i = 0; i < pos; i++) {
1997                 lcm->lcm_entries[i].lcme_offset =
1998                         cpu_to_le32(le32_to_cpu(lcm->lcm_entries[i].
1999                                                 lcme_offset) + sizeof(*lcme));
2000         }
2001
2002         offset = sizeof(*lcm) + sizeof(*lcme) * (count + 1) + size;
2003         /* 4. Insert the new component header (entry) at the slot 'pos'. */
2004         lcme = &lcm->lcm_entries[pos];
2005         lcme->lcme_id = cpu_to_le32(ol->ol_comp_id);
2006         lcme->lcme_flags = cpu_to_le32(LCME_FL_INIT);
2007         lcme->lcme_extent.e_start = cpu_to_le64(ol->ol_comp_start);
2008         lcme->lcme_extent.e_end = cpu_to_le64(ol->ol_comp_end);
2009         lcme->lcme_offset = cpu_to_le32(offset);
2010         lcme->lcme_size = cpu_to_le32(lov_mds_md_size(ol->ol_stripe_count,
2011                                                       LOV_MAGIC_V1));
2012
2013         if (ol->ol_stripe_count > 1)
2014                 pattern |= LOV_PATTERN_F_HOLE;
2015
2016         lmm = buf->lb_buf + offset;
2017         /* 5. Insert teh new component body at the 'offset'. */
2018         objs = __lfsck_layout_new_v1_lovea(lmm, lfsck_dto2fid(parent),
2019                                            ol->ol_stripe_size, ea_off,
2020                                            pattern, ol->ol_stripe_count);
2021
2022         /* 6. Update mirror related flags and version. */
2023         lfsck_layout_update_lcm(lcm, lcme, rec->lor_layout_version,
2024                                 rec->lor_range);
2025
2026         rc = lfsck_layout_refill_lovea(env, lfsck, handle, parent, cfid, buf,
2027                                        lmm, objs, LU_XATTR_REPLACE, ost_idx,
2028                                        le32_to_cpu(lcm->lcm_size));
2029
2030         CDEBUG(D_LFSCK, "%s: layout LFSCK assistant add new COMP for "
2031                DFID": parent "DFID", OST-index %u, stripe-index %u, "
2032                "stripe_size %u, stripe_count %u, comp_id %u, comp_start %llu, "
2033                "comp_end %llu, layout version %u, range %u, "
2034                "%s LOV EA hole: rc = %d\n",
2035                lfsck_lfsck2name(lfsck), PFID(cfid), PFID(lfsck_dto2fid(parent)),
2036                ost_idx, ea_off, ol->ol_stripe_size, ol->ol_stripe_count,
2037                ol->ol_comp_id, ol->ol_comp_start, ol->ol_comp_end,
2038                rec->lor_layout_version, rec->lor_range,
2039                le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_HOLE ?
2040                "with" : "without", rc);
2041
2042         RETURN(rc);
2043 }
2044
2045 static int lfsck_layout_extend_v1v3_lovea(const struct lu_env *env,
2046                                           struct lfsck_instance *lfsck,
2047                                           struct thandle *handle,
2048                                           struct ost_layout *ol,
2049                                           struct dt_object *parent,
2050                                           const struct lu_fid *cfid,
2051                                           struct lu_buf *buf, __u32 ost_idx,
2052                                           __u32 ea_off)
2053 {
2054         struct lov_mds_md_v1 *lmm = buf->lb_buf;
2055         struct lov_ost_data_v1 *objs;
2056         __u16 count = le16_to_cpu(lmm->lmm_stripe_count);
2057         __u32 magic = le32_to_cpu(lmm->lmm_magic);
2058         int size;
2059         int gap;
2060         int rc;
2061         ENTRY;
2062
2063         /* The original LOVEA maybe re-generated via old filter_fid, at
2064          * that time, we do not know the stripe count and stripe size. */
2065         if (ol->ol_stripe_count > count)
2066                 count = ol->ol_stripe_count;
2067         if (ol->ol_stripe_size != 0 &&
2068             ol->ol_stripe_size != le32_to_cpu(lmm->lmm_stripe_size))
2069                 lmm->lmm_stripe_size = cpu_to_le32(ol->ol_stripe_size);
2070
2071         if (magic == LOV_MAGIC_V1)
2072                 objs = &lmm->lmm_objects[count];
2073         else
2074                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[count];
2075
2076         gap = ea_off - count;
2077         if (gap >= 0)
2078                 count = ea_off + 1;
2079
2080         size = lov_mds_md_size(count, magic);
2081         LASSERTF(buf->lb_len >= size,
2082                  "buffer len %d is less than real size %d\n",
2083                  (int)buf->lb_len, size);
2084
2085         if (gap > 0) {
2086                 memset(objs, 0, gap * sizeof(*objs));
2087                 lmm->lmm_pattern |= cpu_to_le32(LOV_PATTERN_F_HOLE);
2088         }
2089
2090         lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2091         lmm->lmm_stripe_count = cpu_to_le16(count);
2092         objs += gap;
2093
2094         rc = lfsck_layout_refill_lovea(env, lfsck, handle, parent, cfid, buf,
2095                                 lmm, objs, LU_XATTR_REPLACE, ost_idx, size);
2096
2097         CDEBUG(D_LFSCK, "%s: layout LFSCK assistant extend layout EA for "
2098                DFID": parent "DFID", OST-index %u, stripe-index %u, "
2099                "stripe_size %u, stripe_count %u, comp_id %u, comp_start %llu, "
2100                "comp_end %llu, %s LOV EA hole: rc = %d\n",
2101                lfsck_lfsck2name(lfsck), PFID(cfid), PFID(lfsck_dto2fid(parent)),
2102                ost_idx, ea_off, ol->ol_stripe_size, ol->ol_stripe_count,
2103                ol->ol_comp_id, ol->ol_comp_start, ol->ol_comp_end,
2104                le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_HOLE ?
2105                "with" : "without", rc);
2106
2107         RETURN(rc);
2108 }
2109
2110 /**
2111  * \retval       +1: repaired
2112  * \retval        0: did nothing
2113  * \retval      -ve: on error
2114  */
2115 static int lfsck_layout_update_lovea(const struct lu_env *env,
2116                                      struct lfsck_instance *lfsck,
2117                                      struct thandle *handle,
2118                                      struct lu_orphan_rec_v3 *rec,
2119                                      struct dt_object *parent,
2120                                      const struct lu_fid *cfid,
2121                                      struct lu_buf *buf, int fl,
2122                                      __u32 ost_idx, __u32 ea_off)
2123 {
2124         struct ost_layout *ol = &rec->lor_layout;
2125         struct lov_mds_md_v1 *lmm = NULL;
2126         struct lov_ost_data_v1 *objs = NULL;
2127         int rc = 0;
2128         ENTRY;
2129
2130         if (ol->ol_comp_id != 0)
2131                 rc = lfsck_layout_new_comp_lovea(env, rec, parent, buf, ea_off,
2132                                                  &lmm, &objs);
2133         else
2134                 rc = lfsck_layout_new_v1_lovea(env, lfsck, &rec->lor_layout,
2135                                                parent, buf, ea_off, &lmm,
2136                                                &objs);
2137         if (rc > 0)
2138                 rc = lfsck_layout_refill_lovea(env, lfsck, handle, parent, cfid,
2139                                                buf, lmm, objs, fl, ost_idx, rc);
2140
2141         CDEBUG(D_LFSCK, "%s: layout LFSCK assistant created layout EA for "
2142                DFID": parent "DFID", OST-index %u, stripe-index %u, "
2143                "stripe_size %u, stripe_count %u, comp_id %u, comp_start %llu, "
2144                "comp_end %llu, layout version %u, range %u, fl %d, "
2145                "%s LOV EA hole: rc = %d\n",
2146                lfsck_lfsck2name(lfsck), PFID(cfid), PFID(lfsck_dto2fid(parent)),
2147                ost_idx, ea_off, ol->ol_stripe_size, ol->ol_stripe_count,
2148                ol->ol_comp_id, ol->ol_comp_start, ol->ol_comp_end,
2149                rec->lor_layout_version, rec->lor_range, fl,
2150                le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_HOLE ?
2151                "with" : "without", rc);
2152
2153         RETURN(rc);
2154 }
2155
2156 static int __lfsck_layout_update_pfid(const struct lu_env *env,
2157                                       struct dt_object *child,
2158                                       const struct lu_fid *pfid,
2159                                       const struct ost_layout *ol, __u32 offset,
2160                                       __u32 version, __u32 range)
2161 {
2162         struct dt_device        *dev    = lfsck_obj2dev(child);
2163         struct filter_fid       *ff     = &lfsck_env_info(env)->lti_ff;
2164         struct thandle          *handle;
2165         struct lu_buf            buf    = { NULL };
2166         int                      rc;
2167
2168         ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
2169         ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
2170         /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
2171          * MDT-object's FID::f_ver, instead it is the OST-object index in its
2172          * parent MDT-object's layout EA. */
2173         ff->ff_parent.f_stripe_idx = cpu_to_le32(offset);
2174         ost_layout_cpu_to_le(&ff->ff_layout, ol);
2175         ff->ff_layout_version = cpu_to_le32(version);
2176         ff->ff_range = cpu_to_le32(range);
2177         lfsck_buf_init(&buf, ff, sizeof(*ff));
2178
2179         handle = dt_trans_create(env, dev);
2180         if (IS_ERR(handle))
2181                 RETURN(PTR_ERR(handle));
2182
2183         rc = dt_declare_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
2184         if (rc != 0)
2185                 GOTO(stop, rc);
2186
2187         rc = dt_trans_start_local(env, dev, handle);
2188         if (rc != 0)
2189                 GOTO(stop, rc);
2190
2191         rc = dt_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
2192
2193         GOTO(stop, rc);
2194
2195 stop:
2196         dt_trans_stop(env, dev, handle);
2197
2198         return rc;
2199 }
2200
2201 /**
2202  * \retval       +1: repaired
2203  * \retval        0: did nothing
2204  * \retval      -ve: on error
2205  */
2206 static int lfsck_layout_update_pfid(const struct lu_env *env,
2207                                     struct lfsck_component *com,
2208                                     struct dt_object *parent,
2209                                     struct lu_fid *cfid,
2210                                     struct dt_device *cdev,
2211                                     struct lu_orphan_rec_v3 *rec, __u32 ea_off)
2212 {
2213         struct dt_object        *child;
2214         int                      rc     = 0;
2215         ENTRY;
2216
2217         child = lfsck_object_find_by_dev(env, cdev, cfid);
2218         if (IS_ERR(child))
2219                 RETURN(PTR_ERR(child));
2220
2221         rc = __lfsck_layout_update_pfid(env, child,
2222                                         lu_object_fid(&parent->do_lu),
2223                                         &rec->lor_layout, ea_off,
2224                                         rec->lor_layout_version,
2225                                         rec->lor_range);
2226         lfsck_object_put(env, child);
2227
2228         RETURN(rc == 0 ? 1 : rc);
2229 }
2230
2231 static int lfsck_lovea_size(struct ost_layout *ol, __u32 ea_off)
2232 {
2233         if (ol->ol_comp_id != 0)
2234                 return sizeof(struct lov_comp_md_v1) +
2235                        sizeof(struct lov_comp_md_entry_v1) +
2236                        lov_mds_md_size(ol->ol_stripe_count, LOV_MAGIC_V1);
2237
2238         if (ol->ol_stripe_count != 0)
2239                 return lov_mds_md_size(ol->ol_stripe_count, LOV_MAGIC_V1);
2240
2241         return lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
2242 }
2243
2244 /**
2245  * This function will create the MDT-object with the given (partial) LOV EA.
2246  *
2247  * Under some data corruption cases, the MDT-object of the file may be lost,
2248  * but its OST-objects, or some of them are there. The layout LFSCK needs to
2249  * re-create the MDT-object with the orphan OST-object(s) information.
2250  *
2251  * On the other hand, the LFSCK may has created some OST-object for repairing
2252  * dangling LOV EA reference, but as the LFSCK processing, it may find that
2253  * the old OST-object is there and should replace the former new created OST
2254  * object. Unfortunately, some others have modified such newly created object.
2255  * To keep the data (both new and old), the LFSCK will create MDT-object with
2256  * new FID to reference the original OST-object.
2257  *
2258  * \param[in] env       pointer to the thread context
2259  * \param[in] com       pointer to the lfsck component
2260  * \param[in] ltd       pointer to target device descriptor
2261  * \param[in] rec       pointer to the record for the orphan OST-object
2262  * \param[in] cfid      pointer to FID for the orphan OST-object
2263  * \param[in] infix     additional information, such as the FID for original
2264  *                      MDT-object and the stripe offset in the LOV EA
2265  * \param[in] type      the type for describing why the orphan MDT-object is
2266  *                      created. The rules are as following:
2267  *
2268  *  type "C":           Multiple OST-objects claim the same MDT-object and the
2269  *                      same slot in the layout EA. Then the LFSCK will create
2270  *                      new MDT-object(s) to hold the conflict OST-object(s).
2271  *
2272  *  type "N":           The orphan OST-object does not know which one was the
2273  *                      real parent MDT-object, so the LFSCK uses new FID for
2274  *                      its parent MDT-object.
2275  *
2276  *  type "R":           The orphan OST-object knows its parent MDT-object FID,
2277  *                      but does not know the position (the file name) in the
2278  *                      layout.
2279  *
2280  *  type "D":           The MDT-object is a directory, it may knows its parent
2281  *                      but because there is no valid linkEA, the LFSCK cannot
2282  *                      know where to put it back to the namespace.
2283  *  type "O":           The MDT-object has no linkEA, and there is no name
2284  *                      entry that references the MDT-object.
2285  *
2286  *  type "P":           The orphan object to be created was a parent directory
2287  *                      of some MDT-object which linkEA shows that the @orphan
2288  *                      object is missing.
2289  *
2290  * The orphan name will be like:
2291  * ${FID}-${infix}-${type}-${conflict_version}
2292  *
2293  * \param[in] ea_off    the stripe offset in the LOV EA
2294  *
2295  * \retval              positive on repaired something
2296  * \retval              0 if needs to repair nothing
2297  * \retval              negative error number on failure
2298  */
2299 static int lfsck_layout_recreate_parent(const struct lu_env *env,
2300                                         struct lfsck_component *com,
2301                                         struct lfsck_tgt_desc *ltd,
2302                                         struct lu_orphan_rec_v3 *rec,
2303                                         struct lu_fid *cfid,
2304                                         const char *infix,
2305                                         const char *type,
2306                                         __u32 ea_off)
2307 {
2308         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2309         struct dt_insert_rec            *dtrec  = &info->lti_dt_rec;
2310         char                            *name   = info->lti_key;
2311         struct lu_attr                  *la     = &info->lti_la2;
2312         struct dt_object_format         *dof    = &info->lti_dof;
2313         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2314         struct lu_fid                   *pfid   = &rec->lor_rec.lor_fid;
2315         struct lu_fid                   *tfid   = &info->lti_fid3;
2316         struct dt_device                *dev    = lfsck->li_bottom;
2317         struct dt_object                *lpf    = lfsck->li_lpf_obj;
2318         struct dt_object                *pobj   = NULL;
2319         struct dt_object                *cobj   = NULL;
2320         struct thandle                  *th     = NULL;
2321         struct lu_buf                   *ea_buf = &info->lti_big_buf;
2322         struct lu_buf                    lov_buf;
2323         struct lfsck_lock_handle        *llh    = &info->lti_llh;
2324         struct linkea_data               ldata  = { NULL };
2325         struct lu_buf                    linkea_buf;
2326         const struct lu_name            *pname;
2327         int                              size   = 0;
2328         int                              idx    = 0;
2329         int                              rc     = 0;
2330         ENTRY;
2331
2332         if (lfsck_is_dryrun(lfsck))
2333                 GOTO(log, rc = 0);
2334
2335         if (unlikely(lpf == NULL))
2336                 GOTO(log, rc = -ENXIO);
2337
2338         /* We use two separated transactions to repair the inconsistency.
2339          *
2340          * 1) create the MDT-object locally.
2341          * 2) update the OST-object's PFID EA if necessary.
2342          *
2343          * If 1) succeed, but 2) failed, then the OST-object's PFID EA will be
2344          * updated when the layout LFSCK run next time.
2345          *
2346          * If 1) failed, but 2) succeed, then such MDT-object will be re-created
2347          * when the layout LFSCK run next time. */
2348
2349         if (fid_is_zero(pfid)) {
2350                 rc = lfsck_fid_alloc(env, lfsck, pfid, false);
2351                 if (rc != 0)
2352                         GOTO(log, rc);
2353
2354                 cobj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, cfid);
2355                 if (IS_ERR(cobj))
2356                         GOTO(log, rc = PTR_ERR(cobj));
2357         }
2358
2359         pobj = lfsck_object_find_by_dev(env, dev, pfid);
2360         if (IS_ERR(pobj))
2361                 GOTO(log, rc = PTR_ERR(pobj));
2362
2363         LASSERT(infix != NULL);
2364         LASSERT(type != NULL);
2365
2366         memset(la, 0, sizeof(*la));
2367         la->la_uid = rec->lor_rec.lor_uid;
2368         la->la_gid = rec->lor_rec.lor_gid;
2369         la->la_mode = S_IFREG | S_IRUSR;
2370         la->la_valid = LA_MODE | LA_UID | LA_GID;
2371
2372         memset(dof, 0, sizeof(*dof));
2373         dof->dof_type = dt_mode_to_dft(S_IFREG);
2374         /* Because the dof->dof_reg.striped = 0, the LOD will not create
2375          * the stripe(s). The LFSCK will specify the LOV EA via
2376          * lfsck_layout_update_lovea(). */
2377
2378         size = lfsck_lovea_size(&rec->lor_layout, ea_off);
2379         if (ea_buf->lb_len < size) {
2380                 lu_buf_realloc(ea_buf, size);
2381                 if (ea_buf->lb_buf == NULL)
2382                         GOTO(log, rc = -ENOMEM);
2383         }
2384
2385 again:
2386         do {
2387                 snprintf(name, NAME_MAX, DFID"%s-%s-%d", PFID(pfid), infix,
2388                          type, idx++);
2389                 rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
2390                                (const struct dt_key *)name);
2391                 if (rc != 0 && rc != -ENOENT)
2392                         GOTO(log, rc);
2393         } while (rc == 0);
2394
2395         rc = lfsck_lock(env, lfsck, lfsck->li_lpf_obj, name, llh,
2396                         MDS_INODELOCK_UPDATE, LCK_PW);
2397         if (rc != 0)
2398                 GOTO(log, rc);
2399
2400         /* Re-check whether the name conflict with othrs after taken
2401          * the ldlm lock. */
2402         rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
2403                        (const struct dt_key *)name);
2404         if (unlikely(rc == 0)) {
2405                 lfsck_unlock(llh);
2406                 goto again;
2407         }
2408
2409         if (rc != -ENOENT)
2410                 GOTO(unlock, rc);
2411
2412         pname = lfsck_name_get_const(env, name, strlen(name));
2413         rc = linkea_links_new(&ldata, &lfsck_env_info(env)->lti_linkea_buf,
2414                               pname, lfsck_dto2fid(lfsck->li_lpf_obj));
2415         if (rc != 0)
2416                 GOTO(unlock, rc);
2417
2418         /* The 1st transaction. */
2419         th = dt_trans_create(env, dev);
2420         if (IS_ERR(th))
2421                 GOTO(unlock, rc = PTR_ERR(th));
2422
2423         rc = dt_declare_create(env, pobj, la, NULL, dof, th);
2424         if (rc != 0)
2425                 GOTO(stop, rc);
2426
2427         lfsck_buf_init(&lov_buf, ea_buf->lb_buf, size);
2428         rc = dt_declare_xattr_set(env, pobj, &lov_buf, XATTR_NAME_LOV,
2429                                   LU_XATTR_CREATE, th);
2430         if (rc != 0)
2431                 GOTO(stop, rc);
2432
2433         dtrec->rec_fid = pfid;
2434         dtrec->rec_type = S_IFREG;
2435         rc = dt_declare_insert(env, lpf,
2436                                (const struct dt_rec *)dtrec,
2437                                (const struct dt_key *)name, th);
2438         if (rc != 0)
2439                 GOTO(stop, rc);
2440
2441         lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
2442                        ldata.ld_leh->leh_len);
2443         rc = dt_declare_xattr_set(env, pobj, &linkea_buf,
2444                                   XATTR_NAME_LINK, 0, th);
2445         if (rc != 0)
2446                 GOTO(stop, rc);
2447
2448         rc = dt_trans_start_local(env, dev, th);
2449         if (rc != 0)
2450                 GOTO(stop, rc);
2451
2452         dt_write_lock(env, pobj, 0);
2453         rc = dt_create(env, pobj, la, NULL, dof, th);
2454         if (rc == 0)
2455                 rc = lfsck_layout_update_lovea(env, lfsck, th, rec, pobj, cfid,
2456                         &lov_buf, LU_XATTR_CREATE, ltd->ltd_index, ea_off);
2457         dt_write_unlock(env, pobj);
2458         if (rc < 0)
2459                 GOTO(stop, rc);
2460
2461         rc = dt_insert(env, lpf, (const struct dt_rec *)dtrec,
2462                        (const struct dt_key *)name, th);
2463         if (rc != 0)
2464                 GOTO(stop, rc);
2465
2466         rc = dt_xattr_set(env, pobj, &linkea_buf, XATTR_NAME_LINK, 0, th);
2467         if (rc == 0 && cobj != NULL) {
2468                 dt_trans_stop(env, dev, th);
2469                 th = NULL;
2470
2471                 /* The 2nd transaction. */
2472                 rc = __lfsck_layout_update_pfid(env, cobj, pfid,
2473                                                 &rec->lor_layout, ea_off,
2474                                                 rec->lor_layout_version,
2475                                                 rec->lor_range);
2476         }
2477
2478         GOTO(stop, rc);
2479
2480 stop:
2481         if (th != NULL)
2482                 dt_trans_stop(env, dev, th);
2483
2484 unlock:
2485         lfsck_unlock(llh);
2486
2487 log:
2488         if (cobj != NULL && !IS_ERR(cobj))
2489                 lfsck_object_put(env, cobj);
2490         if (pobj != NULL && !IS_ERR(pobj))
2491                 lfsck_object_put(env, pobj);
2492
2493         if (rc < 0)
2494                 CDEBUG(D_LFSCK, "%s layout LFSCK assistant failed to "
2495                        "recreate the lost MDT-object: parent "DFID
2496                        ", child "DFID", OST-index %u, stripe-index %u, "
2497                        "infix %s, type %s: rc = %d\n",
2498                        lfsck_lfsck2name(lfsck), PFID(pfid), PFID(cfid),
2499                        ltd->ltd_index, ea_off, infix, type, rc);
2500
2501         return rc >= 0 ? 1 : rc;
2502 }
2503
2504 static int lfsck_layout_master_conditional_destroy(const struct lu_env *env,
2505                                                    struct lfsck_component *com,
2506                                                    const struct lu_fid *fid,
2507                                                    __u32 index)
2508 {
2509         struct lfsck_thread_info *info  = lfsck_env_info(env);
2510         struct lfsck_request     *lr    = &info->lti_lr;
2511         struct lfsck_instance    *lfsck = com->lc_lfsck;
2512         struct lfsck_tgt_desc    *ltd;
2513         struct ptlrpc_request    *req;
2514         struct lfsck_request     *tmp;
2515         struct obd_export        *exp;
2516         int                       rc    = 0;
2517         ENTRY;
2518
2519         ltd = lfsck_tgt_get(&lfsck->li_ost_descs, index);
2520         if (unlikely(ltd == NULL))
2521                 RETURN(-ENXIO);
2522
2523         exp = ltd->ltd_exp;
2524         if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
2525                 GOTO(put, rc = -EOPNOTSUPP);
2526
2527         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
2528         if (req == NULL)
2529                 GOTO(put, rc = -ENOMEM);
2530
2531         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
2532         if (rc != 0) {
2533                 ptlrpc_request_free(req);
2534
2535                 GOTO(put, rc);
2536         }
2537
2538         memset(lr, 0, sizeof(*lr));
2539         lr->lr_event = LE_CONDITIONAL_DESTROY;
2540         lr->lr_active = LFSCK_TYPE_LAYOUT;
2541         lr->lr_fid = *fid;
2542
2543         tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
2544         *tmp = *lr;
2545         ptlrpc_request_set_replen(req);
2546
2547         rc = ptlrpc_queue_wait(req);
2548         ptlrpc_req_finished(req);
2549
2550         GOTO(put, rc);
2551
2552 put:
2553         lfsck_tgt_put(ltd);
2554
2555         return rc;
2556 }
2557
2558 static int lfsck_layout_slave_conditional_destroy(const struct lu_env *env,
2559                                                   struct lfsck_component *com,
2560                                                   struct lfsck_request *lr)
2561 {
2562         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2563         struct lu_attr                  *la     = &info->lti_la;
2564         union ldlm_policy_data          *policy = &info->lti_policy;
2565         struct ldlm_res_id              *resid  = &info->lti_resid;
2566         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2567         struct dt_device                *dev    = lfsck->li_bottom;
2568         struct lu_fid                   *fid    = &lr->lr_fid;
2569         struct dt_object                *obj;
2570         struct thandle                  *th     = NULL;
2571         struct lustre_handle             lh     = { 0 };
2572         __u64                            flags  = 0;
2573         int                              rc     = 0;
2574         ENTRY;
2575
2576         obj = lfsck_object_find_by_dev(env, dev, fid);
2577         if (IS_ERR(obj))
2578                 RETURN(PTR_ERR(obj));
2579
2580         dt_read_lock(env, obj, 0);
2581         if (dt_object_exists(obj) == 0 ||
2582             lfsck_is_dead_obj(obj)) {
2583                 dt_read_unlock(env, obj);
2584
2585                 GOTO(put, rc = -ENOENT);
2586         }
2587
2588         /* Get obj's attr without lock firstly. */
2589         rc = dt_attr_get(env, obj, la);
2590         dt_read_unlock(env, obj);
2591         if (rc != 0)
2592                 GOTO(put, rc);
2593
2594         if (likely(la->la_ctime != 0 || la->la_mode & S_ISUID))
2595                 GOTO(put, rc = -ETXTBSY);
2596
2597         /* Acquire extent lock on [0, EOF] to sync with all possible written. */
2598         LASSERT(lfsck->li_namespace != NULL);
2599
2600         memset(policy, 0, sizeof(*policy));
2601         policy->l_extent.end = OBD_OBJECT_EOF;
2602         ost_fid_build_resid(fid, resid);
2603         rc = ldlm_cli_enqueue_local(env, lfsck->li_namespace, resid,
2604                                     LDLM_EXTENT, policy, LCK_EX, &flags,
2605                                     ldlm_blocking_ast, ldlm_completion_ast,
2606                                     NULL, NULL, 0, LVB_T_NONE, NULL, &lh);
2607         if (rc != ELDLM_OK)
2608                 GOTO(put, rc = -EIO);
2609
2610         dt_write_lock(env, obj, 0);
2611         /* Get obj's attr within lock again. */
2612         rc = dt_attr_get(env, obj, la);
2613         if (rc != 0)
2614                 GOTO(unlock, rc);
2615
2616         if (la->la_ctime != 0)
2617                 GOTO(unlock, rc = -ETXTBSY);
2618
2619         th = dt_trans_create(env, dev);
2620         if (IS_ERR(th))
2621                 GOTO(unlock, rc = PTR_ERR(th));
2622
2623         rc = dt_declare_ref_del(env, obj, th);
2624         if (rc != 0)
2625                 GOTO(stop, rc);
2626
2627         rc = dt_declare_destroy(env, obj, th);
2628         if (rc != 0)
2629                 GOTO(stop, rc);
2630
2631         rc = dt_trans_start_local(env, dev, th);
2632         if (rc != 0)
2633                 GOTO(stop, rc);
2634
2635         rc = dt_ref_del(env, obj, th);
2636         if (rc != 0)
2637                 GOTO(stop, rc);
2638
2639         rc = dt_destroy(env, obj, th);
2640         if (rc == 0)
2641                 CDEBUG(D_LFSCK, "%s: layout LFSCK destroyed the empty "
2642                        "OST-object "DFID" that was created for reparing "
2643                        "dangling referenced case. But the original missing "
2644                        "OST-object is found now.\n",
2645                        lfsck_lfsck2name(lfsck), PFID(fid));
2646
2647         GOTO(stop, rc);
2648
2649 stop:
2650         dt_trans_stop(env, dev, th);
2651
2652 unlock:
2653         dt_write_unlock(env, obj);
2654         ldlm_lock_decref(&lh, LCK_EX);
2655
2656 put:
2657         lfsck_object_put(env, obj);
2658
2659         return rc;
2660 }
2661
2662 /**
2663  * Some OST-object has occupied the specified layout EA slot.
2664  * Such OST-object may be generated by the LFSCK when repair
2665  * dangling referenced MDT-object, which can be indicated by
2666  * attr::la_ctime == 0 but without S_ISUID in la_mode. If it
2667  * is true and such OST-object has not been modified yet, we
2668  * will replace it with the orphan OST-object; otherwise the
2669  * LFSCK will create new MDT-object to reference the orphan.
2670  *
2671  * \retval       +1: repaired
2672  * \retval        0: did nothing
2673  * \retval      -ve: on error
2674  */
2675 static int lfsck_layout_conflict_create(const struct lu_env *env,
2676                                         struct lfsck_component *com,
2677                                         struct lfsck_tgt_desc *ltd,
2678                                         struct lu_orphan_rec_v3 *rec,
2679                                         struct dt_object *parent,
2680                                         struct lu_fid *cfid,
2681                                         struct lu_buf *ea_buf,
2682                                         struct lov_mds_md_v1 *lmm,
2683                                         struct lov_ost_data_v1 *slot,
2684                                         __u32 ea_off, int lovea_size)
2685 {
2686         struct lfsck_thread_info *info          = lfsck_env_info(env);
2687         struct lu_fid            *cfid2         = &info->lti_fid2;
2688         struct ost_id            *oi            = &info->lti_oi;
2689         struct dt_device         *dev           = lfsck_obj2dev(parent);
2690         struct thandle           *th            = NULL;
2691         struct lustre_handle      lh            = { 0 };
2692         __u32                     ost_idx2      = le32_to_cpu(slot->l_ost_idx);
2693         int                       rc            = 0;
2694         ENTRY;
2695
2696         while (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val)) {
2697                 if (unlikely(!thread_is_running(&com->lc_lfsck->li_thread)))
2698                         RETURN(0);
2699         }
2700
2701         ostid_le_to_cpu(&slot->l_ost_oi, oi);
2702         rc = ostid_to_fid(cfid2, oi, ost_idx2);
2703         if (rc != 0)
2704                 GOTO(out, rc);
2705
2706         rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2707                               MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2708                               LCK_EX);
2709         if (rc != 0)
2710                 GOTO(out, rc);
2711
2712         rc = lfsck_layout_master_conditional_destroy(env, com, cfid2, ost_idx2);
2713
2714         /* If the conflict OST-obejct is not created for fixing dangling
2715          * referenced MDT-object in former LFSCK check/repair, or it has
2716          * been modified by others, then we cannot destroy it. Re-create
2717          * a new MDT-object for the orphan OST-object. */
2718         if (rc == -ETXTBSY) {
2719                 /* No need the layout lock on the original parent. */
2720                 lfsck_ibits_unlock(&lh, LCK_EX);
2721
2722                 fid_zero(&rec->lor_rec.lor_fid);
2723                 snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2724                          "-"DFID"-%x", PFID(lu_object_fid(&parent->do_lu)),
2725                          ea_off);
2726                 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2727                                                 info->lti_tmpbuf, "C", ea_off);
2728
2729                 RETURN(rc);
2730         }
2731
2732         if (rc != 0 && rc != -ENOENT)
2733                 GOTO(unlock, rc);
2734
2735         if (lfsck_is_dryrun(com->lc_lfsck))
2736                 GOTO(unlock, rc = 0);
2737
2738         th = dt_trans_create(env, dev);
2739         if (IS_ERR(th))
2740                 GOTO(unlock, rc = PTR_ERR(th));
2741
2742         rc = dt_declare_xattr_set(env, parent, ea_buf, XATTR_NAME_LOV,
2743                                   LU_XATTR_REPLACE, th);
2744         if (rc != 0)
2745                 GOTO(stop, rc);
2746
2747         rc = dt_trans_start_local(env, dev, th);
2748         if (rc != 0)
2749                 GOTO(stop, rc);
2750
2751         dt_write_lock(env, parent, 0);
2752         lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2753         rc = lfsck_layout_refill_lovea(env, com->lc_lfsck, th, parent, cfid,
2754                                        ea_buf, lmm, slot, LU_XATTR_REPLACE,
2755                                        ltd->ltd_index, lovea_size);
2756         dt_write_unlock(env, parent);
2757
2758         GOTO(stop, rc);
2759
2760 stop:
2761         dt_trans_stop(env, dev, th);
2762
2763 unlock:
2764         lfsck_ibits_unlock(&lh, LCK_EX);
2765
2766 out:
2767         CDEBUG(D_LFSCK, "%s: layout LFSCK assistant replaced the conflict "
2768                "OST-object "DFID" on the OST %x with the orphan "DFID" on "
2769                "the OST %x: parent "DFID", stripe-index %u: rc = %d\n",
2770                lfsck_lfsck2name(com->lc_lfsck), PFID(cfid2), ost_idx2,
2771                PFID(cfid), ltd->ltd_index, PFID(lfsck_dto2fid(parent)),
2772                ea_off, rc);
2773
2774         return rc >= 0 ? 1 : rc;
2775 }
2776
2777 /**
2778  * \retval       +1: repaired
2779  * \retval        0: did nothing
2780  * \retval      -ve: on error
2781  */
2782 static int lfsck_layout_recreate_lovea(const struct lu_env *env,
2783                                        struct lfsck_component *com,
2784                                        struct lfsck_tgt_desc *ltd,
2785                                        struct lu_orphan_rec_v3 *rec,
2786                                        struct dt_object *parent,
2787                                        struct lu_fid *cfid,
2788                                        __u32 ost_idx, __u32 ea_off)
2789 {
2790         struct lfsck_thread_info *info          = lfsck_env_info(env);
2791         struct lu_buf            *buf           = &info->lti_big_buf;
2792         struct lu_fid            *fid           = &info->lti_fid2;
2793         struct ost_id            *oi            = &info->lti_oi;
2794         struct lfsck_instance    *lfsck         = com->lc_lfsck;
2795         struct dt_device         *dt            = lfsck_obj2dev(parent);
2796         struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
2797         struct ost_layout        *ol            = &rec->lor_layout;
2798         struct lov_comp_md_v1    *lcm           = NULL;
2799         struct lov_comp_md_entry_v1 *lcme       = NULL;
2800         struct thandle           *handle        = NULL;
2801         size_t                    lovea_size;
2802         struct lov_mds_md_v1     *lmm;
2803         struct lov_ost_data_v1   *objs;
2804         struct lustre_handle      lh            = { 0 };
2805         __u32                     magic;
2806         __u32 flags = 0;
2807         int                       fl            = 0;
2808         int                       rc            = 0;
2809         int                       rc1;
2810         int                       i;
2811         int pos = 0;
2812         __u16 count;
2813         bool locked = false;
2814         bool new_mirror = true;
2815         ENTRY;
2816
2817         if (lfsck_is_dryrun(lfsck))
2818                 RETURN(0);
2819
2820         rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2821                               MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2822                               LCK_EX);
2823         if (rc != 0) {
2824                 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant failed to recreate "
2825                        "LOV EA for "DFID": parent "DFID", OST-index %u, "
2826                        "stripe-index %u, comp_id %u, comp_start %llu, "
2827                        "comp_end %llu, layout version %u, range %u: rc = %d\n",
2828                        lfsck_lfsck2name(lfsck), PFID(cfid),
2829                        PFID(lfsck_dto2fid(parent)), ost_idx, ea_off,
2830                        ol->ol_comp_id, ol->ol_comp_start,
2831                        ol->ol_comp_end, rec->lor_layout_version,
2832                        rec->lor_range, rc);
2833
2834                 RETURN(rc);
2835         }
2836
2837 again:
2838         if (locked) {
2839                 dt_write_unlock(env, parent);
2840                 locked = false;
2841         }
2842
2843         if (handle != NULL) {
2844                 dt_trans_stop(env, dt, handle);
2845                 handle = NULL;
2846         }
2847
2848         if (rc < 0)
2849                 GOTO(unlock_layout, rc);
2850
2851         lovea_size = rc;
2852         if (buf->lb_len < lovea_size) {
2853                 lu_buf_realloc(buf, lovea_size);
2854                 if (buf->lb_buf == NULL)
2855                         GOTO(unlock_layout, rc = -ENOMEM);
2856         }
2857
2858         if (!(bk->lb_param & LPF_DRYRUN)) {
2859                 handle = dt_trans_create(env, dt);
2860                 if (IS_ERR(handle))
2861                         GOTO(unlock_layout, rc = PTR_ERR(handle));
2862
2863                 rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2864                                           fl, handle);
2865                 if (rc != 0)
2866                         GOTO(stop, rc);
2867
2868                 rc = dt_trans_start_local(env, dt, handle);
2869                 if (rc != 0)
2870                         GOTO(stop, rc);
2871         }
2872
2873         dt_write_lock(env, parent, 0);
2874         locked = true;
2875         rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV);
2876         if (rc == -ERANGE) {
2877                 rc = dt_xattr_get(env, parent, &LU_BUF_NULL, XATTR_NAME_LOV);
2878                 LASSERT(rc != 0);
2879                 goto again;
2880         } else if (rc == -ENODATA || rc == 0) {
2881                 lovea_size = lfsck_lovea_size(ol, ea_off);
2882                 /* If the declared is not big enough, re-try. */
2883                 if (buf->lb_len < lovea_size) {
2884                         rc = lovea_size;
2885                         goto again;
2886                 }
2887                 fl = LU_XATTR_CREATE;
2888         } else if (rc < 0) {
2889                 GOTO(unlock_parent, rc);
2890         } else if (unlikely(buf->lb_len == 0)) {
2891                 goto again;
2892         } else {
2893                 fl = LU_XATTR_REPLACE;
2894                 lovea_size = rc;
2895         }
2896
2897         if (fl == LU_XATTR_CREATE) {
2898                 if (bk->lb_param & LPF_DRYRUN)
2899                         GOTO(unlock_parent, rc = 1);
2900
2901                 LASSERT(buf->lb_len >= lovea_size);
2902
2903                 rc = lfsck_layout_update_lovea(env, lfsck, handle, rec, parent,
2904                                                cfid, buf, fl, ost_idx, ea_off);
2905
2906                 GOTO(unlock_parent, rc);
2907         }
2908
2909         lmm = buf->lb_buf;
2910         rc1 = lfsck_layout_verify_header(parent, lmm, lovea_size);
2911
2912         /* If the LOV EA crashed, the rebuild it. */
2913         if (rc1 == -EINVAL) {
2914                 if (bk->lb_param & LPF_DRYRUN)
2915                         GOTO(unlock_parent, rc = 1);
2916
2917                 LASSERT(buf->lb_len >= lovea_size);
2918
2919                 rc = lfsck_layout_update_lovea(env, lfsck, handle, rec, parent,
2920                                                cfid, buf, fl, ost_idx, ea_off);
2921
2922                 GOTO(unlock_parent, rc);
2923         }
2924
2925         /* For other unknown magic/pattern, keep the current LOV EA. */
2926         if (rc1 == -EOPNOTSUPP)
2927                 GOTO(unlock_parent, rc1 = 0);
2928
2929         if (rc1)
2930                 GOTO(unlock_parent, rc = rc1);
2931
2932         magic = le32_to_cpu(lmm->lmm_magic);
2933         if (magic == LOV_MAGIC_COMP_V1) {
2934                 __u64 start;
2935                 __u64 end;
2936                 __u16 mirror_id0 = mirror_id_of(ol->ol_comp_id);
2937                 __u16 mirror_id1;
2938
2939                 if (bk->lb_param & LPF_DRYRUN)
2940                         GOTO(unlock_parent, rc = 1);
2941
2942                 lcm = buf->lb_buf;
2943                 count = le16_to_cpu(lcm->lcm_entry_count);
2944                 for (i = 0; i < count; pos = ++i) {
2945                         lcme = &lcm->lcm_entries[i];
2946                         start = le64_to_cpu(lcme->lcme_extent.e_start);
2947                         end = le64_to_cpu(lcme->lcme_extent.e_end);
2948                         mirror_id1 = mirror_id_of(le32_to_cpu(lcme->lcme_id));
2949
2950                         if (mirror_id0 > mirror_id1)
2951                                 continue;
2952
2953                         if (mirror_id0 < mirror_id1)
2954                                 break;
2955
2956                         new_mirror = false;
2957                         if (end <= ol->ol_comp_start)
2958                                 continue;
2959
2960                         if (start >= ol->ol_comp_end)
2961                                 break;
2962
2963                         lmm = buf->lb_buf + le32_to_cpu(lcme->lcme_offset);
2964                         magic = le32_to_cpu(lmm->lmm_magic);
2965                         flags = le32_to_cpu(lcme->lcme_flags);
2966                         goto further;
2967                 }
2968
2969                 rc = lfsck_layout_add_comp(env, lfsck, handle, rec, parent,
2970                                 cfid, buf, ost_idx, ea_off, pos, new_mirror);
2971
2972                 GOTO(unlock_parent, rc);
2973         }
2974
2975 further:
2976         count = le16_to_cpu(lmm->lmm_stripe_count);
2977         if (count == 0)
2978                 GOTO(unlock_parent, rc = -EINVAL);
2979         LASSERT(count > 0);
2980
2981         /* Exceed the current end of MDT-object layout EA. Then extend it. */
2982         if (count <= ea_off) {
2983                 if (bk->lb_param & LPF_DRYRUN)
2984                         GOTO(unlock_parent, rc = 1);
2985
2986                 lovea_size = lov_mds_md_size(ea_off + 1, magic);
2987                 /* If the declared is not big enough, re-try. */
2988                 if (buf->lb_len < lovea_size) {
2989                         rc = lovea_size;
2990                         goto again;
2991                 }
2992
2993                 if (lcm) {
2994                         LASSERT(lcme);
2995
2996                         lcme->lcme_flags = cpu_to_le32(flags | LCME_FL_INIT);
2997                         lfsck_layout_update_lcm(lcm, lcme,
2998                                                 rec->lor_layout_version,
2999                                                 rec->lor_range);
3000                 }
3001
3002                 rc = lfsck_layout_extend_v1v3_lovea(env, lfsck, handle, ol,
3003                                         parent, cfid, buf, ost_idx, ea_off);
3004
3005                 GOTO(unlock_parent, rc);
3006         }
3007
3008         LASSERTF(rc > 0, "invalid rc = %d\n", rc);
3009
3010         if (magic == LOV_MAGIC_V1) {
3011                 objs = &lmm->lmm_objects[0];
3012         } else {
3013                 LASSERT(magic == LOV_MAGIC_V3);
3014                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
3015         }
3016
3017         for (i = 0; i < count; i++, objs++) {
3018                 /* The MDT-object was created via lfsck_layout_recover_create()
3019                  * by others before, and we fill the dummy layout EA. */
3020                 if ((lcme && !(flags & LCME_FL_INIT)) ||
3021                      lovea_slot_is_dummy(objs)) {
3022                         if (i != ea_off)
3023                                 continue;
3024
3025                         if (bk->lb_param & LPF_DRYRUN)
3026                                 GOTO(unlock_parent, rc = 1);
3027
3028                         lmm->lmm_layout_gen =
3029                             cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
3030                         if (lcme) {
3031                                 LASSERT(lcm);
3032
3033                                 if (le32_to_cpu(lmm->lmm_stripe_size) !=
3034                                         ol->ol_stripe_size ||
3035                                     le16_to_cpu(lmm->lmm_stripe_count) !=
3036                                         ol->ol_stripe_count ||
3037                                     le64_to_cpu(lcme->lcme_extent.e_start) !=
3038                                         ol->ol_comp_start ||
3039                                     le64_to_cpu(lcme->lcme_extent.e_end) !=
3040                                         ol->ol_comp_end) {
3041                                         CDEBUG(D_LFSCK, "%s: found invalid "
3042                                         "component for "DFID ": parent "DFID
3043                                         ", stripe-index %u, stripe_size %u, "
3044                                         "stripe_count %u, comp_id %u, "
3045                                         "comp_start %llu, comp_end %llu, "
3046                                         "cur_stripe_size %u, "
3047                                         "cur_stripe_count %u, "
3048                                         "cur_comp_start %llu, "
3049                                         "cur_comp_end %llu\n",
3050                                         lfsck_lfsck2name(lfsck), PFID(cfid),
3051                                         PFID(lfsck_dto2fid(parent)), ea_off,
3052                                         ol->ol_stripe_size,
3053                                         ol->ol_stripe_count, ol->ol_comp_id,
3054                                         ol->ol_comp_start, ol->ol_comp_end,
3055                                         le32_to_cpu(lmm->lmm_stripe_size),
3056                                         le16_to_cpu(lmm->lmm_stripe_count),
3057                                         le64_to_cpu(lcme->lcme_extent.e_start),
3058                                         le64_to_cpu(lcme->lcme_extent.e_end));
3059
3060                                         GOTO(unlock_parent, rc = -EINVAL);
3061                                 }
3062
3063                                 lovea_size = le32_to_cpu(lcm->lcm_size);
3064                                 lcme->lcme_flags = cpu_to_le32(flags |
3065                                                                LCME_FL_INIT);
3066                                 lfsck_layout_update_lcm(lcm, lcme,
3067                                                         rec->lor_layout_version,
3068                                                         rec->lor_range);
3069                         }
3070
3071                         LASSERTF(buf->lb_len >= lovea_size,
3072                                  "buffer len %d is less than real size %d\n",
3073                                  (int)buf->lb_len, (int)lovea_size);
3074
3075                         rc = lfsck_layout_refill_lovea(env, lfsck, handle,
3076                                                 parent, cfid, buf, lmm, objs,
3077                                                 fl, ost_idx, lovea_size);
3078
3079                         CDEBUG(D_LFSCK, "%s layout LFSCK assistant fill "
3080                                "dummy layout slot for "DFID": parent "DFID
3081                                ", OST-index %u, stripe-index %u: rc = %d\n",
3082                                lfsck_lfsck2name(lfsck), PFID(cfid),
3083                                PFID(lfsck_dto2fid(parent)), ost_idx, i, rc);
3084
3085                         GOTO(unlock_parent, rc);
3086                 }
3087
3088                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
3089                 rc = ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
3090                 if (rc != 0) {
3091                         CDEBUG(D_LFSCK, "%s: the parent "DFID" contains "
3092                                "invalid layout EA at the slot %d, index %u\n",
3093                                lfsck_lfsck2name(lfsck),
3094                                PFID(lfsck_dto2fid(parent)), i,
3095                                le32_to_cpu(objs->l_ost_idx));
3096
3097                         GOTO(unlock_parent, rc);
3098                 }
3099
3100                 /* It should be rare case, the slot is there, but the LFSCK
3101                  * does not handle it during the first-phase cycle scanning. */
3102                 if (unlikely(lu_fid_eq(fid, cfid))) {
3103                         if (i == ea_off) {
3104                                 GOTO(unlock_parent, rc = 0);
3105                         } else {
3106                                 /* Rare case that the OST-object index
3107                                  * does not match the parent MDT-object
3108                                  * layout EA. We trust the later one. */
3109                                 if (bk->lb_param & LPF_DRYRUN)
3110                                         GOTO(unlock_parent, rc = 1);
3111
3112                                 dt_write_unlock(env, parent);
3113                                 if (handle != NULL)
3114                                         dt_trans_stop(env, dt, handle);
3115                                 lfsck_ibits_unlock(&lh, LCK_EX);
3116                                 rc = lfsck_layout_update_pfid(env, com, parent,
3117                                                         cfid, ltd->ltd_tgt,
3118                                                         rec, i);
3119
3120                                 CDEBUG(D_LFSCK, "%s layout LFSCK assistant "
3121                                        "updated OST-object's pfid for "DFID
3122                                        ": parent "DFID", OST-index %u, "
3123                                        "stripe-index %u: rc = %d\n",
3124                                        lfsck_lfsck2name(lfsck), PFID(cfid),
3125                                        PFID(lfsck_dto2fid(parent)),
3126                                        ltd->ltd_index, i, rc);
3127
3128                                 RETURN(rc);
3129                         }
3130                 }
3131         }
3132
3133         /* The MDT-object exists, but related layout EA slot is occupied
3134          * by others. */
3135         if (bk->lb_param & LPF_DRYRUN)
3136                 GOTO(unlock_parent, rc = 1);
3137
3138         dt_write_unlock(env, parent);
3139         if (handle != NULL)
3140                 dt_trans_stop(env, dt, handle);
3141         lfsck_ibits_unlock(&lh, LCK_EX);
3142         if (magic == LOV_MAGIC_V1)
3143                 objs = &lmm->lmm_objects[ea_off];
3144         else
3145                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[ea_off];
3146         rc = lfsck_layout_conflict_create(env, com, ltd, rec, parent, cfid,
3147                                           buf, lmm, objs, ea_off, lovea_size);
3148
3149         RETURN(rc);
3150
3151 unlock_parent:
3152         if (locked)
3153                 dt_write_unlock(env, parent);
3154
3155 stop:
3156         if (handle != NULL)
3157                 dt_trans_stop(env, dt, handle);
3158
3159 unlock_layout:
3160         lfsck_ibits_unlock(&lh, LCK_EX);
3161
3162         return rc;
3163 }
3164
3165 static int lfsck_layout_scan_orphan_one(const struct lu_env *env,
3166                                         struct lfsck_component *com,
3167                                         struct lfsck_tgt_desc *ltd,
3168                                         struct lu_orphan_rec_v3 *rec,
3169                                         struct lu_fid *cfid)
3170 {
3171         struct lfsck_layout     *lo     = com->lc_file_ram;
3172         struct lu_fid           *pfid   = &rec->lor_rec.lor_fid;
3173         struct dt_object        *parent = NULL;
3174         __u32                    ea_off = pfid->f_stripe_idx;
3175         int                      rc     = 0;
3176         ENTRY;
3177
3178         if (!fid_is_sane(cfid))
3179                 GOTO(out, rc = -EINVAL);
3180
3181         pfid->f_ver = 0;
3182         if (fid_is_zero(pfid)) {
3183                 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
3184                                                   "", "N", ea_off);
3185                 GOTO(out, rc);
3186         }
3187
3188         if (!fid_is_sane(pfid))
3189                 GOTO(out, rc = -EINVAL);
3190
3191         parent = lfsck_object_find_by_dev(env, com->lc_lfsck->li_bottom, pfid);
3192         if (IS_ERR(parent))
3193                 GOTO(out, rc = PTR_ERR(parent));
3194
3195         if (unlikely(dt_object_remote(parent) != 0))
3196                 GOTO(put, rc = -EXDEV);
3197
3198         if (dt_object_exists(parent) == 0) {
3199                 lfsck_object_put(env, parent);
3200                 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
3201                                                   "", "R", ea_off);
3202                 GOTO(out, rc);
3203         }
3204
3205         if (!S_ISREG(lu_object_attr(&parent->do_lu)))
3206                 GOTO(put, rc = -EISDIR);
3207
3208         /* The orphan OST-object claims to be the parent's stripe, then
3209          * related dangling record in the trace file is meaningless. */
3210         rc = lfsck_layout_del_dangling_rec(env, com, pfid,
3211                                            rec->lor_layout.ol_comp_id, ea_off);
3212         if (rc && rc != -ENOENT)
3213                 GOTO(put, rc);
3214
3215         rc = lfsck_layout_recreate_lovea(env, com, ltd, rec, parent, cfid,
3216                                          ltd->ltd_index, ea_off);
3217
3218         GOTO(put, rc);
3219
3220 put:
3221         if (rc <= 0)
3222                 lfsck_object_put(env, parent);
3223         else
3224                 /* The layout EA is changed, need to be reloaded next time. */
3225                 dt_object_put_nocache(env, parent);
3226
3227 out:
3228         down_write(&com->lc_sem);
3229         com->lc_new_scanned++;
3230         com->lc_new_checked++;
3231         if (rc > 0) {
3232                 lo->ll_objs_repaired[LLIT_ORPHAN - 1]++;
3233                 rc = 0;
3234         } else if (rc < 0) {
3235                 lo->ll_objs_failed_phase2++;
3236         }
3237         up_write(&com->lc_sem);
3238
3239         return rc;
3240 }
3241
3242 static int lfsck_layout_scan_orphan(const struct lu_env *env,
3243                                     struct lfsck_component *com,
3244                                     struct lfsck_tgt_desc *ltd)
3245 {
3246         struct lfsck_assistant_data     *lad    = com->lc_data;
3247         struct lfsck_instance           *lfsck  = com->lc_lfsck;
3248         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
3249         struct lfsck_thread_info        *info   = lfsck_env_info(env);
3250         struct lu_fid                   *fid    = &info->lti_fid;
3251         struct dt_object                *obj;
3252         const struct dt_it_ops          *iops;
3253         struct dt_it                    *di;
3254         int                              rc     = 0;
3255         ENTRY;
3256
3257         CDEBUG(D_LFSCK, "%s: layout LFSCK assistant starts the orphan "
3258                "scanning for OST%04x\n",
3259                lfsck_lfsck2name(lfsck), ltd->ltd_index);
3260
3261         if (cfs_bitmap_check(lad->lad_bitmap, ltd->ltd_index)) {
3262                 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant skip the orphan "
3263                        "scanning for OST%04x\n",
3264                        lfsck_lfsck2name(lfsck), ltd->ltd_index);
3265
3266                 RETURN(0);
3267         }
3268
3269         fid->f_seq = fid_idif_seq(0, ltd->ltd_index);
3270         fid->f_oid = fid->f_ver = 0;
3271
3272         obj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, fid);
3273         if (unlikely(IS_ERR(obj)))
3274                 GOTO(log, rc = PTR_ERR(obj));
3275
3276         rc = obj->do_ops->do_index_try(env, obj,
3277                                        &dt_lfsck_layout_orphan_features);
3278         if (rc != 0)
3279                 GOTO(put, rc);
3280
3281         iops = &obj->do_index_ops->dio_it;
3282         di = iops->init(env, obj, 0);
3283         if (IS_ERR(di))
3284                 GOTO(put, rc = PTR_ERR(di));
3285
3286         rc = iops->load(env, di, 0);
3287         if (rc == -ESRCH) {
3288                 /* -ESRCH means that the orphan OST-objects rbtree has been
3289                  * cleanup because of the OSS server restart or other errors. */
3290                 lfsck_lad_set_bitmap(env, com, ltd->ltd_index);
3291                 GOTO(fini, rc);
3292         }
3293
3294         if (rc == 0)
3295                 rc = iops->next(env, di);
3296         else if (rc > 0)
3297                 rc = 0;
3298
3299         if (rc < 0)
3300                 GOTO(fini, rc);
3301
3302         if (rc > 0)
3303                 GOTO(fini, rc = 0);
3304
3305         do {
3306                 struct dt_key           *key;
3307                 struct lu_orphan_rec_v3 *rec = &info->lti_rec;
3308
3309                 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
3310                     unlikely(!thread_is_running(&lfsck->li_thread)))
3311                         break;
3312
3313                 key = iops->key(env, di);
3314                 com->lc_fid_latest_scanned_phase2 = *(struct lu_fid *)key;
3315                 /* Remote target OST may be runnning old LFSCK */
3316                 memset(rec, 0, sizeof(*rec));
3317                 rc = iops->rec(env, di, (struct dt_rec *)rec, 0);
3318                 if (rc == 0)
3319                         rc = lfsck_layout_scan_orphan_one(env, com, ltd, rec,
3320                                         &com->lc_fid_latest_scanned_phase2);
3321                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
3322                         GOTO(fini, rc);
3323
3324                 lfsck_control_speed_by_self(com);
3325                 do {
3326                         rc = iops->next(env, di);
3327                 } while (rc < 0 && !(bk->lb_param & LPF_FAILOUT));
3328         } while (rc == 0);
3329
3330         GOTO(fini, rc);
3331
3332 fini:
3333         iops->put(env, di);
3334         iops->fini(env, di);
3335 put:
3336         lfsck_object_put(env, obj);
3337
3338 log:
3339         CDEBUG(D_LFSCK, "%s: layout LFSCK assistant finished the orphan "
3340                "scanning for OST%04x: rc = %d\n",
3341                lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
3342
3343         return rc > 0 ? 0 : rc;
3344 }
3345
3346 static int lfsck_lov2layout(struct lov_mds_md_v1 *lmm, struct filter_fid *ff,
3347                             __u32 comp_id)
3348 {
3349         struct ost_layout *ol = &ff->ff_layout;
3350         __u32 magic = le32_to_cpu(lmm->lmm_magic);
3351         int rc = 0;
3352         ENTRY;
3353
3354         if (magic == LOV_MAGIC_V1 || magic == LOV_MAGIC_V3) {
3355                 ol->ol_stripe_size = lmm->lmm_stripe_size;
3356                 ol->ol_stripe_count = lmm->lmm_stripe_count;
3357                 ol->ol_comp_start = 0;
3358                 ol->ol_comp_end = 0;
3359                 ol->ol_comp_id = 0;
3360                 ff->ff_layout_version = 0;
3361                 ff->ff_range = 0;
3362         } else if (magic == LOV_MAGIC_COMP_V1) {
3363                 struct lov_comp_md_v1 *lcm = (struct lov_comp_md_v1 *)lmm;
3364                 struct lov_comp_md_entry_v1 *lcme = NULL;
3365                 __u16 count = le16_to_cpu(lcm->lcm_entry_count);
3366                 int i;
3367
3368                 for (i = 0; i < count; i++) {
3369                         lcme = &lcm->lcm_entries[i];
3370                         if (le32_to_cpu(lcme->lcme_id) == comp_id) {
3371                                 LASSERT(le32_to_cpu(lcme->lcme_flags) &
3372                                         LCME_FL_INIT);
3373
3374                                 break;
3375                         }
3376                 }
3377
3378                 /* The comp has been removed, do nothing. */
3379                 if (i == count)
3380                         GOTO(out, rc = 1);
3381
3382                 lmm = (void *)lmm + le32_to_cpu(lcme->lcme_offset);
3383                 ol->ol_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
3384                 ol->ol_stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
3385                 ol->ol_comp_start = le64_to_cpu(lcme->lcme_extent.e_start);
3386                 ol->ol_comp_end = le64_to_cpu(lcme->lcme_extent.e_end);
3387                 ol->ol_comp_id = le32_to_cpu(lcme->lcme_id);
3388                 ff->ff_layout_version = le32_to_cpu(lcme->lcme_layout_gen);
3389                 ff->ff_range = 0;
3390         } else {
3391                 GOTO(out, rc = -EINVAL);
3392         }
3393
3394         EXIT;
3395
3396 out:
3397         return rc;
3398 }
3399
3400 /**
3401  * Repair the MDT-object with dangling LOV EA reference.
3402  *
3403  * we need to repair the inconsistency according to the users' requirement:
3404  *
3405  * 1) Keep the inconsistency there and report the inconsistency case,
3406  *    then give the chance to the application to find related issues,
3407  *    and the users can make the decision about how to handle it with
3408  *    more human knownledge. (by default)
3409  *
3410  * 2) Re-create the missing OST-object with the FID/owner information.
3411  *
3412  * \param[in] env       pointer to the thread context
3413  * \param[in] com       the layout LFSCK component
3414  * \param[in] parent    the MDT-object with dangling LOV EA reference
3415  * \param[in] child     the OST-object to be created
3416  * \param[in] comp_id   the component ID of the OST-object in the LOV EA
3417  * \param[in] ea_off    the offset of the OST-object in the LOV EA
3418  * \param[in] ost_idx   the index of OST on which the OST-object resides
3419  *
3420  * \retval              +1 for repair successfully
3421  * \retval              0 for did nothing
3422  * \retval              negative error number on failure
3423  */
3424 static int __lfsck_layout_repair_dangling(const struct lu_env *env,
3425                                           struct lfsck_component *com,
3426                                           struct dt_object *parent,
3427                                           struct dt_object *child,
3428                                           __u32 comp_id, __u32 ea_off,
3429                                           __u32 ost_idx, bool log)
3430 {
3431         struct lfsck_thread_info *info = lfsck_env_info(env);
3432         struct filter_fid *ff = &info->lti_ff;
3433         struct dt_object_format *dof = &info->lti_dof;
3434         struct lu_attr *la = &info->lti_la;
3435         struct lfsck_instance *lfsck = com->lc_lfsck;
3436         struct dt_device *dev = lfsck_obj2dev(child);
3437         const struct lu_fid *pfid = lfsck_dto2fid(parent);
3438         const struct lu_fid *cfid = lfsck_dto2fid(child);
3439         struct lu_buf *tbuf = &info->lti_big_buf;
3440         struct thandle *handle;
3441         struct lu_buf *buf;
3442         struct lustre_handle lh = { 0 };
3443         int rc;
3444         ENTRY;
3445
3446         if (!(lfsck->li_bookmark_ram.lb_param & LPF_CREATE_OSTOBJ))
3447                 GOTO(log, rc = 1);
3448
3449         rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
3450                               MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
3451                               LCK_EX);
3452         if (rc != 0)
3453                 GOTO(log, rc);
3454
3455         rc = dt_attr_get(env, parent, la);
3456         if (rc != 0)
3457                 GOTO(unlock1, rc);
3458
3459         la->la_mode = S_IFREG | 0666;
3460         la->la_atime = la->la_mtime = la->la_ctime = 0;
3461         la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
3462                        LA_ATIME | LA_MTIME | LA_CTIME;
3463         memset(dof, 0, sizeof(*dof));
3464         ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
3465         ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
3466         /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
3467          * MDT-object's FID::f_ver, instead it is the OST-object index in its
3468          * parent MDT-object's layout EA. */
3469         ff->ff_parent.f_stripe_idx = cpu_to_le32(ea_off);
3470
3471         rc = lfsck_layout_get_lovea(env, parent, tbuf);
3472         if (unlikely(rc == -ENODATA))
3473                 rc = 0;
3474         if (rc <= 0)
3475                 GOTO(unlock1, rc);
3476
3477         rc = lfsck_lov2layout(tbuf->lb_buf, ff, comp_id);
3478         if (rc)
3479                 GOTO(unlock1, rc);
3480
3481         buf = lfsck_buf_get(env, ff, sizeof(struct filter_fid));
3482         handle = dt_trans_create(env, dev);
3483         if (IS_ERR(handle))
3484                 GOTO(unlock1, rc = PTR_ERR(handle));
3485
3486         rc = dt_declare_create(env, child, la, NULL, dof, handle);
3487         if (rc != 0)
3488                 GOTO(stop, rc);
3489
3490         rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID,
3491                                   LU_XATTR_CREATE, handle);
3492         if (rc != 0)
3493                 GOTO(stop, rc);
3494
3495         rc = dt_trans_start_local(env, dev, handle);
3496         if (rc != 0)
3497                 GOTO(stop, rc);
3498
3499         dt_read_lock(env, parent, 0);
3500         if (unlikely(lfsck_is_dead_obj(parent)))
3501                 GOTO(unlock2, rc = 0);
3502
3503         if (lfsck->li_bookmark_ram.lb_param & LPF_DELAY_CREATE_OSTOBJ) {
3504                 struct ost_id *oi = &info->lti_oi;
3505                 struct lu_fid *tfid = &info->lti_fid2;
3506                 struct lu_buf *lovea = &info->lti_big_buf;
3507                 struct lov_mds_md_v1 *lmm;
3508                 struct lov_ost_data_v1 *objs;
3509                 __u32 magic;
3510                 int count;
3511                 int idx2;
3512
3513                 rc = lfsck_layout_get_lovea(env, parent, lovea);
3514                 if (unlikely(rc == -ENODATA))
3515                         rc = 0;
3516                 if (rc <= 0)
3517                         GOTO(unlock2, rc);
3518
3519                 lmm = lovea->lb_buf;
3520                 magic = le32_to_cpu(lmm->lmm_magic);
3521                 if (magic == LOV_MAGIC_COMP_V1) {
3522                         struct lov_comp_md_v1 *lcm = buf->lb_buf;
3523                         struct lov_comp_md_entry_v1 *lcme;
3524                         __u16 count = le16_to_cpu(lcm->lcm_entry_count);
3525                         int i;
3526
3527                         for (i = 0; i < count; i++) {
3528                                 lcme = &lcm->lcm_entries[i];
3529                                 if (le32_to_cpu(lcme->lcme_id) == comp_id) {
3530                                         LASSERT(le32_to_cpu(lcme->lcme_flags) &
3531                                                 LCME_FL_INIT);
3532
3533                                         lmm = lovea->lb_buf +
3534                                                 le32_to_cpu(lcme->lcme_offset);
3535                                         magic = le32_to_cpu(lmm->lmm_magic);
3536                                         goto check;
3537                                 }
3538                         }
3539
3540                         /* Someone removed the component, do nothing. */
3541                         GOTO(unlock2, rc = 0);
3542                 }
3543
3544 check:
3545                 count = le16_to_cpu(lmm->lmm_stripe_count);
3546                 /* Someone changed the LOV EA, do nothing. */
3547                 if (count <= ea_off)
3548                         GOTO(unlock2, rc = 0);
3549
3550                 if (magic == LOV_MAGIC_V1) {
3551                         objs = &lmm->lmm_objects[ea_off];
3552                 } else {
3553                         LASSERT(magic == LOV_MAGIC_V3);
3554
3555                         objs = &((struct lov_mds_md_v3 *)lmm)->\
3556                                               &nbs