4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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.
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
23 * Copyright (c) 2014, Intel Corporation.
26 * lustre/lfsck/lfsck_layout.c
28 * Author: Fan, Yong <fan.yong@intel.com>
32 # define EXPORT_SYMTAB
34 #define DEBUG_SUBSYSTEM S_LFSCK
36 #include <linux/bitops.h>
37 #include <linux/rbtree.h>
39 #include <lustre/lustre_idl.h>
40 #include <lu_object.h>
41 #include <dt_object.h>
42 #include <lustre_fid.h>
43 #include <lustre_lib.h>
44 #include <lustre_net.h>
45 #include <lustre/lustre_user.h>
46 #include <md_object.h>
47 #include <obd_class.h>
49 #include "lfsck_internal.h"
51 #define LFSCK_LAYOUT_MAGIC_V1 0xB173AE14
52 #define LFSCK_LAYOUT_MAGIC_V2 0xB1734D76
54 #define LFSCK_LAYOUT_MAGIC LFSCK_LAYOUT_MAGIC_V2
56 struct lfsck_layout_seq {
57 struct list_head lls_list;
60 __u64 lls_lastid_known;
61 struct dt_object *lls_lastid_obj;
62 unsigned int lls_dirty:1;
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. */
77 struct lfsck_layout_slave_data {
78 /* list for lfsck_layout_seq */
79 struct list_head llsd_seq_list;
81 /* list for the masters involve layout verification. */
82 struct list_head llsd_master_list;
85 struct dt_object *llsd_rb_obj;
86 struct rb_root llsd_rb_root;
87 rwlock_t llsd_rb_lock;
88 unsigned int llsd_rbtree_valid:1;
91 struct lfsck_layout_slave_async_args {
92 struct obd_export *llsaa_exp;
93 struct lfsck_component *llsaa_com;
94 struct lfsck_layout_slave_target *llsaa_llst;
98 lfsck_layout_llst_put(struct lfsck_layout_slave_target *llst)
100 if (atomic_dec_and_test(&llst->llst_ref)) {
101 LASSERT(list_empty(&llst->llst_list));
108 lfsck_layout_llst_add(struct lfsck_layout_slave_data *llsd, __u32 index)
110 struct lfsck_layout_slave_target *llst;
111 struct lfsck_layout_slave_target *tmp;
118 INIT_LIST_HEAD(&llst->llst_list);
120 llst->llst_index = index;
121 atomic_set(&llst->llst_ref, 1);
123 spin_lock(&llsd->llsd_lock);
124 list_for_each_entry(tmp, &llsd->llsd_master_list, llst_list) {
125 if (tmp->llst_index == index) {
131 list_add_tail(&llst->llst_list, &llsd->llsd_master_list);
132 spin_unlock(&llsd->llsd_lock);
141 lfsck_layout_llst_del(struct lfsck_layout_slave_data *llsd,
142 struct lfsck_layout_slave_target *llst)
146 spin_lock(&llsd->llsd_lock);
147 if (!list_empty(&llst->llst_list)) {
148 list_del_init(&llst->llst_list);
151 spin_unlock(&llsd->llsd_lock);
154 lfsck_layout_llst_put(llst);
157 static inline struct lfsck_layout_slave_target *
158 lfsck_layout_llst_find_and_del(struct lfsck_layout_slave_data *llsd,
159 __u32 index, bool unlink)
161 struct lfsck_layout_slave_target *llst;
163 spin_lock(&llsd->llsd_lock);
164 list_for_each_entry(llst, &llsd->llsd_master_list, llst_list) {
165 if (llst->llst_index == index) {
167 list_del_init(&llst->llst_list);
169 atomic_inc(&llst->llst_ref);
170 spin_unlock(&llsd->llsd_lock);
175 spin_unlock(&llsd->llsd_lock);
180 static struct lfsck_layout_req *
181 lfsck_layout_assistant_req_init(struct lfsck_assistant_object *lso,
182 struct dt_object *child, __u32 ost_idx,
185 struct lfsck_layout_req *llr;
189 return ERR_PTR(-ENOMEM);
191 INIT_LIST_HEAD(&llr->llr_lar.lar_list);
192 llr->llr_lar.lar_parent = lfsck_assistant_object_get(lso);
193 llr->llr_child = child;
194 llr->llr_ost_idx = ost_idx;
195 llr->llr_lov_idx = lov_idx;
200 static void lfsck_layout_assistant_req_fini(const struct lu_env *env,
201 struct lfsck_assistant_req *lar)
203 struct lfsck_layout_req *llr =
204 container_of0(lar, struct lfsck_layout_req, llr_lar);
206 lfsck_object_put(env, llr->llr_child);
207 lfsck_assistant_object_put(env, lar->lar_parent);
212 lfsck_layout_assistant_sync_failures_interpret(const struct lu_env *env,
213 struct ptlrpc_request *req,
217 struct lfsck_async_interpret_args *laia = args;
218 struct lfsck_tgt_desc *ltd = laia->laia_ltd;
220 ltd->ltd_synced_failures = 1;
221 atomic_dec(laia->laia_count);
228 * Notify remote LFSCK instances about former failures.
230 * The local LFSCK instance has recorded which OSTs have ever failed to respond
231 * some LFSCK verification requests (maybe because of network issues or the OST
232 * itself trouble). During the respond gap, the OST may missed some OST-objects
233 * verification, then the OST cannot know whether related OST-objects have been
234 * referenced by related MDT-objects or not, then in the second-stage scanning,
235 * these OST-objects will be regarded as orphan, if the OST-object contains bad
236 * parent FID for back reference, then it will misguide the LFSCK to make wrong
237 * fixing for the fake orphan.
239 * To avoid above trouble, when layout LFSCK finishes the first-stage scanning,
240 * it will scan the bitmap for the ever failed OSTs, and notify them that they
241 * have ever missed some OST-object verification and should skip the handling
242 * for orphan OST-objects on all MDTs that are in the layout LFSCK.
244 * \param[in] env pointer to the thread context
245 * \param[in] com pointer to the lfsck component
246 * \param[in] lr pointer to the lfsck request
248 static void lfsck_layout_assistant_sync_failures(const struct lu_env *env,
249 struct lfsck_component *com,
250 struct lfsck_request *lr)
252 struct lfsck_async_interpret_args *laia =
253 &lfsck_env_info(env)->lti_laia2;
254 struct lfsck_assistant_data *lad = com->lc_data;
255 struct lfsck_layout *lo = com->lc_file_ram;
256 struct lfsck_instance *lfsck = com->lc_lfsck;
257 struct lfsck_tgt_descs *ltds = &lfsck->li_ost_descs;
258 struct lfsck_tgt_desc *ltd;
259 struct ptlrpc_request_set *set;
265 if (!lad->lad_incomplete || lo->ll_flags & LF_INCOMPLETE)
268 /* If the MDT has ever failed to verfiy some OST-objects,
269 * then sync failures with them firstly. */
270 lr->lr_flags2 = lo->ll_flags | LF_INCOMPLETE;
272 atomic_set(&count, 0);
273 memset(laia, 0, sizeof(*laia));
274 laia->laia_count = &count;
275 set = ptlrpc_prep_set();
277 GOTO(out, rc = -ENOMEM);
279 down_read(<ds->ltd_rw_sem);
280 cfs_foreach_bit(lad->lad_bitmap, idx) {
281 ltd = LTD_TGT(ltds, idx);
282 LASSERT(ltd != NULL);
284 laia->laia_ltd = ltd;
285 rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
286 lfsck_layout_assistant_sync_failures_interpret,
289 CDEBUG(D_LFSCK, "%s: LFSCK assistant fail to "
290 "notify target %x for %s phase1 done: "
291 "rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
292 ltd->ltd_index, lad->lad_name, rc);
299 up_read(<ds->ltd_rw_sem);
301 if (rc == 0 && atomic_read(&count) > 0)
302 rc = ptlrpc_set_wait(set);
304 ptlrpc_set_destroy(set);
306 if (rc == 0 && atomic_read(&count) > 0)
313 /* If failed to sync failures with the OSTs, then have to
314 * mark the whole LFSCK as LF_INCOMPLETE to skip the whole
315 * subsequent orphan OST-object handling. */
316 lo->ll_flags |= LF_INCOMPLETE;
318 lr->lr_flags2 = lo->ll_flags;
321 static int lfsck_layout_get_lovea(const struct lu_env *env,
322 struct dt_object *obj, struct lu_buf *buf)
327 rc = dt_xattr_get(env, obj, buf, XATTR_NAME_LOV);
329 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LOV);
333 lu_buf_realloc(buf, rc);
334 if (buf->lb_buf == NULL)
346 if (unlikely(buf->lb_buf == NULL)) {
347 lu_buf_alloc(buf, rc);
348 if (buf->lb_buf == NULL)
357 static int lfsck_layout_verify_header(struct lov_mds_md_v1 *lmm)
362 magic = le32_to_cpu(lmm->lmm_magic);
363 /* If magic crashed, keep it there. Sometime later, during OST-object
364 * orphan handling, if some OST-object(s) back-point to it, it can be
365 * verified and repaired. */
366 if (magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3) {
370 lmm_oi_le_to_cpu(&oi, &lmm->lmm_oi);
371 if ((magic & LOV_MAGIC_MASK) == LOV_MAGIC_MAGIC)
376 CDEBUG(D_LFSCK, "%s LOV EA magic %u on "DOSTID"\n",
377 rc == -EINVAL ? "Unknown" : "Unsupported",
383 pattern = le32_to_cpu(lmm->lmm_pattern);
384 /* XXX: currently, we only support LOV_PATTERN_RAID0. */
385 if (lov_pattern(pattern) != LOV_PATTERN_RAID0) {
388 lmm_oi_le_to_cpu(&oi, &lmm->lmm_oi);
389 CDEBUG(D_LFSCK, "Unsupported LOV EA pattern %u on "DOSTID"\n",
390 pattern, POSTID(&oi));
398 #define LFSCK_RBTREE_BITMAP_SIZE PAGE_CACHE_SIZE
399 #define LFSCK_RBTREE_BITMAP_WIDTH (LFSCK_RBTREE_BITMAP_SIZE << 3)
400 #define LFSCK_RBTREE_BITMAP_MASK (LFSCK_RBTREE_BITMAP_WIDTH - 1)
402 struct lfsck_rbtree_node {
403 struct rb_node lrn_node;
406 atomic_t lrn_known_count;
407 atomic_t lrn_accessed_count;
408 void *lrn_known_bitmap;
409 void *lrn_accessed_bitmap;
412 static inline int lfsck_rbtree_cmp(struct lfsck_rbtree_node *lrn,
413 __u64 seq, __u32 oid)
415 if (seq < lrn->lrn_seq)
418 if (seq > lrn->lrn_seq)
421 if (oid < lrn->lrn_first_oid)
424 if (oid - lrn->lrn_first_oid >= LFSCK_RBTREE_BITMAP_WIDTH)
430 /* The caller should hold llsd->llsd_rb_lock. */
431 static struct lfsck_rbtree_node *
432 lfsck_rbtree_search(struct lfsck_layout_slave_data *llsd,
433 const struct lu_fid *fid, bool *exact)
435 struct rb_node *node = llsd->llsd_rb_root.rb_node;
436 struct rb_node *prev = NULL;
437 struct lfsck_rbtree_node *lrn = NULL;
443 while (node != NULL) {
445 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
446 rc = lfsck_rbtree_cmp(lrn, fid_seq(fid), fid_oid(fid));
448 node = node->rb_left;
450 node = node->rb_right;
458 /* If there is no exactly matched one, then to the next valid one. */
461 /* The rbtree is empty. */
468 node = rb_next(prev);
470 /* The end of the rbtree. */
474 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
479 static struct lfsck_rbtree_node *lfsck_rbtree_new(const struct lu_env *env,
480 const struct lu_fid *fid)
482 struct lfsck_rbtree_node *lrn;
486 return ERR_PTR(-ENOMEM);
488 OBD_ALLOC(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
489 if (lrn->lrn_known_bitmap == NULL) {
492 return ERR_PTR(-ENOMEM);
495 OBD_ALLOC(lrn->lrn_accessed_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
496 if (lrn->lrn_accessed_bitmap == NULL) {
497 OBD_FREE(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
500 return ERR_PTR(-ENOMEM);
503 RB_CLEAR_NODE(&lrn->lrn_node);
504 lrn->lrn_seq = fid_seq(fid);
505 lrn->lrn_first_oid = fid_oid(fid) & ~LFSCK_RBTREE_BITMAP_MASK;
506 atomic_set(&lrn->lrn_known_count, 0);
507 atomic_set(&lrn->lrn_accessed_count, 0);
512 static void lfsck_rbtree_free(struct lfsck_rbtree_node *lrn)
514 OBD_FREE(lrn->lrn_accessed_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
515 OBD_FREE(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
519 /* The caller should hold lock. */
520 static struct lfsck_rbtree_node *
521 lfsck_rbtree_insert(struct lfsck_layout_slave_data *llsd,
522 struct lfsck_rbtree_node *lrn)
524 struct rb_node **pos = &llsd->llsd_rb_root.rb_node;
525 struct rb_node *parent = NULL;
526 struct lfsck_rbtree_node *tmp;
529 while (*pos != NULL) {
531 tmp = rb_entry(parent, struct lfsck_rbtree_node, lrn_node);
532 rc = lfsck_rbtree_cmp(tmp, lrn->lrn_seq, lrn->lrn_first_oid);
534 pos = &(*pos)->rb_left;
536 pos = &(*pos)->rb_right;
541 rb_link_node(&lrn->lrn_node, parent, pos);
542 rb_insert_color(&lrn->lrn_node, &llsd->llsd_rb_root);
547 extern const struct dt_index_operations lfsck_orphan_index_ops;
549 static int lfsck_rbtree_setup(const struct lu_env *env,
550 struct lfsck_component *com)
552 struct lu_fid *fid = &lfsck_env_info(env)->lti_fid;
553 struct lfsck_instance *lfsck = com->lc_lfsck;
554 struct dt_device *dev = lfsck->li_bottom;
555 struct lfsck_layout_slave_data *llsd = com->lc_data;
556 struct dt_object *obj;
558 fid->f_seq = FID_SEQ_LAYOUT_RBTREE;
559 fid->f_oid = lfsck_dev_idx(lfsck);
561 obj = dt_locate(env, dev, fid);
563 RETURN(PTR_ERR(obj));
565 /* Generate an in-RAM object to stand for the layout rbtree.
566 * Scanning the layout rbtree will be via the iteration over
567 * the object. In the future, the rbtree may be written onto
568 * disk with the object.
570 * Mark the object to be as exist. */
571 obj->do_lu.lo_header->loh_attr |= LOHA_EXISTS;
572 obj->do_index_ops = &lfsck_orphan_index_ops;
573 llsd->llsd_rb_obj = obj;
574 llsd->llsd_rbtree_valid = 1;
575 dev->dd_record_fid_accessed = 1;
577 CDEBUG(D_LFSCK, "%s: layout LFSCK init OST-objects accessing bitmap\n",
578 lfsck_lfsck2name(lfsck));
583 static void lfsck_rbtree_cleanup(const struct lu_env *env,
584 struct lfsck_component *com)
586 struct lfsck_instance *lfsck = com->lc_lfsck;
587 struct lfsck_layout_slave_data *llsd = com->lc_data;
588 struct rb_node *node = rb_first(&llsd->llsd_rb_root);
589 struct rb_node *next;
590 struct lfsck_rbtree_node *lrn;
592 lfsck->li_bottom->dd_record_fid_accessed = 0;
593 /* Invalid the rbtree, then no others will use it. */
594 write_lock(&llsd->llsd_rb_lock);
595 llsd->llsd_rbtree_valid = 0;
596 write_unlock(&llsd->llsd_rb_lock);
598 while (node != NULL) {
599 next = rb_next(node);
600 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
601 rb_erase(node, &llsd->llsd_rb_root);
602 lfsck_rbtree_free(lrn);
606 if (llsd->llsd_rb_obj != NULL) {
607 lfsck_object_put(env, llsd->llsd_rb_obj);
608 llsd->llsd_rb_obj = NULL;
611 CDEBUG(D_LFSCK, "%s: layout LFSCK fini OST-objects accessing bitmap\n",
612 lfsck_lfsck2name(lfsck));
615 static void lfsck_rbtree_update_bitmap(const struct lu_env *env,
616 struct lfsck_component *com,
617 const struct lu_fid *fid,
620 struct lfsck_layout_slave_data *llsd = com->lc_data;
621 struct lfsck_rbtree_node *lrn;
627 if (unlikely(!fid_is_sane(fid) || fid_is_last_id(fid)))
630 if (!fid_is_idif(fid) && !fid_is_norm(fid))
633 read_lock(&llsd->llsd_rb_lock);
634 if (!llsd->llsd_rbtree_valid)
635 GOTO(unlock, rc = 0);
637 lrn = lfsck_rbtree_search(llsd, fid, NULL);
639 struct lfsck_rbtree_node *tmp;
643 read_unlock(&llsd->llsd_rb_lock);
644 tmp = lfsck_rbtree_new(env, fid);
646 GOTO(out, rc = PTR_ERR(tmp));
649 write_lock(&llsd->llsd_rb_lock);
650 if (!llsd->llsd_rbtree_valid) {
651 lfsck_rbtree_free(tmp);
652 GOTO(unlock, rc = 0);
655 lrn = lfsck_rbtree_insert(llsd, tmp);
657 lfsck_rbtree_free(tmp);
660 idx = fid_oid(fid) & LFSCK_RBTREE_BITMAP_MASK;
661 /* Any accessed object must be a known object. */
662 if (!test_and_set_bit(idx, lrn->lrn_known_bitmap))
663 atomic_inc(&lrn->lrn_known_count);
664 if (accessed && !test_and_set_bit(idx, lrn->lrn_accessed_bitmap))
665 atomic_inc(&lrn->lrn_accessed_count);
667 GOTO(unlock, rc = 0);
671 write_unlock(&llsd->llsd_rb_lock);
673 read_unlock(&llsd->llsd_rb_lock);
675 if (rc != 0 && accessed) {
676 struct lfsck_layout *lo = com->lc_file_ram;
678 CDEBUG(D_LFSCK, "%s: fail to update OST-objects accessing "
679 "bitmap, and will cause incorrect LFSCK OST-object "
680 "handling, so disable it to cancel orphan handling "
681 "for related device. rc = %d\n",
682 lfsck_lfsck2name(com->lc_lfsck), rc);
684 lo->ll_flags |= LF_INCOMPLETE;
685 lfsck_rbtree_cleanup(env, com);
689 static void lfsck_layout_le_to_cpu(struct lfsck_layout *des,
690 const struct lfsck_layout *src)
694 des->ll_magic = le32_to_cpu(src->ll_magic);
695 des->ll_status = le32_to_cpu(src->ll_status);
696 des->ll_flags = le32_to_cpu(src->ll_flags);
697 des->ll_success_count = le32_to_cpu(src->ll_success_count);
698 des->ll_run_time_phase1 = le32_to_cpu(src->ll_run_time_phase1);
699 des->ll_run_time_phase2 = le32_to_cpu(src->ll_run_time_phase2);
700 des->ll_time_last_complete = le64_to_cpu(src->ll_time_last_complete);
701 des->ll_time_latest_start = le64_to_cpu(src->ll_time_latest_start);
702 des->ll_time_last_checkpoint =
703 le64_to_cpu(src->ll_time_last_checkpoint);
704 des->ll_pos_latest_start = le64_to_cpu(src->ll_pos_latest_start);
705 des->ll_pos_last_checkpoint = le64_to_cpu(src->ll_pos_last_checkpoint);
706 des->ll_pos_first_inconsistent =
707 le64_to_cpu(src->ll_pos_first_inconsistent);
708 des->ll_objs_checked_phase1 = le64_to_cpu(src->ll_objs_checked_phase1);
709 des->ll_objs_failed_phase1 = le64_to_cpu(src->ll_objs_failed_phase1);
710 des->ll_objs_checked_phase2 = le64_to_cpu(src->ll_objs_checked_phase2);
711 des->ll_objs_failed_phase2 = le64_to_cpu(src->ll_objs_failed_phase2);
712 for (i = 0; i < LLIT_MAX; i++)
713 des->ll_objs_repaired[i] =
714 le64_to_cpu(src->ll_objs_repaired[i]);
715 des->ll_objs_skipped = le64_to_cpu(src->ll_objs_skipped);
716 des->ll_bitmap_size = le32_to_cpu(src->ll_bitmap_size);
719 static void lfsck_layout_cpu_to_le(struct lfsck_layout *des,
720 const struct lfsck_layout *src)
724 des->ll_magic = cpu_to_le32(src->ll_magic);
725 des->ll_status = cpu_to_le32(src->ll_status);
726 des->ll_flags = cpu_to_le32(src->ll_flags);
727 des->ll_success_count = cpu_to_le32(src->ll_success_count);
728 des->ll_run_time_phase1 = cpu_to_le32(src->ll_run_time_phase1);
729 des->ll_run_time_phase2 = cpu_to_le32(src->ll_run_time_phase2);
730 des->ll_time_last_complete = cpu_to_le64(src->ll_time_last_complete);
731 des->ll_time_latest_start = cpu_to_le64(src->ll_time_latest_start);
732 des->ll_time_last_checkpoint =
733 cpu_to_le64(src->ll_time_last_checkpoint);
734 des->ll_pos_latest_start = cpu_to_le64(src->ll_pos_latest_start);
735 des->ll_pos_last_checkpoint = cpu_to_le64(src->ll_pos_last_checkpoint);
736 des->ll_pos_first_inconsistent =
737 cpu_to_le64(src->ll_pos_first_inconsistent);
738 des->ll_objs_checked_phase1 = cpu_to_le64(src->ll_objs_checked_phase1);
739 des->ll_objs_failed_phase1 = cpu_to_le64(src->ll_objs_failed_phase1);
740 des->ll_objs_checked_phase2 = cpu_to_le64(src->ll_objs_checked_phase2);
741 des->ll_objs_failed_phase2 = cpu_to_le64(src->ll_objs_failed_phase2);
742 for (i = 0; i < LLIT_MAX; i++)
743 des->ll_objs_repaired[i] =
744 cpu_to_le64(src->ll_objs_repaired[i]);
745 des->ll_objs_skipped = cpu_to_le64(src->ll_objs_skipped);
746 des->ll_bitmap_size = cpu_to_le32(src->ll_bitmap_size);
750 * Load the OST bitmap from the lfsck_layout trace file.
752 * \param[in] env pointer to the thread context
753 * \param[in] com pointer to the lfsck component
755 * \retval 0 for success
756 * \retval negative error number on failure or data corruption
758 static int lfsck_layout_load_bitmap(const struct lu_env *env,
759 struct lfsck_component *com)
761 struct dt_object *obj = com->lc_obj;
762 struct lfsck_assistant_data *lad = com->lc_data;
763 struct lfsck_layout *lo = com->lc_file_ram;
764 cfs_bitmap_t *bitmap = lad->lad_bitmap;
765 loff_t pos = com->lc_file_size;
771 if (com->lc_lfsck->li_ost_descs.ltd_tgts_bitmap->size >
773 nbits = com->lc_lfsck->li_ost_descs.ltd_tgts_bitmap->size;
775 nbits = lo->ll_bitmap_size;
777 if (unlikely(nbits < BITS_PER_LONG))
778 nbits = BITS_PER_LONG;
780 if (nbits > bitmap->size) {
781 __u32 new_bits = bitmap->size;
782 cfs_bitmap_t *new_bitmap;
784 while (new_bits < nbits)
787 new_bitmap = CFS_ALLOCATE_BITMAP(new_bits);
788 if (new_bitmap == NULL)
791 lad->lad_bitmap = new_bitmap;
792 CFS_FREE_BITMAP(bitmap);
796 if (lo->ll_bitmap_size == 0) {
797 lad->lad_incomplete = 0;
798 CFS_RESET_BITMAP(bitmap);
803 size = (lo->ll_bitmap_size + 7) >> 3;
804 rc = dt_read(env, obj, lfsck_buf_get(env, bitmap->data, size), &pos);
806 RETURN(rc >= 0 ? -EINVAL : rc);
808 if (cfs_bitmap_check_empty(bitmap))
809 lad->lad_incomplete = 0;
811 lad->lad_incomplete = 1;
817 * Load the layout LFSCK trace file from disk.
819 * The layout LFSCK trace file records the layout LFSCK status information
820 * and other statistics, such as how many objects have been scanned, and how
821 * many objects have been repaired, and etc. It also contains the bitmap for
822 * failed OSTs during the layout LFSCK. All these information will be loaded
823 * from disk to RAM when the layout LFSCK component setup.
825 * \param[in] env pointer to the thread context
826 * \param[in] com pointer to the lfsck component
828 * \retval positive number for file data corruption, the caller
829 * should reset the layout LFSCK trace file
830 * \retval 0 for success
831 * \retval negative error number on failure
833 static int lfsck_layout_load(const struct lu_env *env,
834 struct lfsck_component *com)
836 struct lfsck_layout *lo = com->lc_file_ram;
837 ssize_t size = com->lc_file_size;
841 rc = dt_read(env, com->lc_obj,
842 lfsck_buf_get(env, com->lc_file_disk, size), &pos);
846 CDEBUG(D_LFSCK, "%s: failed to load lfsck_layout: rc = %d\n",
847 lfsck_lfsck2name(com->lc_lfsck), rc);
849 } else if (rc != size) {
850 CDEBUG(D_LFSCK, "%s: lfsck_layout size %u != %u; reset it\n",
851 lfsck_lfsck2name(com->lc_lfsck), rc, (unsigned int)size);
855 lfsck_layout_le_to_cpu(lo, com->lc_file_disk);
856 if (lo->ll_magic != LFSCK_LAYOUT_MAGIC) {
857 CDEBUG(D_LFSCK, "%s: invalid lfsck_layout magic %#x != %#x, "
858 "to be reset\n", lfsck_lfsck2name(com->lc_lfsck),
859 lo->ll_magic, LFSCK_LAYOUT_MAGIC);
867 * Store the layout LFSCK trace file on disk.
869 * The layout LFSCK trace file records the layout LFSCK status information
870 * and other statistics, such as how many objects have been scanned, and how
871 * many objects have been repaired, and etc. It also contains the bitmap for
872 * failed OSTs during the layout LFSCK. All these information will be synced
873 * from RAM to disk periodically.
875 * \param[in] env pointer to the thread context
876 * \param[in] com pointer to the lfsck component
878 * \retval 0 for success
879 * \retval negative error number on failure
881 static int lfsck_layout_store(const struct lu_env *env,
882 struct lfsck_component *com)
884 struct dt_object *obj = com->lc_obj;
885 struct lfsck_instance *lfsck = com->lc_lfsck;
886 struct lfsck_layout *lo_ram = com->lc_file_ram;
887 struct lfsck_layout *lo = com->lc_file_disk;
889 struct dt_device *dev = lfsck_obj2dev(obj);
890 cfs_bitmap_t *bitmap = NULL;
892 ssize_t size = com->lc_file_size;
897 if (lfsck->li_master) {
898 struct lfsck_assistant_data *lad = com->lc_data;
900 bitmap = lad->lad_bitmap;
901 nbits = bitmap->size;
904 LASSERTF((nbits & 7) == 0, "Invalid nbits %u\n", nbits);
907 lo_ram->ll_bitmap_size = nbits;
908 lfsck_layout_cpu_to_le(lo, lo_ram);
909 th = dt_trans_create(env, dev);
911 GOTO(log, rc = PTR_ERR(th));
913 rc = dt_declare_record_write(env, obj, lfsck_buf_get(env, lo, size),
918 if (bitmap != NULL) {
919 rc = dt_declare_record_write(env, obj,
920 lfsck_buf_get(env, bitmap->data, nbits >> 3),
926 rc = dt_trans_start_local(env, dev, th);
931 rc = dt_record_write(env, obj, lfsck_buf_get(env, lo, size), &pos, th);
935 if (bitmap != NULL) {
937 rc = dt_record_write(env, obj,
938 lfsck_buf_get(env, bitmap->data, nbits >> 3),
945 dt_trans_stop(env, dev, th);
949 CDEBUG(D_LFSCK, "%s: fail to store lfsck_layout: rc = %d\n",
950 lfsck_lfsck2name(lfsck), rc);
955 static int lfsck_layout_init(const struct lu_env *env,
956 struct lfsck_component *com)
958 struct lfsck_layout *lo = com->lc_file_ram;
961 memset(lo, 0, com->lc_file_size);
962 lo->ll_magic = LFSCK_LAYOUT_MAGIC;
963 lo->ll_status = LS_INIT;
964 down_write(&com->lc_sem);
965 rc = lfsck_layout_store(env, com);
966 up_write(&com->lc_sem);
971 static int fid_is_for_ostobj(const struct lu_env *env,
972 struct lfsck_instance *lfsck,
973 struct dt_object *obj, const struct lu_fid *fid)
975 struct seq_server_site *ss = lfsck_dev_site(lfsck);
976 struct lu_seq_range *range = &lfsck_env_info(env)->lti_range;
977 struct lustre_mdt_attrs *lma;
980 fld_range_set_any(range);
981 rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), range);
983 if (fld_range_is_ost(range))
989 lma = &lfsck_env_info(env)->lti_lma;
990 rc = dt_xattr_get(env, obj, lfsck_buf_get(env, lma, sizeof(*lma)),
992 if (rc == sizeof(*lma)) {
993 lustre_lma_swab(lma);
995 return lma->lma_compat & LMAC_FID_ON_OST ? 1 : 0;
998 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_FID);
1003 static struct lfsck_layout_seq *
1004 lfsck_layout_seq_lookup(struct lfsck_layout_slave_data *llsd, __u64 seq)
1006 struct lfsck_layout_seq *lls;
1008 list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
1009 if (lls->lls_seq == seq)
1012 if (lls->lls_seq > seq)
1020 lfsck_layout_seq_insert(struct lfsck_layout_slave_data *llsd,
1021 struct lfsck_layout_seq *lls)
1023 struct lfsck_layout_seq *tmp;
1024 struct list_head *pos = &llsd->llsd_seq_list;
1026 list_for_each_entry(tmp, &llsd->llsd_seq_list, lls_list) {
1027 if (lls->lls_seq < tmp->lls_seq) {
1028 pos = &tmp->lls_list;
1032 list_add_tail(&lls->lls_list, pos);
1036 lfsck_layout_lastid_create(const struct lu_env *env,
1037 struct lfsck_instance *lfsck,
1038 struct dt_object *obj)
1040 struct lfsck_thread_info *info = lfsck_env_info(env);
1041 struct lu_attr *la = &info->lti_la;
1042 struct dt_object_format *dof = &info->lti_dof;
1043 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
1044 struct dt_device *dt = lfsck_obj2dev(obj);
1051 if (bk->lb_param & LPF_DRYRUN)
1054 memset(la, 0, sizeof(*la));
1055 la->la_mode = S_IFREG | S_IRUGO | S_IWUSR;
1056 la->la_valid = LA_MODE | LA_UID | LA_GID;
1057 memset(dof, 0, sizeof(*dof));
1058 dof->dof_type = dt_mode_to_dft(S_IFREG);
1060 th = dt_trans_create(env, dt);
1062 GOTO(log, rc = PTR_ERR(th));
1064 rc = dt_declare_create(env, obj, la, NULL, dof, th);
1068 rc = dt_declare_record_write(env, obj,
1069 lfsck_buf_get(env, &lastid,
1075 rc = dt_trans_start_local(env, dt, th);
1079 dt_write_lock(env, obj, 0);
1080 if (likely(dt_object_exists(obj) == 0)) {
1081 rc = dt_create(env, obj, la, NULL, dof, th);
1083 rc = dt_record_write(env, obj,
1084 lfsck_buf_get(env, &lastid, sizeof(lastid)),
1087 dt_write_unlock(env, obj);
1092 dt_trans_stop(env, dt, th);
1095 CDEBUG(D_LFSCK, "%s: layout LFSCK will create LAST_ID for <seq> "
1097 lfsck_lfsck2name(lfsck), fid_seq(lfsck_dto2fid(obj)), rc);
1103 lfsck_layout_lastid_reload(const struct lu_env *env,
1104 struct lfsck_component *com,
1105 struct lfsck_layout_seq *lls)
1111 dt_read_lock(env, lls->lls_lastid_obj, 0);
1112 rc = dt_record_read(env, lls->lls_lastid_obj,
1113 lfsck_buf_get(env, &lastid, sizeof(lastid)), &pos);
1114 dt_read_unlock(env, lls->lls_lastid_obj);
1115 if (unlikely(rc != 0))
1118 lastid = le64_to_cpu(lastid);
1119 if (lastid < lls->lls_lastid_known) {
1120 struct lfsck_instance *lfsck = com->lc_lfsck;
1121 struct lfsck_layout *lo = com->lc_file_ram;
1123 lls->lls_lastid = lls->lls_lastid_known;
1125 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
1126 LASSERT(lfsck->li_out_notify != NULL);
1128 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1129 LE_LASTID_REBUILDING);
1130 lo->ll_flags |= LF_CRASHED_LASTID;
1132 CDEBUG(D_LFSCK, "%s: layout LFSCK finds crashed "
1133 "LAST_ID file (1) for the sequence "LPX64
1134 ", old value "LPU64", known value "LPU64"\n",
1135 lfsck_lfsck2name(lfsck), lls->lls_seq,
1136 lastid, lls->lls_lastid);
1138 } else if (lastid >= lls->lls_lastid) {
1139 lls->lls_lastid = lastid;
1147 lfsck_layout_lastid_store(const struct lu_env *env,
1148 struct lfsck_component *com)
1150 struct lfsck_instance *lfsck = com->lc_lfsck;
1151 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
1152 struct dt_device *dt = lfsck->li_bottom;
1153 struct lfsck_layout_slave_data *llsd = com->lc_data;
1154 struct lfsck_layout_seq *lls;
1160 list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
1163 if (!lls->lls_dirty)
1166 CDEBUG(D_LFSCK, "%s: layout LFSCK will sync the LAST_ID for "
1167 "<seq> "LPX64" as <oid> "LPU64"\n",
1168 lfsck_lfsck2name(lfsck), lls->lls_seq, lls->lls_lastid);
1170 if (bk->lb_param & LPF_DRYRUN) {
1175 th = dt_trans_create(env, dt);
1178 CDEBUG(D_LFSCK, "%s: layout LFSCK failed to store "
1179 "the LAST_ID for <seq> "LPX64"(1): rc = %d\n",
1180 lfsck_lfsck2name(com->lc_lfsck),
1185 lastid = cpu_to_le64(lls->lls_lastid);
1186 rc = dt_declare_record_write(env, lls->lls_lastid_obj,
1187 lfsck_buf_get(env, &lastid,
1193 rc = dt_trans_start_local(env, dt, th);
1197 dt_write_lock(env, lls->lls_lastid_obj, 0);
1198 rc = dt_record_write(env, lls->lls_lastid_obj,
1199 lfsck_buf_get(env, &lastid,
1200 sizeof(lastid)), &pos, th);
1201 dt_write_unlock(env, lls->lls_lastid_obj);
1206 dt_trans_stop(env, dt, th);
1209 CDEBUG(D_LFSCK, "%s: layout LFSCK failed to store "
1210 "the LAST_ID for <seq> "LPX64"(2): rc = %d\n",
1211 lfsck_lfsck2name(com->lc_lfsck),
1220 lfsck_layout_lastid_load(const struct lu_env *env,
1221 struct lfsck_component *com,
1222 struct lfsck_layout_seq *lls)
1224 struct lfsck_instance *lfsck = com->lc_lfsck;
1225 struct lfsck_layout *lo = com->lc_file_ram;
1226 struct lu_fid *fid = &lfsck_env_info(env)->lti_fid;
1227 struct dt_object *obj;
1232 lu_last_id_fid(fid, lls->lls_seq, lfsck_dev_idx(lfsck));
1233 obj = dt_locate(env, lfsck->li_bottom, fid);
1235 RETURN(PTR_ERR(obj));
1237 /* LAST_ID crashed, to be rebuilt */
1238 if (dt_object_exists(obj) == 0) {
1239 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
1240 LASSERT(lfsck->li_out_notify != NULL);
1242 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1243 LE_LASTID_REBUILDING);
1244 lo->ll_flags |= LF_CRASHED_LASTID;
1246 CDEBUG(D_LFSCK, "%s: layout LFSCK cannot find the "
1247 "LAST_ID file for sequence "LPX64"\n",
1248 lfsck_lfsck2name(lfsck), lls->lls_seq);
1250 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY4) &&
1252 struct l_wait_info lwi = LWI_TIMEOUT(
1253 cfs_time_seconds(cfs_fail_val),
1256 /* Some others may changed the cfs_fail_val
1257 * as zero after above check, re-check it for
1258 * sure to avoid falling into wait for ever. */
1259 if (likely(lwi.lwi_timeout > 0)) {
1260 struct ptlrpc_thread *thread =
1263 up_write(&com->lc_sem);
1264 l_wait_event(thread->t_ctl_waitq,
1265 !thread_is_running(thread),
1267 down_write(&com->lc_sem);
1272 rc = lfsck_layout_lastid_create(env, lfsck, obj);
1274 dt_read_lock(env, obj, 0);
1275 rc = dt_read(env, obj,
1276 lfsck_buf_get(env, &lls->lls_lastid, sizeof(__u64)),
1278 dt_read_unlock(env, obj);
1279 if (rc != 0 && rc != sizeof(__u64))
1280 GOTO(out, rc = (rc > 0 ? -EFAULT : rc));
1282 if (rc == 0 && !(lo->ll_flags & LF_CRASHED_LASTID)) {
1283 LASSERT(lfsck->li_out_notify != NULL);
1285 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1286 LE_LASTID_REBUILDING);
1287 lo->ll_flags |= LF_CRASHED_LASTID;
1289 CDEBUG(D_LFSCK, "%s: layout LFSCK finds invalid "
1290 "LAST_ID file for the sequence "LPX64
1292 lfsck_lfsck2name(lfsck), lls->lls_seq, rc);
1295 lls->lls_lastid = le64_to_cpu(lls->lls_lastid);
1303 lfsck_object_put(env, obj);
1305 lls->lls_lastid_obj = obj;
1310 static void lfsck_layout_record_failure(const struct lu_env *env,
1311 struct lfsck_instance *lfsck,
1312 struct lfsck_layout *lo)
1316 lo->ll_objs_failed_phase1++;
1317 cookie = lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
1319 if (lo->ll_pos_first_inconsistent == 0 ||
1320 lo->ll_pos_first_inconsistent < cookie) {
1321 lo->ll_pos_first_inconsistent = cookie;
1323 CDEBUG(D_LFSCK, "%s: layout LFSCK hit first non-repaired "
1324 "inconsistency at the pos ["LPU64"]\n",
1325 lfsck_lfsck2name(lfsck),
1326 lo->ll_pos_first_inconsistent);
1330 static int lfsck_layout_double_scan_result(const struct lu_env *env,
1331 struct lfsck_component *com,
1334 struct lfsck_instance *lfsck = com->lc_lfsck;
1335 struct lfsck_layout *lo = com->lc_file_ram;
1337 down_write(&com->lc_sem);
1338 lo->ll_run_time_phase2 += cfs_duration_sec(cfs_time_current() +
1339 HALF_SEC - com->lc_time_last_checkpoint);
1340 lo->ll_time_last_checkpoint = cfs_time_current_sec();
1341 lo->ll_objs_checked_phase2 += com->lc_new_checked;
1344 if (lo->ll_flags & LF_INCOMPLETE) {
1345 lo->ll_status = LS_PARTIAL;
1347 if (lfsck->li_master) {
1348 struct lfsck_assistant_data *lad = com->lc_data;
1350 if (lad->lad_incomplete)
1351 lo->ll_status = LS_PARTIAL;
1353 lo->ll_status = LS_COMPLETED;
1355 lo->ll_status = LS_COMPLETED;
1358 if (!(lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN))
1359 lo->ll_flags &= ~(LF_SCANNED_ONCE | LF_INCONSISTENT);
1360 lo->ll_time_last_complete = lo->ll_time_last_checkpoint;
1361 lo->ll_success_count++;
1362 } else if (rc == 0) {
1363 if (lfsck->li_status != 0)
1364 lo->ll_status = lfsck->li_status;
1366 lo->ll_status = LS_STOPPED;
1368 lo->ll_status = LS_FAILED;
1371 rc = lfsck_layout_store(env, com);
1372 up_write(&com->lc_sem);
1377 static int lfsck_layout_trans_stop(const struct lu_env *env,
1378 struct dt_device *dev,
1379 struct thandle *handle, int result)
1383 handle->th_result = result;
1384 rc = dt_trans_stop(env, dev, handle);
1394 * Get the system default stripe size.
1396 * \param[in] env pointer to the thread context
1397 * \param[in] lfsck pointer to the lfsck instance
1398 * \param[out] size pointer to the default stripe size
1400 * \retval 0 for success
1401 * \retval negative error number on failure
1403 static int lfsck_layout_get_def_stripesize(const struct lu_env *env,
1404 struct lfsck_instance *lfsck,
1407 struct lov_user_md *lum = &lfsck_env_info(env)->lti_lum;
1408 struct dt_object *root;
1411 root = dt_locate(env, lfsck->li_next, &lfsck->li_local_root_fid);
1413 return PTR_ERR(root);
1415 /* Get the default stripe size via xattr_get on the backend root. */
1416 rc = dt_xattr_get(env, root, lfsck_buf_get(env, lum, sizeof(*lum)),
1419 /* The lum->lmm_stripe_size is LE mode. The *size also
1420 * should be LE mode. So it is unnecessary to convert. */
1421 *size = lum->lmm_stripe_size;
1423 } else if (unlikely(rc == 0)) {
1427 lfsck_object_put(env, root);
1433 * \retval +1: repaired
1434 * \retval 0: did nothing
1435 * \retval -ve: on error
1437 static int lfsck_layout_refill_lovea(const struct lu_env *env,
1438 struct thandle *handle,
1439 struct dt_object *parent,
1440 struct lu_fid *cfid,
1442 struct lov_ost_data_v1 *slot,
1443 int fl, __u32 ost_idx)
1445 struct ost_id *oi = &lfsck_env_info(env)->lti_oi;
1446 struct lov_mds_md_v1 *lmm = buf->lb_buf;
1447 struct lu_buf ea_buf;
1453 magic = le32_to_cpu(lmm->lmm_magic);
1454 count = le16_to_cpu(lmm->lmm_stripe_count);
1456 fid_to_ostid(cfid, oi);
1457 ostid_cpu_to_le(oi, &slot->l_ost_oi);
1458 slot->l_ost_gen = cpu_to_le32(0);
1459 slot->l_ost_idx = cpu_to_le32(ost_idx);
1461 if (le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_HOLE) {
1462 struct lov_ost_data_v1 *objs;
1465 if (magic == LOV_MAGIC_V1)
1466 objs = &lmm->lmm_objects[0];
1468 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1469 for (i = 0; i < count; i++, objs++) {
1470 if (objs != slot && lovea_slot_is_dummy(objs))
1474 /* If the @slot is the last dummy slot to be refilled,
1475 * then drop LOV_PATTERN_F_HOLE from lmm::lmm_pattern. */
1477 lmm->lmm_pattern &= ~cpu_to_le32(LOV_PATTERN_F_HOLE);
1480 lfsck_buf_init(&ea_buf, lmm, lov_mds_md_size(count, magic));
1481 rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV, fl, handle);
1489 * \retval +1: repaired
1490 * \retval 0: did nothing
1491 * \retval -ve: on error
1493 static int lfsck_layout_extend_lovea(const struct lu_env *env,
1494 struct lfsck_instance *lfsck,
1495 struct thandle *handle,
1496 struct dt_object *parent,
1497 struct lu_fid *cfid,
1498 struct lu_buf *buf, int fl,
1499 __u32 ost_idx, __u32 ea_off, bool reset)
1501 struct lov_mds_md_v1 *lmm = buf->lb_buf;
1502 struct lov_ost_data_v1 *objs;
1508 if (fl == LU_XATTR_CREATE || reset) {
1509 __u32 pattern = LOV_PATTERN_RAID0;
1512 LASSERT(buf->lb_len >= lov_mds_md_size(count, LOV_MAGIC_V1));
1514 if (ea_off != 0 || reset) {
1515 pattern |= LOV_PATTERN_F_HOLE;
1519 memset(lmm, 0, buf->lb_len);
1520 lmm->lmm_magic = cpu_to_le32(LOV_MAGIC_V1);
1521 lmm->lmm_pattern = cpu_to_le32(pattern);
1522 fid_to_lmm_oi(lfsck_dto2fid(parent), &lmm->lmm_oi);
1523 lmm_oi_cpu_to_le(&lmm->lmm_oi, &lmm->lmm_oi);
1525 rc = lfsck_layout_get_def_stripesize(env, lfsck,
1526 &lmm->lmm_stripe_size);
1530 objs = &lmm->lmm_objects[ea_off];
1532 __u32 magic = le32_to_cpu(lmm->lmm_magic);
1535 count = le16_to_cpu(lmm->lmm_stripe_count);
1536 if (magic == LOV_MAGIC_V1)
1537 objs = &lmm->lmm_objects[count];
1539 objs = &((struct lov_mds_md_v3 *)lmm)->
1542 gap = ea_off - count;
1545 LASSERT(buf->lb_len >= lov_mds_md_size(count, magic));
1548 memset(objs, 0, gap * sizeof(*objs));
1549 lmm->lmm_pattern |= cpu_to_le32(LOV_PATTERN_F_HOLE);
1553 lmm->lmm_layout_gen =
1554 cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
1558 lmm->lmm_stripe_count = cpu_to_le16(count);
1559 rc = lfsck_layout_refill_lovea(env, handle, parent, cfid, buf, objs,
1562 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant extend layout EA for "
1563 DFID": parent "DFID", OST-index %u, stripe-index %u, fl %d, "
1564 "reset %s, %s LOV EA hole: rc = %d\n",
1565 lfsck_lfsck2name(lfsck), PFID(cfid), PFID(lfsck_dto2fid(parent)),
1566 ost_idx, ea_off, fl, reset ? "yes" : "no",
1567 hole ? "with" : "without", rc);
1572 static int __lfsck_layout_update_pfid(const struct lu_env *env,
1573 struct dt_object *child,
1574 const struct lu_fid *pfid, __u32 offset)
1576 struct dt_device *dev = lfsck_obj2dev(child);
1577 struct filter_fid *ff = &lfsck_env_info(env)->lti_new_pfid;
1578 struct thandle *handle;
1579 struct lu_buf buf = { NULL };
1582 ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
1583 ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
1584 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
1585 * MDT-object's FID::f_ver, instead it is the OST-object index in its
1586 * parent MDT-object's layout EA. */
1587 ff->ff_parent.f_stripe_idx = cpu_to_le32(offset);
1588 lfsck_buf_init(&buf, ff, sizeof(struct filter_fid));
1590 handle = dt_trans_create(env, dev);
1592 RETURN(PTR_ERR(handle));
1594 rc = dt_declare_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
1598 rc = dt_trans_start_local(env, dev, handle);
1602 rc = dt_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
1607 dt_trans_stop(env, dev, handle);
1613 * \retval +1: repaired
1614 * \retval 0: did nothing
1615 * \retval -ve: on error
1617 static int lfsck_layout_update_pfid(const struct lu_env *env,
1618 struct lfsck_component *com,
1619 struct dt_object *parent,
1620 struct lu_fid *cfid,
1621 struct dt_device *cdev, __u32 ea_off)
1623 struct dt_object *child;
1627 child = lfsck_object_find_by_dev(env, cdev, cfid);
1629 RETURN(PTR_ERR(child));
1631 rc = __lfsck_layout_update_pfid(env, child,
1632 lu_object_fid(&parent->do_lu), ea_off);
1633 lfsck_object_put(env, child);
1635 RETURN(rc == 0 ? 1 : rc);
1639 * This function will create the MDT-object with the given (partial) LOV EA.
1641 * Under some data corruption cases, the MDT-object of the file may be lost,
1642 * but its OST-objects, or some of them are there. The layout LFSCK needs to
1643 * re-create the MDT-object with the orphan OST-object(s) information.
1645 * On the other hand, the LFSCK may has created some OST-object for repairing
1646 * dangling LOV EA reference, but as the LFSCK processing, it may find that
1647 * the old OST-object is there and should replace the former new created OST
1648 * object. Unfortunately, some others have modified such newly created object.
1649 * To keep the data (both new and old), the LFSCK will create MDT-object with
1650 * new FID to reference the original OST-object.
1652 * \param[in] env pointer to the thread context
1653 * \param[in] com pointer to the lfsck component
1654 * \param[in] ltd pointer to target device descriptor
1655 * \param[in] rec pointer to the record for the orphan OST-object
1656 * \param[in] cfid pointer to FID for the orphan OST-object
1657 * \param[in] infix additional information, such as the FID for original
1658 * MDT-object and the stripe offset in the LOV EA
1659 * \param[in] type the type for describing why the orphan MDT-object is
1660 * created. The rules are as following:
1662 * type "C": Multiple OST-objects claim the same MDT-object and the
1663 * same slot in the layout EA. Then the LFSCK will create
1664 * new MDT-object(s) to hold the conflict OST-object(s).
1666 * type "N": The orphan OST-object does not know which one was the
1667 * real parent MDT-object, so the LFSCK uses new FID for
1668 * its parent MDT-object.
1670 * type "R": The orphan OST-object knows its parent MDT-object FID,
1671 * but does not know the position (the file name) in the
1674 * type "D": The MDT-object is a directory, it may knows its parent
1675 * but because there is no valid linkEA, the LFSCK cannot
1676 * know where to put it back to the namespace.
1677 * type "O": The MDT-object has no linkEA, and there is no name
1678 * entry that references the MDT-object.
1680 * type "P": The orphan object to be created was a parent directory
1681 * of some MDT-object which linkEA shows that the @orphan
1682 * object is missing.
1684 * The orphan name will be like:
1685 * ${FID}-${infix}-${type}-${conflict_version}
1687 * \param[in] ea_off the stripe offset in the LOV EA
1689 * \retval positive on repaired something
1690 * \retval 0 if needs to repair nothing
1691 * \retval negative error number on failure
1693 static int lfsck_layout_recreate_parent(const struct lu_env *env,
1694 struct lfsck_component *com,
1695 struct lfsck_tgt_desc *ltd,
1696 struct lu_orphan_rec *rec,
1697 struct lu_fid *cfid,
1702 struct lfsck_thread_info *info = lfsck_env_info(env);
1703 struct dt_insert_rec *dtrec = &info->lti_dt_rec;
1704 char *name = info->lti_key;
1705 struct lu_attr *la = &info->lti_la;
1706 struct dt_object_format *dof = &info->lti_dof;
1707 struct lfsck_instance *lfsck = com->lc_lfsck;
1708 struct lu_fid *pfid = &rec->lor_fid;
1709 struct lu_fid *tfid = &info->lti_fid3;
1710 struct dt_device *dev = lfsck->li_bottom;
1711 struct dt_object *lpf = lfsck->li_lpf_obj;
1712 struct dt_object *pobj = NULL;
1713 struct dt_object *cobj = NULL;
1714 struct thandle *th = NULL;
1715 struct lu_buf *ea_buf = &info->lti_big_buf;
1716 struct lu_buf lov_buf;
1717 struct lfsck_lock_handle *llh = &info->lti_llh;
1718 struct linkea_data ldata = { NULL };
1719 struct lu_buf linkea_buf;
1720 const struct lu_name *pname;
1726 if (unlikely(lpf == NULL))
1727 GOTO(log, rc = -ENXIO);
1729 /* We use two separated transactions to repair the inconsistency.
1731 * 1) create the MDT-object locally.
1732 * 2) update the OST-object's PFID EA if necessary.
1734 * If 1) succeed, but 2) failed, then the OST-object's PFID EA will be
1735 * updated when the layout LFSCK run next time.
1737 * If 1) failed, but 2) succeed, then such MDT-object will be re-created
1738 * when the layout LFSCK run next time. */
1740 if (fid_is_zero(pfid)) {
1741 rc = lfsck_fid_alloc(env, lfsck, pfid, false);
1745 cobj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, cfid);
1747 GOTO(log, rc = PTR_ERR(cobj));
1750 pobj = lfsck_object_find_by_dev(env, dev, pfid);
1752 GOTO(log, rc = PTR_ERR(pobj));
1754 LASSERT(infix != NULL);
1755 LASSERT(type != NULL);
1757 memset(la, 0, sizeof(*la));
1758 la->la_uid = rec->lor_uid;
1759 la->la_gid = rec->lor_gid;
1760 la->la_mode = S_IFREG | S_IRUSR;
1761 la->la_valid = LA_MODE | LA_UID | LA_GID;
1763 memset(dof, 0, sizeof(*dof));
1764 dof->dof_type = dt_mode_to_dft(S_IFREG);
1765 /* Because the dof->dof_reg.striped = 0, the LOD will not create
1766 * the stripe(s). The LFSCK will specify the LOV EA via
1767 * lfsck_layout_extend_lovea(). */
1769 size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
1770 if (ea_buf->lb_len < size) {
1771 lu_buf_realloc(ea_buf, size);
1772 if (ea_buf->lb_buf == NULL)
1773 GOTO(log, rc = -ENOMEM);
1778 snprintf(name, NAME_MAX, DFID"%s-%s-%d", PFID(pfid), infix,
1780 rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
1781 (const struct dt_key *)name);
1782 if (rc != 0 && rc != -ENOENT)
1786 rc = lfsck_lock(env, lfsck, lfsck->li_lpf_obj, name, llh,
1787 MDS_INODELOCK_UPDATE, LCK_PW);
1791 /* Re-check whether the name conflict with othrs after taken
1793 rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
1794 (const struct dt_key *)name);
1795 if (unlikely(rc == 0)) {
1803 rc = linkea_data_new(&ldata,
1804 &lfsck_env_info(env)->lti_linkea_buf);
1808 pname = lfsck_name_get_const(env, name, strlen(name));
1809 rc = linkea_add_buf(&ldata, pname, lfsck_dto2fid(lfsck->li_lpf_obj));
1813 /* The 1st transaction. */
1814 th = dt_trans_create(env, dev);
1816 GOTO(unlock, rc = PTR_ERR(th));
1818 rc = dt_declare_create(env, pobj, la, NULL, dof, th);
1822 lfsck_buf_init(&lov_buf, ea_buf->lb_buf, size);
1823 rc = dt_declare_xattr_set(env, pobj, &lov_buf, XATTR_NAME_LOV,
1824 LU_XATTR_REPLACE, th);
1828 dtrec->rec_fid = pfid;
1829 dtrec->rec_type = S_IFREG;
1830 rc = dt_declare_insert(env, lpf,
1831 (const struct dt_rec *)dtrec,
1832 (const struct dt_key *)name, th);
1836 lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
1837 ldata.ld_leh->leh_len);
1838 rc = dt_declare_xattr_set(env, pobj, &linkea_buf,
1839 XATTR_NAME_LINK, 0, th);
1843 rc = dt_trans_start_local(env, dev, th);
1847 dt_write_lock(env, pobj, 0);
1848 rc = dt_create(env, pobj, la, NULL, dof, th);
1850 rc = lfsck_layout_extend_lovea(env, lfsck, th, pobj, cfid,
1851 &lov_buf, 0, ltd->ltd_index, ea_off, true);
1852 dt_write_unlock(env, pobj);
1856 rc = dt_insert(env, lpf, (const struct dt_rec *)dtrec,
1857 (const struct dt_key *)name, th, 1);
1861 rc = dt_xattr_set(env, pobj, &linkea_buf, XATTR_NAME_LINK, 0, th);
1862 if (rc == 0 && cobj != NULL) {
1863 dt_trans_stop(env, dev, th);
1866 /* The 2nd transaction. */
1867 rc = __lfsck_layout_update_pfid(env, cobj, pfid, ea_off);
1874 dt_trans_stop(env, dev, th);
1880 if (cobj != NULL && !IS_ERR(cobj))
1881 lfsck_object_put(env, cobj);
1882 if (pobj != NULL && !IS_ERR(pobj))
1883 lfsck_object_put(env, pobj);
1886 CDEBUG(D_LFSCK, "%s layout LFSCK assistant failed to "
1887 "recreate the lost MDT-object: parent "DFID
1888 ", child "DFID", OST-index %u, stripe-index %u, "
1889 "infix %s, type %s: rc = %d\n",
1890 lfsck_lfsck2name(lfsck), PFID(pfid), PFID(cfid),
1891 ltd->ltd_index, ea_off, infix, type, rc);
1893 return rc >= 0 ? 1 : rc;
1896 static int lfsck_layout_master_conditional_destroy(const struct lu_env *env,
1897 struct lfsck_component *com,
1898 const struct lu_fid *fid,
1901 struct lfsck_thread_info *info = lfsck_env_info(env);
1902 struct lfsck_request *lr = &info->lti_lr;
1903 struct lfsck_instance *lfsck = com->lc_lfsck;
1904 struct lfsck_tgt_desc *ltd;
1905 struct ptlrpc_request *req;
1906 struct lfsck_request *tmp;
1907 struct obd_export *exp;
1911 ltd = lfsck_tgt_get(&lfsck->li_ost_descs, index);
1912 if (unlikely(ltd == NULL))
1916 if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
1917 GOTO(put, rc = -EOPNOTSUPP);
1919 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
1921 GOTO(put, rc = -ENOMEM);
1923 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
1925 ptlrpc_request_free(req);
1930 memset(lr, 0, sizeof(*lr));
1931 lr->lr_event = LE_CONDITIONAL_DESTROY;
1932 lr->lr_active = LFSCK_TYPE_LAYOUT;
1935 tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
1937 ptlrpc_request_set_replen(req);
1939 rc = ptlrpc_queue_wait(req);
1940 ptlrpc_req_finished(req);
1950 static int lfsck_layout_slave_conditional_destroy(const struct lu_env *env,
1951 struct lfsck_component *com,
1952 struct lfsck_request *lr)
1954 struct lfsck_thread_info *info = lfsck_env_info(env);
1955 struct lu_attr *la = &info->lti_la;
1956 ldlm_policy_data_t *policy = &info->lti_policy;
1957 struct ldlm_res_id *resid = &info->lti_resid;
1958 struct lfsck_instance *lfsck = com->lc_lfsck;
1959 struct dt_device *dev = lfsck->li_bottom;
1960 struct lu_fid *fid = &lr->lr_fid;
1961 struct dt_object *obj;
1962 struct thandle *th = NULL;
1963 struct lustre_handle lh = { 0 };
1968 obj = lfsck_object_find_by_dev(env, dev, fid);
1970 RETURN(PTR_ERR(obj));
1972 dt_read_lock(env, obj, 0);
1973 if (dt_object_exists(obj) == 0 ||
1974 lfsck_is_dead_obj(obj)) {
1975 dt_read_unlock(env, obj);
1977 GOTO(put, rc = -ENOENT);
1980 /* Get obj's attr without lock firstly. */
1981 rc = dt_attr_get(env, obj, la);
1982 dt_read_unlock(env, obj);
1986 if (likely(la->la_ctime != 0 || la->la_mode & S_ISUID))
1987 GOTO(put, rc = -ETXTBSY);
1989 /* Acquire extent lock on [0, EOF] to sync with all possible written. */
1990 LASSERT(lfsck->li_namespace != NULL);
1992 memset(policy, 0, sizeof(*policy));
1993 policy->l_extent.end = OBD_OBJECT_EOF;
1994 ost_fid_build_resid(fid, resid);
1995 rc = ldlm_cli_enqueue_local(lfsck->li_namespace, resid, LDLM_EXTENT,
1996 policy, LCK_EX, &flags, ldlm_blocking_ast,
1997 ldlm_completion_ast, NULL, NULL, 0,
1998 LVB_T_NONE, NULL, &lh);
2000 GOTO(put, rc = -EIO);
2002 dt_write_lock(env, obj, 0);
2003 /* Get obj's attr within lock again. */
2004 rc = dt_attr_get(env, obj, la);
2008 if (la->la_ctime != 0)
2009 GOTO(unlock, rc = -ETXTBSY);
2011 th = dt_trans_create(env, dev);
2013 GOTO(unlock, rc = PTR_ERR(th));
2015 rc = dt_declare_ref_del(env, obj, th);
2019 rc = dt_declare_destroy(env, obj, th);
2023 rc = dt_trans_start_local(env, dev, th);
2027 rc = dt_ref_del(env, obj, th);
2031 rc = dt_destroy(env, obj, th);
2033 CDEBUG(D_LFSCK, "%s: layout LFSCK destroyed the empty "
2034 "OST-object "DFID" that was created for reparing "
2035 "dangling referenced case. But the original missing "
2036 "OST-object is found now.\n",
2037 lfsck_lfsck2name(lfsck), PFID(fid));
2042 dt_trans_stop(env, dev, th);
2045 dt_write_unlock(env, obj);
2046 ldlm_lock_decref(&lh, LCK_EX);
2049 lfsck_object_put(env, obj);
2055 * Some OST-object has occupied the specified layout EA slot.
2056 * Such OST-object may be generated by the LFSCK when repair
2057 * dangling referenced MDT-object, which can be indicated by
2058 * attr::la_ctime == 0 but without S_ISUID in la_mode. If it
2059 * is true and such OST-object has not been modified yet, we
2060 * will replace it with the orphan OST-object; otherwise the
2061 * LFSCK will create new MDT-object to reference the orphan.
2063 * \retval +1: repaired
2064 * \retval 0: did nothing
2065 * \retval -ve: on error
2067 static int lfsck_layout_conflict_create(const struct lu_env *env,
2068 struct lfsck_component *com,
2069 struct lfsck_tgt_desc *ltd,
2070 struct lu_orphan_rec *rec,
2071 struct dt_object *parent,
2072 struct lu_fid *cfid,
2073 struct lu_buf *ea_buf,
2074 struct lov_ost_data_v1 *slot,
2077 struct lfsck_thread_info *info = lfsck_env_info(env);
2078 struct lu_fid *cfid2 = &info->lti_fid2;
2079 struct ost_id *oi = &info->lti_oi;
2080 struct lov_mds_md_v1 *lmm = ea_buf->lb_buf;
2081 struct dt_device *dev = lfsck_obj2dev(parent);
2082 struct thandle *th = NULL;
2083 struct lustre_handle lh = { 0 };
2084 __u32 ost_idx2 = le32_to_cpu(slot->l_ost_idx);
2088 ostid_le_to_cpu(&slot->l_ost_oi, oi);
2089 rc = ostid_to_fid(cfid2, oi, ost_idx2);
2093 rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2094 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2099 rc = lfsck_layout_master_conditional_destroy(env, com, cfid2, ost_idx2);
2101 /* If the conflict OST-obejct is not created for fixing dangling
2102 * referenced MDT-object in former LFSCK check/repair, or it has
2103 * been modified by others, then we cannot destroy it. Re-create
2104 * a new MDT-object for the orphan OST-object. */
2105 if (rc == -ETXTBSY) {
2106 /* No need the layout lock on the original parent. */
2107 lfsck_ibits_unlock(&lh, LCK_EX);
2109 fid_zero(&rec->lor_fid);
2110 snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2111 "-"DFID"-%x", PFID(lu_object_fid(&parent->do_lu)),
2113 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2114 info->lti_tmpbuf, "C", ea_off);
2119 if (rc != 0 && rc != -ENOENT)
2122 th = dt_trans_create(env, dev);
2124 GOTO(unlock, rc = PTR_ERR(th));
2126 rc = dt_declare_xattr_set(env, parent, ea_buf, XATTR_NAME_LOV,
2127 LU_XATTR_REPLACE, th);
2131 rc = dt_trans_start_local(env, dev, th);
2135 dt_write_lock(env, parent, 0);
2136 lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2137 rc = lfsck_layout_refill_lovea(env, th, parent, cfid, ea_buf, slot,
2138 LU_XATTR_REPLACE, ltd->ltd_index);
2139 dt_write_unlock(env, parent);
2144 dt_trans_stop(env, dev, th);
2147 lfsck_ibits_unlock(&lh, LCK_EX);
2150 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant replaced the conflict "
2151 "OST-object "DFID" on the OST %x with the orphan "DFID" on "
2152 "the OST %x: parent "DFID", stripe-index %u: rc = %d\n",
2153 lfsck_lfsck2name(com->lc_lfsck), PFID(cfid2), ost_idx2,
2154 PFID(cfid), ltd->ltd_index, PFID(lfsck_dto2fid(parent)),
2157 return rc >= 0 ? 1 : rc;
2161 * \retval +1: repaired
2162 * \retval 0: did nothing
2163 * \retval -ve: on error
2165 static int lfsck_layout_recreate_lovea(const struct lu_env *env,
2166 struct lfsck_component *com,
2167 struct lfsck_tgt_desc *ltd,
2168 struct lu_orphan_rec *rec,
2169 struct dt_object *parent,
2170 struct lu_fid *cfid,
2171 __u32 ost_idx, __u32 ea_off)
2173 struct lfsck_thread_info *info = lfsck_env_info(env);
2174 struct lu_buf *buf = &info->lti_big_buf;
2175 struct lu_fid *fid = &info->lti_fid2;
2176 struct ost_id *oi = &info->lti_oi;
2177 struct lfsck_instance *lfsck = com->lc_lfsck;
2178 struct dt_device *dt = lfsck_obj2dev(parent);
2179 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
2180 struct thandle *handle = NULL;
2182 struct lov_mds_md_v1 *lmm;
2183 struct lov_ost_data_v1 *objs;
2184 struct lustre_handle lh = { 0 };
2191 bool locked = false;
2194 rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2195 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2198 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant failed to recreate "
2199 "LOV EA for "DFID": parent "DFID", OST-index %u, "
2200 "stripe-index %u: rc = %d\n",
2201 lfsck_lfsck2name(lfsck), PFID(cfid),
2202 PFID(lfsck_dto2fid(parent)), ost_idx, ea_off, rc);
2209 dt_write_unlock(env, parent);
2213 if (handle != NULL) {
2214 dt_trans_stop(env, dt, handle);
2219 GOTO(unlock_layout, rc);
2222 if (buf->lb_len < lovea_size) {
2223 lu_buf_realloc(buf, lovea_size);
2224 if (buf->lb_buf == NULL)
2225 GOTO(unlock_layout, rc = -ENOMEM);
2228 if (!(bk->lb_param & LPF_DRYRUN)) {
2229 handle = dt_trans_create(env, dt);
2231 GOTO(unlock_layout, rc = PTR_ERR(handle));
2233 rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2238 rc = dt_trans_start_local(env, dt, handle);
2243 dt_write_lock(env, parent, 0);
2245 rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV);
2246 if (rc == -ERANGE) {
2247 rc = dt_xattr_get(env, parent, &LU_BUF_NULL, XATTR_NAME_LOV);
2250 } else if (rc == -ENODATA || rc == 0) {
2251 lovea_size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
2252 /* If the declared is not big enough, re-try. */
2253 if (buf->lb_len < lovea_size) {
2257 fl = LU_XATTR_CREATE;
2258 } else if (rc < 0) {
2259 GOTO(unlock_parent, rc);
2260 } else if (unlikely(buf->lb_len == 0)) {
2263 fl = LU_XATTR_REPLACE;
2267 if (fl == LU_XATTR_CREATE) {
2268 if (bk->lb_param & LPF_DRYRUN)
2269 GOTO(unlock_parent, rc = 1);
2271 LASSERT(buf->lb_len >= lovea_size);
2273 rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
2274 buf, fl, ost_idx, ea_off, false);
2276 GOTO(unlock_parent, rc);
2280 rc1 = lfsck_layout_verify_header(lmm);
2282 /* If the LOV EA crashed, the rebuild it. */
2283 if (rc1 == -EINVAL) {
2284 if (bk->lb_param & LPF_DRYRUN)
2285 GOTO(unlock_parent, rc = 1);
2287 LASSERT(buf->lb_len >= lovea_size);
2289 rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
2290 buf, fl, ost_idx, ea_off, true);
2292 GOTO(unlock_parent, rc);
2295 /* For other unknown magic/pattern, keep the current LOV EA. */
2297 GOTO(unlock_parent, rc = rc1);
2299 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
2300 * been verified in lfsck_layout_verify_header() already. If some
2301 * new magic introduced in the future, then layout LFSCK needs to
2302 * be updated also. */
2303 magic = le32_to_cpu(lmm->lmm_magic);
2304 if (magic == LOV_MAGIC_V1) {
2305 objs = &lmm->lmm_objects[0];
2307 LASSERT(magic == LOV_MAGIC_V3);
2308 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
2311 count = le16_to_cpu(lmm->lmm_stripe_count);
2313 GOTO(unlock_parent, rc = -EINVAL);
2316 /* Exceed the current end of MDT-object layout EA. Then extend it. */
2317 if (count <= ea_off) {
2318 if (bk->lb_param & LPF_DRYRUN)
2319 GOTO(unlock_parent, rc = 1);
2321 lovea_size = lov_mds_md_size(ea_off + 1, magic);
2322 /* If the declared is not big enough, re-try. */
2323 if (buf->lb_len < lovea_size) {
2328 rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
2329 buf, fl, ost_idx, ea_off, false);
2331 GOTO(unlock_parent, rc);
2334 LASSERTF(rc > 0, "invalid rc = %d\n", rc);
2336 for (i = 0; i < count; i++, objs++) {
2337 /* The MDT-object was created via lfsck_layout_recover_create()
2338 * by others before, and we fill the dummy layout EA. */
2339 if (lovea_slot_is_dummy(objs)) {
2343 if (bk->lb_param & LPF_DRYRUN)
2344 GOTO(unlock_parent, rc = 1);
2346 lmm->lmm_layout_gen =
2347 cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2348 rc = lfsck_layout_refill_lovea(env, handle, parent,
2349 cfid, buf, objs, fl,
2352 CDEBUG(D_LFSCK, "%s layout LFSCK assistant fill "
2353 "dummy layout slot for "DFID": parent "DFID
2354 ", OST-index %u, stripe-index %u: rc = %d\n",
2355 lfsck_lfsck2name(lfsck), PFID(cfid),
2356 PFID(lfsck_dto2fid(parent)), ost_idx, i, rc);
2358 GOTO(unlock_parent, rc);
2361 ostid_le_to_cpu(&objs->l_ost_oi, oi);
2362 rc = ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
2364 CDEBUG(D_LFSCK, "%s: the parent "DFID" contains "
2365 "invalid layout EA at the slot %d, index %u\n",
2366 lfsck_lfsck2name(lfsck),
2367 PFID(lfsck_dto2fid(parent)), i,
2368 le32_to_cpu(objs->l_ost_idx));
2370 GOTO(unlock_parent, rc);
2373 /* It should be rare case, the slot is there, but the LFSCK
2374 * does not handle it during the first-phase cycle scanning. */
2375 if (unlikely(lu_fid_eq(fid, cfid))) {
2377 GOTO(unlock_parent, rc = 0);
2379 /* Rare case that the OST-object index
2380 * does not match the parent MDT-object
2381 * layout EA. We trust the later one. */
2382 if (bk->lb_param & LPF_DRYRUN)
2383 GOTO(unlock_parent, rc = 1);
2385 dt_write_unlock(env, parent);
2387 dt_trans_stop(env, dt, handle);
2388 lfsck_ibits_unlock(&lh, LCK_EX);
2389 rc = lfsck_layout_update_pfid(env, com, parent,
2390 cfid, ltd->ltd_tgt, i);
2392 CDEBUG(D_LFSCK, "%s layout LFSCK assistant "
2393 "updated OST-object's pfid for "DFID
2394 ": parent "DFID", OST-index %u, "
2395 "stripe-index %u: rc = %d\n",
2396 lfsck_lfsck2name(lfsck), PFID(cfid),
2397 PFID(lfsck_dto2fid(parent)),
2398 ltd->ltd_index, i, rc);
2405 /* The MDT-object exists, but related layout EA slot is occupied
2407 if (bk->lb_param & LPF_DRYRUN)
2408 GOTO(unlock_parent, rc = 1);
2410 dt_write_unlock(env, parent);
2412 dt_trans_stop(env, dt, handle);
2413 lfsck_ibits_unlock(&lh, LCK_EX);
2414 if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1)
2415 objs = &lmm->lmm_objects[ea_off];
2417 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[ea_off];
2418 rc = lfsck_layout_conflict_create(env, com, ltd, rec, parent, cfid,
2425 dt_write_unlock(env, parent);
2429 dt_trans_stop(env, dt, handle);
2432 lfsck_ibits_unlock(&lh, LCK_EX);
2437 static int lfsck_layout_scan_orphan_one(const struct lu_env *env,
2438 struct lfsck_component *com,
2439 struct lfsck_tgt_desc *ltd,
2440 struct lu_orphan_rec *rec,
2441 struct lu_fid *cfid)
2443 struct lfsck_layout *lo = com->lc_file_ram;
2444 struct lu_fid *pfid = &rec->lor_fid;
2445 struct dt_object *parent = NULL;
2446 __u32 ea_off = pfid->f_stripe_idx;
2450 if (!fid_is_sane(cfid))
2451 GOTO(out, rc = -EINVAL);
2453 if (fid_is_zero(pfid)) {
2454 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2460 if (!fid_is_sane(pfid))
2461 GOTO(out, rc = -EINVAL);
2463 parent = lfsck_object_find_by_dev(env, com->lc_lfsck->li_bottom, pfid);
2465 GOTO(out, rc = PTR_ERR(parent));
2467 if (unlikely(dt_object_remote(parent) != 0))
2468 GOTO(put, rc = -EXDEV);
2470 if (dt_object_exists(parent) == 0) {
2471 lfsck_object_put(env, parent);
2472 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2477 if (!S_ISREG(lu_object_attr(&parent->do_lu)))
2478 GOTO(put, rc = -EISDIR);
2480 rc = lfsck_layout_recreate_lovea(env, com, ltd, rec, parent, cfid,
2481 ltd->ltd_index, ea_off);
2487 lfsck_object_put(env, parent);
2489 /* The layout EA is changed, need to be reloaded next time. */
2490 lu_object_put_nocache(env, &parent->do_lu);
2493 down_write(&com->lc_sem);
2494 com->lc_new_scanned++;
2495 com->lc_new_checked++;
2497 lo->ll_objs_repaired[LLIT_ORPHAN - 1]++;
2499 } else if (rc < 0) {
2500 lo->ll_objs_failed_phase2++;
2502 up_write(&com->lc_sem);
2507 static int lfsck_layout_scan_orphan(const struct lu_env *env,
2508 struct lfsck_component *com,
2509 struct lfsck_tgt_desc *ltd)
2511 struct lfsck_assistant_data *lad = com->lc_data;
2512 struct lfsck_instance *lfsck = com->lc_lfsck;
2513 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
2514 struct lfsck_thread_info *info = lfsck_env_info(env);
2515 struct ost_id *oi = &info->lti_oi;
2516 struct lu_fid *fid = &info->lti_fid;
2517 struct dt_object *obj;
2518 const struct dt_it_ops *iops;
2523 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant starts the orphan "
2524 "scanning for OST%04x\n",
2525 lfsck_lfsck2name(lfsck), ltd->ltd_index);
2527 if (cfs_bitmap_check(lad->lad_bitmap, ltd->ltd_index)) {
2528 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant skip the orphan "
2529 "scanning for OST%04x\n",
2530 lfsck_lfsck2name(lfsck), ltd->ltd_index);
2535 ostid_set_seq(oi, FID_SEQ_IDIF);
2536 ostid_set_id(oi, 0);
2537 rc = ostid_to_fid(fid, oi, ltd->ltd_index);
2541 obj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, fid);
2542 if (unlikely(IS_ERR(obj)))
2543 GOTO(log, rc = PTR_ERR(obj));
2545 rc = obj->do_ops->do_index_try(env, obj, &dt_lfsck_orphan_features);
2549 iops = &obj->do_index_ops->dio_it;
2550 di = iops->init(env, obj, 0);
2552 GOTO(put, rc = PTR_ERR(di));
2554 rc = iops->load(env, di, 0);
2556 /* -ESRCH means that the orphan OST-objects rbtree has been
2557 * cleanup because of the OSS server restart or other errors. */
2558 lfsck_lad_set_bitmap(env, com, ltd->ltd_index);
2563 rc = iops->next(env, di);
2575 struct lu_orphan_rec *rec = &info->lti_rec;
2577 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
2578 unlikely(!thread_is_running(&lfsck->li_thread)))
2581 key = iops->key(env, di);
2582 com->lc_fid_latest_scanned_phase2 = *(struct lu_fid *)key;
2583 rc = iops->rec(env, di, (struct dt_rec *)rec, 0);
2585 rc = lfsck_layout_scan_orphan_one(env, com, ltd, rec,
2586 &com->lc_fid_latest_scanned_phase2);
2587 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
2590 lfsck_control_speed_by_self(com);
2592 rc = iops->next(env, di);
2593 } while (rc < 0 && !(bk->lb_param & LPF_FAILOUT));
2600 iops->fini(env, di);
2602 lfsck_object_put(env, obj);
2605 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant finished the orphan "
2606 "scanning for OST%04x: rc = %d\n",
2607 lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
2609 return rc > 0 ? 0 : rc;
2612 /* For the MDT-object with dangling reference, we need to repare the
2613 * inconsistency according to the LFSCK sponsor's requirement:
2615 * 1) Keep the inconsistency there and report the inconsistency case,
2616 * then give the chance to the application to find related issues,
2617 * and the users can make the decision about how to handle it with
2618 * more human knownledge. (by default)
2620 * 2) Re-create the missing OST-object with the FID/owner information. */
2621 static int lfsck_layout_repair_dangling(const struct lu_env *env,
2622 struct lfsck_component *com,
2623 struct dt_object *parent,
2624 struct lfsck_layout_req *llr,
2627 struct lfsck_thread_info *info = lfsck_env_info(env);
2628 struct filter_fid *pfid = &info->lti_new_pfid;
2629 struct dt_object_format *dof = &info->lti_dof;
2630 struct dt_object *child = llr->llr_child;
2631 struct dt_device *dev = lfsck_obj2dev(child);
2632 const struct lu_fid *tfid = lu_object_fid(&parent->do_lu);
2633 struct thandle *handle;
2635 struct lustre_handle lh = { 0 };
2640 if (com->lc_lfsck->li_bookmark_ram.lb_param & LPF_CREATE_OSTOBJ)
2648 rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2649 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2654 rc = dt_attr_get(env, parent, la);
2658 la->la_mode = S_IFREG | 0666;
2659 la->la_atime = la->la_mtime = la->la_ctime = 0;
2660 la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
2661 LA_ATIME | LA_MTIME | LA_CTIME;
2662 memset(dof, 0, sizeof(*dof));
2663 pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
2664 pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
2665 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
2666 * MDT-object's FID::f_ver, instead it is the OST-object index in its
2667 * parent MDT-object's layout EA. */
2668 pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
2669 buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
2671 handle = dt_trans_create(env, dev);
2673 GOTO(unlock1, rc = PTR_ERR(handle));
2675 rc = dt_declare_create(env, child, la, NULL, dof, handle);
2679 rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID,
2680 LU_XATTR_CREATE, handle);
2684 rc = dt_trans_start_local(env, dev, handle);
2688 dt_read_lock(env, parent, 0);
2689 if (unlikely(lfsck_is_dead_obj(parent)))
2690 GOTO(unlock2, rc = 1);
2692 rc = dt_create(env, child, la, NULL, dof, handle);
2696 rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, LU_XATTR_CREATE,
2702 dt_read_unlock(env, parent);
2705 rc = lfsck_layout_trans_stop(env, dev, handle, rc);
2708 lfsck_ibits_unlock(&lh, LCK_EX);
2711 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant found dangling "
2712 "reference for: parent "DFID", child "DFID", OST-index %u, "
2713 "stripe-index %u, owner %u/%u. %s: rc = %d\n",
2714 lfsck_lfsck2name(com->lc_lfsck), PFID(lfsck_dto2fid(parent)),
2715 PFID(lfsck_dto2fid(child)), llr->llr_ost_idx,
2716 llr->llr_lov_idx, la->la_uid, la->la_gid,
2717 create ? "Create the lost OST-object as required" :
2718 "Keep the MDT-object there by default", rc);
2723 /* If the OST-object does not recognize the MDT-object as its parent, and
2724 * there is no other MDT-object claims as its parent, then just trust the
2725 * given MDT-object as its parent. So update the OST-object filter_fid. */
2726 static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
2727 struct lfsck_component *com,
2728 struct dt_object *parent,
2729 struct lfsck_layout_req *llr,
2732 struct lfsck_thread_info *info = lfsck_env_info(env);
2733 struct filter_fid *pfid = &info->lti_new_pfid;
2734 struct dt_object *child = llr->llr_child;
2735 struct dt_device *dev = lfsck_obj2dev(child);
2736 const struct lu_fid *tfid = lu_object_fid(&parent->do_lu);
2737 struct thandle *handle;
2739 struct lustre_handle lh = { 0 };
2743 rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2744 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2749 pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
2750 pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
2751 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
2752 * MDT-object's FID::f_ver, instead it is the OST-object index in its
2753 * parent MDT-object's layout EA. */
2754 pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
2755 buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
2757 handle = dt_trans_create(env, dev);
2759 GOTO(unlock1, rc = PTR_ERR(handle));
2761 rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
2765 rc = dt_attr_get(env, parent, la);
2769 la->la_valid = LA_UID | LA_GID;
2770 rc = dt_declare_attr_set(env, child, la, handle);
2774 rc = dt_trans_start_local(env, dev, handle);
2778 dt_write_lock(env, parent, 0);
2779 if (unlikely(lfsck_is_dead_obj(parent)))
2780 GOTO(unlock2, rc = 1);
2782 rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
2786 /* Get the latest parent's owner. */
2787 rc = dt_attr_get(env, parent, la);
2791 la->la_valid = LA_UID | LA_GID;
2792 rc = dt_attr_set(env, child, la, handle);
2797 dt_write_unlock(env, parent);
2800 rc = lfsck_layout_trans_stop(env, dev, handle, rc);
2803 lfsck_ibits_unlock(&lh, LCK_EX);
2806 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired unmatched "
2807 "MDT-OST pair for: parent "DFID", child "DFID", OST-index %u, "
2808 "stripe-index %u, owner %u/%u: rc = %d\n",
2809 lfsck_lfsck2name(com->lc_lfsck), PFID(lfsck_dto2fid(parent)),
2810 PFID(lfsck_dto2fid(child)), llr->llr_ost_idx, llr->llr_lov_idx,
2811 la->la_uid, la->la_gid, rc);
2816 /* If there are more than one MDT-objects claim as the OST-object's parent,
2817 * and the OST-object only recognizes one of them, then we need to generate
2818 * new OST-object(s) with new fid(s) for the non-recognized MDT-object(s). */
2819 static int lfsck_layout_repair_multiple_references(const struct lu_env *env,
2820 struct lfsck_component *com,
2821 struct dt_object *parent,
2822 struct lfsck_layout_req *llr,
2826 struct lfsck_thread_info *info = lfsck_env_info(env);
2827 struct dt_allocation_hint *hint = &info->lti_hint;
2828 struct dt_object_format *dof = &info->lti_dof;
2829 struct ost_id *oi = &info->lti_oi;
2830 struct lfsck_instance *lfsck = com->lc_lfsck;
2831 struct dt_device *dev;
2832 struct lu_device *d =
2833 &lfsck_obj2dev(llr->llr_child)->dd_lu_dev;
2834 struct lu_object *o;
2835 struct lu_object *n;
2836 struct dt_object *child = NULL;
2837 struct thandle *handle = NULL;
2838 struct lov_mds_md_v1 *lmm;
2839 struct lov_ost_data_v1 *objs;
2840 const struct lu_fid *pfid = lfsck_dto2fid(parent);
2842 struct lustre_handle lh = { 0 };
2843 struct lu_buf ea_buf;
2849 /* We use two separated transactions to repair the inconsistency.
2851 * 1) create the child (OST-object).
2852 * 2) update the parent LOV EA according to the child's FID.
2854 * If 1) succeed, but 2) failed or aborted, then such OST-object will be
2855 * handled as orphan when the layout LFSCK run next time.
2857 * If 1) failed, but 2) succeed, then such OST-object will be re-created
2858 * as dangling referened case when the layout LFSCK run next time. */
2860 /* The 1st transaction. */
2861 o = lu_object_anon(env, d, NULL);
2863 GOTO(log, rc = PTR_ERR(o));
2865 n = lu_object_locate(o->lo_header, d->ld_type);
2866 if (unlikely(n == NULL)) {
2867 lu_object_put_nocache(env, o);
2869 GOTO(log, rc = -EINVAL);
2872 child = container_of(n, struct dt_object, do_lu);
2873 memset(hint, 0, sizeof(*hint));
2874 rc = dt_attr_get(env, parent, la);
2878 la->la_valid = LA_UID | LA_GID;
2879 memset(dof, 0, sizeof(*dof));
2881 dev = lfsck_obj2dev(child);
2882 handle = dt_trans_create(env, dev);
2884 GOTO(log, rc = PTR_ERR(handle));
2886 rc = dt_declare_create(env, child, la, hint, dof, handle);
2890 rc = dt_trans_start_local(env, dev, handle);
2894 rc = dt_create(env, child, la, hint, dof, handle);
2895 dt_trans_stop(env, dev, handle);
2900 rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2901 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2906 /* The 2nd transaction. */
2908 /* XXX: Generally, we should use bottom device (OSD) to update parent
2909 * LOV EA. But because the LOD-object still references the wrong
2910 * OSP-object that should be detached after the parent's LOV EA
2911 * refreshed. Unfortunately, there is no suitable API for that.
2912 * So we have to make the LOD to re-load the OSP-object(s) via
2913 * replacing the LOV EA against the LOD-object.
2915 * Once the DNE2 patches have been landed, we can replace the
2916 * LOD device with the OSD device. LU-6230. */
2918 dev = lfsck->li_next;
2919 parent = lfsck_object_locate(dev, parent);
2921 GOTO(log, rc = PTR_ERR(parent));
2923 handle = dt_trans_create(env, dev);
2925 GOTO(log, rc = PTR_ERR(handle));
2927 rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2928 LU_XATTR_REPLACE, handle);
2932 rc = dt_trans_start_local(env, dev, handle);
2936 dt_write_lock(env, parent, 0);
2937 if (unlikely(lfsck_is_dead_obj(parent)))
2938 GOTO(unlock, rc = 0);
2940 rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV);
2941 if (unlikely(rc == 0 || rc == -ENODATA || rc == -ERANGE))
2942 GOTO(unlock, rc = 0);
2945 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
2946 * been verified in lfsck_layout_verify_header() already. If some
2947 * new magic introduced in the future, then layout LFSCK needs to
2948 * be updated also. */
2949 magic = le32_to_cpu(lmm->lmm_magic);
2950 if (magic == LOV_MAGIC_V1) {
2951 objs = &lmm->lmm_objects[llr->llr_lov_idx];
2953 LASSERT(magic == LOV_MAGIC_V3);
2955 &((struct lov_mds_md_v3 *)lmm)->lmm_objects[llr->llr_lov_idx];
2958 ostid_le_to_cpu(&objs->l_ost_oi, oi);
2959 index = le32_to_cpu(objs->l_ost_idx);
2960 rc = ostid_to_fid(&tfid, oi, index);
2961 /* Someone changed layout during the LFSCK, no need to repair then. */
2962 if (rc == 0 && !lu_fid_eq(&tfid, lu_object_fid(&llr->llr_child->do_lu)))
2963 GOTO(unlock, rc = 0);
2965 lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2966 fid_to_ostid(lu_object_fid(&child->do_lu), oi);
2967 ostid_cpu_to_le(oi, &objs->l_ost_oi);
2968 objs->l_ost_gen = cpu_to_le32(0);
2969 objs->l_ost_idx = cpu_to_le32(llr->llr_ost_idx);
2970 lfsck_buf_init(&ea_buf, lmm,
2971 lov_mds_md_size(le16_to_cpu(lmm->lmm_stripe_count),
2973 rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV,
2974 LU_XATTR_REPLACE, handle);
2976 GOTO(unlock, rc = (rc == 0 ? 1 : rc));
2979 dt_write_unlock(env, parent);
2983 dt_trans_stop(env, dev, handle);
2986 lfsck_ibits_unlock(&lh, LCK_EX);
2988 lfsck_object_put(env, child);
2990 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired multiple "
2991 "references for: parent "DFID", OST-index %u, stripe-index %u, "
2992 "owner %u/%u: rc = %d\n", lfsck_lfsck2name(lfsck), PFID(pfid),
2993 llr->llr_ost_idx, llr->llr_lov_idx, la->la_uid, la->la_gid, rc);
2998 /* If the MDT-object and the OST-object have different owner information,
2999 * then trust the MDT-object, because the normal chown/chgrp handle order
3000 * is from MDT to OST, and it is possible that some chown/chgrp operation
3001 * is partly done. */
3002 static int lfsck_layout_repair_owner(const struct lu_env *env,
3003 struct lfsck_component *com,
3004 struct dt_object *parent,
3005 struct lfsck_layout_req *llr,
3008 struct lfsck_thread_info *info = lfsck_env_info(env);
3009 struct lu_attr *tla = &info->lti_la;
3010 struct dt_object *child = llr->llr_child;
3011 struct dt_device *dev = lfsck_obj2dev(child);
3012 struct thandle *handle;
3016 handle = dt_trans_create(env, dev);
3018 GOTO(log, rc = PTR_ERR(handle));
3020 tla->la_uid = la->la_uid;
3021 tla->la_gid = la->la_gid;
3022 tla->la_valid = LA_UID | LA_GID;
3023 rc = dt_declare_attr_set(env, child, tla, handle);
3027 rc = dt_trans_start_local(env, dev, handle);
3031 /* Use the dt_object lock to serialize with destroy and attr_set. */
3032 dt_read_lock(env, parent, 0);
3033 if (unlikely(lfsck_is_dead_obj(parent)))
3034 GOTO(unlock, rc = 1);
3036 /* Get the latest parent's owner. */
3037 rc = dt_attr_get(env, parent, la);
3041 /* Some others chown/chgrp during the LFSCK, needs to do nothing. */
3042 if (unlikely(tla->la_uid != la->la_uid ||
3043 tla->la_gid != la->la_gid))
3044 GOTO(unlock, rc = 1);
3046 rc = dt_attr_set(env, child, tla, handle);
3051 dt_read_unlock(env, parent);
3054 rc = lfsck_layout_trans_stop(env, dev, handle, rc);
3057 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired inconsistent "
3058 "file owner for: parent "DFID", child "DFID", OST-index %u, "
3059 "stripe-index %u, owner %u/%u: rc = %d\n",
3060 lfsck_lfsck2name(com->lc_lfsck), PFID(lfsck_dto2fid(parent)),
3061 PFID(lfsck_dto2fid(child)), llr->llr_ost_idx, llr->llr_lov_idx,
3062 la->la_uid, la->la_gid, rc);
3067 /* Check whether the OST-object correctly back points to the
3068 * MDT-object (@parent) via the XATTR_NAME_FID xattr (@pfid). */
3069 static int lfsck_layout_check_parent(const struct lu_env *env,
3070 struct lfsck_component *com,
3071 struct lfsck_assistant_object *lso,
3072 const struct lu_fid *pfid,
3073 const struct lu_fid *cfid,
3074 const struct lu_attr *cla,
3075 struct lfsck_layout_req *llr,
3076 struct lu_buf *lov_ea, __u32 idx)
3078 struct lfsck_thread_info *info = lfsck_env_info(env);
3079 struct lu_buf *buf = &info->lti_big_buf;
3080 struct dt_object *tobj;
3081 struct lov_mds_md_v1 *lmm;
3082 struct lov_ost_data_v1 *objs;
3083 struct lustre_handle lh = { 0 };
3090 if (fid_is_zero(pfid)) {
3091 /* client never wrote. */
3092 if (cla->la_size == 0 && cla->la_blocks == 0) {
3093 struct lu_attr *pla = &lso->lso_attr;
3095 /* Someone may has changed the owner after the @pla
3096 * pre-loaded. It can be handled when repairs owner
3097 * inside lfsck_layout_repair_owner(). */
3098 if (unlikely(cla->la_uid != pla->la_uid ||
3099 cla->la_gid != pla->la_gid))
3100 RETURN(LLIT_INCONSISTENT_OWNER);
3105 RETURN(LLIT_UNMATCHED_PAIR);
3108 if (unlikely(!fid_is_sane(pfid)))
3109 RETURN(LLIT_UNMATCHED_PAIR);
3111 if (lu_fid_eq(pfid, &lso->lso_fid)) {
3112 if (llr->llr_lov_idx == idx)
3115 RETURN(LLIT_UNMATCHED_PAIR);
3118 tobj = lfsck_object_find_bottom(env, com->lc_lfsck, pfid);
3120 RETURN(PTR_ERR(tobj));
3122 if (dt_object_exists(tobj) == 0 ||
3123 lfsck_is_dead_obj(tobj))
3124 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3126 if (!S_ISREG(lfsck_object_type(tobj)))
3127 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3129 /* Load the tobj's layout EA, in spite of it is a local MDT-object or
3130 * remote one on another MDT. Then check whether the given OST-object
3131 * is in such layout. If yes, it is multiple referenced, otherwise it
3132 * is unmatched referenced case. */
3133 rc = lfsck_layout_get_lovea(env, tobj, buf);
3134 if (rc == 0 || rc == -ENOENT)
3135 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3141 magic = le32_to_cpu(lmm->lmm_magic);
3142 if (magic == LOV_MAGIC_V1) {
3143 objs = &lmm->lmm_objects[0];
3145 LASSERT(magic == LOV_MAGIC_V3);
3146 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
3149 count = le16_to_cpu(lmm->lmm_stripe_count);
3150 for (i = 0; i < count; i++, objs++) {
3151 struct lu_fid *tfid = &info->lti_fid2;
3152 struct ost_id *oi = &info->lti_oi;
3155 if (lovea_slot_is_dummy(objs))
3158 ostid_le_to_cpu(&objs->l_ost_oi, oi);
3159 idx2 = le32_to_cpu(objs->l_ost_idx);
3160 rc = ostid_to_fid(tfid, oi, idx2);
3162 CDEBUG(D_LFSCK, "%s: the parent "DFID" contains "
3163 "invalid layout EA at the slot %d, index %u\n",
3164 lfsck_lfsck2name(com->lc_lfsck),
3165 PFID(pfid), i, idx2);
3167 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3170 if (lu_fid_eq(cfid, tfid)) {
3171 rc = lfsck_ibits_lock(env, com->lc_lfsck, tobj, &lh,
3172 MDS_INODELOCK_UPDATE |
3173 MDS_INODELOCK_LAYOUT |
3174 MDS_INODELOCK_XATTR,
3179 dt_read_lock(env, tobj, 0);
3181 /* For local MDT-object, re-check existence
3182 * after taken the lock. */
3183 if (!dt_object_remote(tobj)) {
3184 if (dt_object_exists(tobj) == 0 ||
3185 lfsck_is_dead_obj(tobj)) {
3186 rc = LLIT_UNMATCHED_PAIR;
3189 rc = LLIT_MULTIPLE_REFERENCED;
3195 /* For migration case, the new MDT-object and old
3196 * MDT-object may reference the same OST-object at
3197 * some migration internal time.
3199 * For remote MDT-object, the local MDT may not know
3200 * whether it has been removed or not. Try checking
3201 * for a non-existent xattr to check if this object
3202 * has been been removed or not. */
3203 rc = dt_xattr_get(env, tobj, &LU_BUF_NULL,
3205 if (unlikely(rc == -ENOENT || rc >= 0)) {
3206 rc = LLIT_UNMATCHED_PAIR;
3207 } else if (rc == -ENODATA) {
3209 rc = LLIT_MULTIPLE_REFERENCED;
3216 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3219 if (lustre_handle_is_used(&lh)) {
3220 dt_read_unlock(env, tobj);
3221 lfsck_ibits_unlock(&lh, LCK_EX);
3225 lfsck_object_put(env, tobj);
3230 static int lfsck_layout_assistant_handler_p1(const struct lu_env *env,
3231 struct lfsck_component *com,
3232 struct lfsck_assistant_req *lar)
3234 struct lfsck_layout_req *llr =
3235 container_of0(lar, struct lfsck_layout_req, llr_lar);
3236 struct lfsck_assistant_object *lso = lar->lar_parent;
3237 struct lfsck_layout *lo = com->lc_file_ram;
3238 struct lfsck_thread_info *info = lfsck_env_info(env);
3239 struct filter_fid_old *pea = &info->lti_old_pfid;
3240 struct lu_fid *pfid = &info->lti_fid;
3241 struct lu_buf buf = { NULL };
3242 struct dt_object *parent = NULL;
3243 struct dt_object *child = llr->llr_child;
3244 struct lu_attr *pla = &lso->lso_attr;
3245 struct lu_attr *cla = &info->lti_la;
3246 struct lfsck_instance *lfsck = com->lc_lfsck;
3247 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
3248 enum lfsck_layout_inconsistency_type type = LLIT_NONE;
3256 rc = dt_attr_get(env, child, cla);
3257 if (rc == -ENOENT) {
3258 parent = lfsck_assistant_object_load(env, lfsck, lso);
3259 if (IS_ERR(parent)) {
3260 rc = PTR_ERR(parent);
3262 RETURN(rc == -ENOENT ? 0 : rc);
3265 type = LLIT_DANGLING;
3272 lfsck_buf_init(&buf, pea, sizeof(struct filter_fid_old));
3273 rc = dt_xattr_get(env, child, &buf, XATTR_NAME_FID);
3274 if (unlikely(rc >= 0 && rc != sizeof(struct filter_fid_old) &&
3275 rc != sizeof(struct filter_fid))) {
3276 type = LLIT_UNMATCHED_PAIR;
3280 if (rc < 0 && rc != -ENODATA)
3283 if (rc == -ENODATA) {
3286 fid_le_to_cpu(pfid, &pea->ff_parent);
3287 /* Currently, the filter_fid::ff_parent::f_ver is not the
3288 * real parent MDT-object's FID::f_ver, instead it is the
3289 * OST-object index in its parent MDT-object's layout EA. */
3290 idx = pfid->f_stripe_idx;
3294 rc = lfsck_layout_check_parent(env, com, lso, pfid,
3295 lu_object_fid(&child->do_lu),
3296 cla, llr, &buf, idx);
3305 /* Someone may has changed the owner after the parent attr pre-loaded.
3306 * It can be handled later inside the lfsck_layout_repair_owner(). */
3307 if (unlikely(cla->la_uid != pla->la_uid ||
3308 cla->la_gid != pla->la_gid)) {
3309 type = LLIT_INCONSISTENT_OWNER;
3314 if (type == LLIT_NONE)
3317 if (bk->lb_param & LPF_DRYRUN)
3320 if (parent == NULL) {
3321 parent = lfsck_assistant_object_load(env, lfsck, lso);
3322 if (IS_ERR(parent)) {
3323 rc = PTR_ERR(parent);
3334 rc = lfsck_layout_repair_dangling(env, com, parent, llr, pla);
3336 case LLIT_UNMATCHED_PAIR:
3337 rc = lfsck_layout_repair_unmatched_pair(env, com, parent,
3340 case LLIT_MULTIPLE_REFERENCED:
3341 rc = lfsck_layout_repair_multiple_references(env, com, parent,
3344 case LLIT_INCONSISTENT_OWNER:
3345 rc = lfsck_layout_repair_owner(env, com, parent, llr, pla);
3355 down_write(&com->lc_sem);
3357 struct lfsck_assistant_data *lad = com->lc_data;
3359 if (unlikely(lad->lad_exit)) {
3361 } else if (rc == -ENOTCONN || rc == -ESHUTDOWN ||
3362 rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
3363 rc == -EHOSTUNREACH) {
3364 /* If cannot touch the target server,
3365 * mark the LFSCK as INCOMPLETE. */
3366 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant fail to "
3367 "talk with OST %x: rc = %d\n",
3368 lfsck_lfsck2name(lfsck), llr->llr_ost_idx, rc);
3369 lfsck_lad_set_bitmap(env, com, llr->llr_ost_idx);
3370 lo->ll_objs_skipped++;
3373 lfsck_layout_record_failure(env, lfsck, lo);
3375 } else if (rc > 0) {
3376 LASSERTF(type > LLIT_NONE && type <= LLIT_MAX,
3377 "unknown type = %d\n", type);
3379 lo->ll_objs_repaired[type - 1]++;
3380 if (bk->lb_param & LPF_DRYRUN &&
3381 unlikely(lo->ll_pos_first_inconsistent == 0))
3382 lo->ll_pos_first_inconsistent =
3383 lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
3386 up_write(&com->lc_sem);
3388 if (parent != NULL && !IS_ERR(parent))
3389 lfsck_object_put(env, parent);
3394 static int lfsck_layout_assistant_handler_p2(const struct lu_env *env,
3395 struct lfsck_component *com)
3397 struct lfsck_assistant_data *lad = com->lc_data;
3398 struct lfsck_instance *lfsck = com->lc_lfsck;
3399 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
3400 struct lfsck_tgt_descs *ltds = &lfsck->li_ost_descs;
3401 struct lfsck_tgt_desc *ltd;
3405 CDEBUG(D_LFSCK, "%s: layout LFSCK phase2 scan start\n",
3406 lfsck_lfsck2name(lfsck));
3408 spin_lock(<ds->ltd_lock);
3409 while (!list_empty(&lad->lad_ost_phase2_list)) {
3410 ltd = list_entry(lad->lad_ost_phase2_list.next,
3411 struct lfsck_tgt_desc,
3412 ltd_layout_phase_list);
3413 list_del_init(<d->ltd_layout_phase_list);
3414 if (bk->lb_param & LPF_ALL_TGT) {
3415 spin_unlock(<ds->ltd_lock);
3416 rc = lfsck_layout_scan_orphan(env, com, ltd);
3417 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
3420 if (unlikely(lad->lad_exit ||
3421 !thread_is_running(&lfsck->li_thread)))
3423 spin_lock(<ds->ltd_lock);
3427 if (list_empty(&lad->lad_ost_phase1_list))
3431 spin_unlock(<ds->ltd_lock);
3433 CDEBUG(D_LFSCK, "%s: layout LFSCK phase2 scan stop: rc = %d\n",
3434 lfsck_lfsck2name(lfsck), rc);
3440 lfsck_layout_slave_async_interpret(const struct lu_env *env,
3441 struct ptlrpc_request *req,
3444 struct lfsck_layout_slave_async_args *llsaa = args;
3445 struct obd_export *exp = llsaa->llsaa_exp;
3446 struct lfsck_component *com = llsaa->llsaa_com;
3447 struct lfsck_layout_slave_target *llst = llsaa->llsaa_llst;
3448 struct lfsck_layout_slave_data *llsd = com->lc_data;
3449 struct lfsck_reply *lr = NULL;
3453 /* It is quite probably caused by target crash,
3454 * to make the LFSCK can go ahead, assume that
3455 * the target finished the LFSCK prcoessing. */
3458 lr = req_capsule_server_get(&req->rq_pill, &RMF_LFSCK_REPLY);
3459 if (lr->lr_status != LS_SCANNING_PHASE1 &&
3460 lr->lr_status != LS_SCANNING_PHASE2)
3465 CDEBUG(D_LFSCK, "%s: layout LFSCK slave gets the MDT %x "
3466 "status %d\n", lfsck_lfsck2name(com->lc_lfsck),
3467 llst->llst_index, lr != NULL ? lr->lr_status : rc);
3469 lfsck_layout_llst_del(llsd, llst);
3472 lfsck_layout_llst_put(llst);
3473 lfsck_component_put(env, com);
3474 class_export_put(exp);
3479 static int lfsck_layout_async_query(const struct lu_env *env,
3480 struct lfsck_component *com,
3481 struct obd_export *exp,
3482 struct lfsck_layout_slave_target *llst,
3483 struct lfsck_request *lr,
3484 struct ptlrpc_request_set *set)
3486 struct lfsck_layout_slave_async_args *llsaa;
3487 struct ptlrpc_request *req;
3488 struct lfsck_request *tmp;
3492 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_QUERY);
3496 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_QUERY);
3498 ptlrpc_request_free(req);
3502 tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3504 ptlrpc_request_set_replen(req);
3506 llsaa = ptlrpc_req_async_args(req);
3507 llsaa->llsaa_exp = exp;
3508 llsaa->llsaa_com = lfsck_component_get(com);
3509 llsaa->llsaa_llst = llst;
3510 req->rq_interpret_reply = lfsck_layout_slave_async_interpret;
3511 ptlrpc_set_add_req(set, req);
3516 static int lfsck_layout_async_notify(const struct lu_env *env,
3517 struct obd_export *exp,
3518 struct lfsck_request *lr,
3519 struct ptlrpc_request_set *set)
3521 struct ptlrpc_request *req;
3522 struct lfsck_request *tmp;
3526 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
3530 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
3532 ptlrpc_request_free(req);
3536 tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3538 ptlrpc_request_set_replen(req);
3539 ptlrpc_set_add_req(set, req);
3545 lfsck_layout_slave_query_master(const struct lu_env *env,
3546 struct lfsck_component *com)
3548 struct lfsck_request *lr = &lfsck_env_info(env)->lti_lr;
3549 struct lfsck_instance *lfsck = com->lc_lfsck;
3550 struct lfsck_layout_slave_data *llsd = com->lc_data;
3551 struct lfsck_layout_slave_target *llst;
3552 struct obd_export *exp;
3553 struct ptlrpc_request_set *set;
3558 set = ptlrpc_prep_set();
3560 GOTO(log, rc = -ENOMEM);
3562 memset(lr, 0, sizeof(*lr));
3563 lr->lr_event = LE_QUERY;
3564 lr->lr_active = LFSCK_TYPE_LAYOUT;
3566 llsd->llsd_touch_gen++;
3567 spin_lock(&llsd->llsd_lock);
3568 while (!list_empty(&llsd->llsd_master_list)) {
3569 llst = list_entry(llsd->llsd_master_list.next,
3570 struct lfsck_layout_slave_target,
3572 if (llst->llst_gen == llsd->llsd_touch_gen)
3575 llst->llst_gen = llsd->llsd_touch_gen;
3576 list_move_tail(&llst->llst_list,
3577 &llsd->llsd_master_list);
3578 atomic_inc(&llst->llst_ref);
3579 spin_unlock(&llsd->llsd_lock);
3581 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3584 lfsck_layout_llst_del(llsd, llst);
3585 lfsck_layout_llst_put(llst);
3586 spin_lock(&llsd->llsd_lock);
3590 rc = lfsck_layout_async_query(env, com, exp, llst, lr, set);
3592 CDEBUG(D_LFSCK, "%s: layout LFSCK slave fail to "
3593 "query %s for layout: rc = %d\n",
3594 lfsck_lfsck2name(lfsck),
3595 exp->exp_obd->obd_name, rc);
3598 lfsck_layout_llst_put(llst);
3599 class_export_put(exp);
3601 spin_lock(&llsd->llsd_lock);
3603 spin_unlock(&llsd->llsd_lock);
3605 rc = ptlrpc_set_wait(set);
3606 ptlrpc_set_destroy(set);
3608 GOTO(log, rc = (rc1 != 0 ? rc1 : rc));
3611 CDEBUG(D_LFSCK, "%s: layout LFSCK slave queries master: rc = %d\n",
3612 lfsck_lfsck2name(com->lc_lfsck), rc);
3618 lfsck_layout_slave_notify_master(const struct lu_env *env,
3619 struct lfsck_component *com,
3620 enum lfsck_events event, int result)
3622 struct lfsck_layout *lo = com->lc_file_ram;
3623 struct lfsck_instance *lfsck = com->lc_lfsck;
3624 struct lfsck_layout_slave_data *llsd = com->lc_data;
3625 struct lfsck_request *lr = &lfsck_env_info(env)->lti_lr;
3626 struct lfsck_layout_slave_target *llst;
3627 struct obd_export *exp;
3628 struct ptlrpc_request_set *set;
3632 CDEBUG(D_LFSCK, "%s: layout LFSCK slave notifies master\n",
3633 lfsck_lfsck2name(com->lc_lfsck));
3635 set = ptlrpc_prep_set();
3639 memset(lr, 0, sizeof(*lr));
3640 lr->lr_event = event;
3641 lr->lr_flags = LEF_FROM_OST;
3642 lr->lr_status = result;
3643 lr->lr_index = lfsck_dev_idx(lfsck);
3644 lr->lr_active = LFSCK_TYPE_LAYOUT;
3645 lr->lr_flags2 = lo->ll_flags;
3646 llsd->llsd_touch_gen++;
3647 spin_lock(&llsd->llsd_lock);
3648 while (!list_empty(&llsd->llsd_master_list)) {
3649 llst = list_entry(llsd->llsd_master_list.next,
3650 struct lfsck_layout_slave_target,
3652 if (llst->llst_gen == llsd->llsd_touch_gen)
3655 llst->llst_gen = llsd->llsd_touch_gen;
3656 list_move_tail(&llst->llst_list,
3657 &llsd->llsd_master_list);
3658 atomic_inc(&llst->llst_ref);
3659 spin_unlock(&llsd->llsd_lock);
3661 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3664 lfsck_layout_llst_del(llsd, llst);
3665 lfsck_layout_llst_put(llst);
3666 spin_lock(&llsd->llsd_lock);
3670 rc = lfsck_layout_async_notify(env, exp, lr, set);
3672 CDEBUG(D_LFSCK, "%s: layout LFSCK slave fail to "
3673 "notify %s for layout: rc = %d\n",
3674 lfsck_lfsck2name(lfsck),
3675 exp->exp_obd->obd_name, rc);
3677 lfsck_layout_llst_put(llst);
3678 class_export_put(exp);
3679 spin_lock(&llsd->llsd_lock);
3681 spin_unlock(&llsd->llsd_lock);
3683 ptlrpc_set_wait(set);
3684 ptlrpc_set_destroy(set);
3690 * \ret -ENODATA: unrecognized stripe
3691 * \ret = 0 : recognized stripe
3692 * \ret < 0 : other failures
3694 static int lfsck_layout_master_check_pairs(const struct lu_env *env,
3695 struct lfsck_component *com,
3696 struct lu_fid *cfid,
3697 struct lu_fid *pfid)
3699 struct lfsck_thread_info *info = lfsck_env_info(env);
3700 struct lu_buf *buf = &info->lti_big_buf;
3701 struct ost_id *oi = &info->lti_oi;
3702 struct dt_object *obj;
3703 struct lov_mds_md_v1 *lmm;
3704 struct lov_ost_data_v1 *objs;
3705 __u32 idx = pfid->f_stripe_idx;
3713 obj = lfsck_object_find_bottom(env, com->lc_lfsck, pfid);
3715 RETURN(PTR_ERR(obj));
3717 dt_read_lock(env, obj, 0);
3718 if (unlikely(dt_object_exists(obj) == 0 ||
3719 lfsck_is_dead_obj(obj)))
3720 GOTO(unlock, rc = -ENOENT);
3722 if (!S_ISREG(lfsck_object_type(obj)))
3723 GOTO(unlock, rc = -ENODATA);
3725 rc = lfsck_layout_get_lovea(env, obj, buf);
3730 GOTO(unlock, rc = -ENODATA);
3733 rc = lfsck_layout_verify_header(lmm);
3737 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
3738 * been verified in lfsck_layout_verify_header() already. If some
3739 * new magic introduced in the future, then layout LFSCK needs to
3740 * be updated also. */
3741 magic = le32_to_cpu(lmm->lmm_magic);
3742 if (magic == LOV_MAGIC_V1) {
3743 objs = &lmm->lmm_objects[0];
3745 LASSERT(magic == LOV_MAGIC_V3);
3746 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
3749 fid_to_ostid(cfid, oi);
3750 count = le16_to_cpu(lmm->lmm_stripe_count);
3751 for (i = 0; i < count; i++, objs++) {
3754 ostid_le_to_cpu(&objs->l_ost_oi, &oi2);
3755 if (memcmp(oi, &oi2, sizeof(*oi)) == 0)
3756 GOTO(unlock, rc = (i != idx ? -ENODATA : 0));
3759 GOTO(unlock, rc = -ENODATA);
3762 dt_read_unlock(env, obj);
3763 lfsck_object_put(env, obj);
3769 * The LFSCK-on-OST will ask the LFSCK-on-MDT to check whether the given
3770 * MDT-object/OST-object pairs match or not to aviod transfer MDT-object
3771 * layout EA from MDT to OST. On one hand, the OST no need to understand
3772 * the layout EA structure; on the other hand, it may cause trouble when
3773 * transfer large layout EA from MDT to OST via normal OUT RPC.
3775 * \ret > 0: unrecognized stripe
3776 * \ret = 0: recognized stripe
3777 * \ret < 0: other failures
3779 static int lfsck_layout_slave_check_pairs(const struct lu_env *env,
3780 struct lfsck_component *com,
3781 struct lu_fid *cfid,
3782 struct lu_fid *pfid)
3784 struct lfsck_instance *lfsck = com->lc_lfsck;
3785 struct obd_device *obd = lfsck->li_obd;
3786 struct seq_server_site *ss = lfsck_dev_site(lfsck);
3787 struct obd_export *exp = NULL;
3788 struct ptlrpc_request *req = NULL;
3789 struct lfsck_request *lr;
3790 struct lu_seq_range *range = &lfsck_env_info(env)->lti_range;
3794 if (unlikely(fid_is_idif(pfid)))
3797 fld_range_set_any(range);
3798 rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(pfid), range);
3800 RETURN(rc == -ENOENT ? 1 : rc);
3802 if (unlikely(!fld_range_is_mdt(range)))
3805 exp = lustre_find_lwp_by_index(obd->obd_name, range->lsr_index);
3806 if (unlikely(exp == NULL))
3809 if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
3810 GOTO(out, rc = -EOPNOTSUPP);
3812 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
3814 GOTO(out, rc = -ENOMEM);
3816 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
3818 ptlrpc_request_free(req);
3823 lr = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3824 memset(lr, 0, sizeof(*lr));
3825 lr->lr_event = LE_PAIRS_VERIFY;
3826 lr->lr_active = LFSCK_TYPE_LAYOUT;
3827 lr->lr_fid = *cfid; /* OST-object itself FID. */
3828 lr->lr_fid2 = *pfid; /* The claimed parent FID. */
3830 ptlrpc_request_set_replen(req);
3831 rc = ptlrpc_queue_wait(req);
3832 ptlrpc_req_finished(req);
3834 if (rc == -ENOENT || rc == -ENODATA)
3841 class_export_put(exp);
3846 static int lfsck_layout_slave_repair_pfid(const struct lu_env *env,
3847 struct lfsck_component *com,
3848 struct lfsck_request *lr)
3850 struct dt_object *obj;
3854 obj = lfsck_object_find_bottom(env, com->lc_lfsck, &lr->lr_fid);
3856 GOTO(log, rc = PTR_ERR(obj));
3858 dt_write_lock(env, obj, 0);
3859 if (unlikely(dt_object_exists(obj) == 0 ||
3860 lfsck_is_dead_obj(obj)))
3861 GOTO(unlock, rc = 0);
3863 rc = __lfsck_layout_update_pfid(env, obj, &lr->lr_fid2,
3869 dt_write_unlock(env, obj);
3870 lfsck_object_put(env, obj);
3873 CDEBUG(D_LFSCK, "%s: layout LFSCK slave repaired pfid for "DFID
3874 ", parent "DFID": rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
3875 PFID(&lr->lr_fid), PFID(&lr->lr_fid2), rc);
3882 static void lfsck_layout_slave_quit(const struct lu_env *env,
3883 struct lfsck_component *com);
3885 static int lfsck_layout_reset(const struct lu_env *env,
3886 struct lfsck_component *com, bool init)
3888 struct lfsck_layout *lo = com->lc_file_ram;
3891 down_write(&com->lc_sem);
3893 memset(lo, 0, com->lc_file_size);
3895 __u32 count = lo->ll_success_count;
3896 __u64 last_time = lo->ll_time_last_complete;
3898 memset(lo, 0, com->lc_file_size);
3899 lo->ll_success_count = count;
3900 lo->ll_time_last_complete = last_time;
3903 lo->ll_magic = LFSCK_LAYOUT_MAGIC;
3904 lo->ll_status = LS_INIT;
3906 if (com->lc_lfsck->li_master) {
3907 struct lfsck_assistant_data *lad = com->lc_data;
3909 lad->lad_incomplete = 0;
3910 CFS_RESET_BITMAP(lad->lad_bitmap);
3913 rc = lfsck_layout_store(env, com);
3914 up_write(&com->lc_sem);
3916 CDEBUG(D_LFSCK, "%s: layout LFSCK reset: rc = %d\n",
3917 lfsck_lfsck2name(com->lc_lfsck), rc);
3922 static void lfsck_layout_fail(const struct lu_env *env,
3923 struct lfsck_component *com, bool new_checked)
3925 struct lfsck_layout *lo = com->lc_file_ram;
3927 down_write(&com->lc_sem);
3929 com->lc_new_checked++;
3930 lfsck_layout_record_failure(env, com->lc_lfsck, lo);
3931 up_write(&com->lc_sem);
3934 static int lfsck_layout_master_checkpoint(const struct lu_env *env,
3935 struct lfsck_component *com, bool init)
3937 struct lfsck_instance *lfsck = com->lc_lfsck;
3938 struct lfsck_layout *lo = com->lc_file_ram;
3942 rc = lfsck_checkpoint_generic(env, com);
3944 return rc > 0 ? 0 : rc;
3947 down_write(&com->lc_sem);
3949 lo->ll_pos_latest_start =
3950 lfsck->li_pos_checkpoint.lp_oit_cookie;
3952 lo->ll_pos_last_checkpoint =
3953 lfsck->li_pos_checkpoint.lp_oit_cookie;
3954 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3955 HALF_SEC - lfsck->li_time_last_checkpoint);
3956 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3957 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3958 com->lc_new_checked = 0;
3961 rc = lfsck_layout_store(env, com);
3962 up_write(&com->lc_sem);
3964 CDEBUG(D_LFSCK, "%s: layout LFSCK master checkpoint at the pos ["
3965 LPU64"]: rc = %d\n", lfsck_lfsck2name(lfsck),
3966 lfsck->li_pos_current.lp_oit_cookie, rc);
3971 static int lfsck_layout_slave_checkpoint(const struct lu_env *env,
3972 struct lfsck_component *com, bool init)
3974 struct lfsck_instance *lfsck = com->lc_lfsck;
3975 struct lfsck_layout *lo = com->lc_file_ram;
3978 if (com->lc_new_checked == 0 && !init)
3981 down_write(&com->lc_sem);
3983 lo->ll_pos_latest_start =
3984 lfsck->li_pos_checkpoint.lp_oit_cookie;
3986 lo->ll_pos_last_checkpoint =
3987 lfsck->li_pos_checkpoint.lp_oit_cookie;
3988 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3989 HALF_SEC - lfsck->li_time_last_checkpoint);
3990 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3991 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3992 com->lc_new_checked = 0;
3995 rc = lfsck_layout_store(env, com);
3996 up_write(&com->lc_sem);
3998 CDEBUG(D_LFSCK, "%s: layout LFSCK slave checkpoint at the pos ["
3999 LPU64"]: rc = %d\n", lfsck_lfsck2name(lfsck),
4000 lfsck->li_pos_current.lp_oit_cookie, rc);
4005 static int lfsck_layout_prep(const struct lu_env *env,
4006 struct lfsck_component *com,
4007 struct lfsck_start *start)
4009 struct lfsck_instance *lfsck = com->lc_lfsck;
4010 struct lfsck_layout *lo = com->lc_file_ram;
4011 struct lfsck_position *pos = &com->lc_pos_start;
4013 fid_zero(&pos->lp_dir_parent);
4014 pos->lp_dir_cookie = 0;
4015 if (lo->ll_status == LS_COMPLETED ||
4016 lo->ll_status == LS_PARTIAL ||
4017 /* To handle orphan, must scan from the beginning. */
4018 (start != NULL && start->ls_flags & LPF_OST_ORPHAN)) {
4021 rc = lfsck_layout_reset(env, com, false);
4023 rc = lfsck_set_param(env, lfsck, start, true);
4026 CDEBUG(D_LFSCK, "%s: layout LFSCK prep failed: "
4027 "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
4033 down_write(&com->lc_sem);
4034 lo->ll_time_latest_start = cfs_time_current_sec();
4035 spin_lock(&lfsck->li_lock);
4036 if (lo->ll_flags & LF_SCANNED_ONCE) {
4037 if (!lfsck->li_drop_dryrun ||
4038 lo->ll_pos_first_inconsistent == 0) {
4039 lo->ll_status = LS_SCANNING_PHASE2;
4040 list_move_tail(&com->lc_link,
4041 &lfsck->li_list_double_scan);
4042 pos->lp_oit_cookie = 0;
4046 lo->ll_status = LS_SCANNING_PHASE1;
4047 lo->ll_run_time_phase1 = 0;
4048 lo->ll_run_time_phase2 = 0;
4049 lo->ll_objs_checked_phase1 = 0;
4050 lo->ll_objs_checked_phase2 = 0;
4051 lo->ll_objs_failed_phase1 = 0;
4052 lo->ll_objs_failed_phase2 = 0;
4053 for (i = 0; i < LLIT_MAX; i++)
4054 lo->ll_objs_repaired[i] = 0;
4056 pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
4057 fid_zero(&com->lc_fid_latest_scanned_phase2);
4060 lo->ll_status = LS_SCANNING_PHASE1;
4061 if (!lfsck->li_drop_dryrun ||
4062 lo->ll_pos_first_inconsistent == 0)
4063 pos->lp_oit_cookie = lo->ll_pos_last_checkpoint + 1;
4065 pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
4067 spin_unlock(&lfsck->li_lock);
4068 up_write(&com->lc_sem);
4073 static int lfsck_layout_slave_prep(const struct lu_env *env,
4074 struct lfsck_component *com,
4075 struct lfsck_start_param *lsp)
4077 struct lfsck_layout_slave_data *llsd = com->lc_data;
4078 struct lfsck_instance *lfsck = com->lc_lfsck;
4079 struct lfsck_layout *lo = com->lc_file_ram;
4080 struct lfsck_start *start = lsp->lsp_start;
4083 rc = lfsck_layout_prep(env, com, start);
4087 if (lo->ll_flags & LF_CRASHED_LASTID &&
4088 list_empty(&llsd->llsd_master_list)) {
4089 LASSERT(lfsck->li_out_notify != NULL);
4091 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4092 LE_LASTID_REBUILDING);
4095 if (!lsp->lsp_index_valid)
4098 rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
4099 if (rc == 0 && start != NULL && start->ls_flags & LPF_OST_ORPHAN) {
4100 LASSERT(!llsd->llsd_rbtree_valid);
4102 write_lock(&llsd->llsd_rb_lock);
4103 rc = lfsck_rbtree_setup(env, com);
4104 write_unlock(&llsd->llsd_rb_lock);
4107 CDEBUG(D_LFSCK, "%s: layout LFSCK slave prep done, start pos ["
4108 LPU64"]\n", lfsck_lfsck2name(lfsck),
4109 com->lc_pos_start.lp_oit_cookie);
4114 static int lfsck_layout_master_prep(const struct lu_env *env,
4115 struct lfsck_component *com,
4116 struct lfsck_start_param *lsp)
4121 rc = lfsck_layout_load_bitmap(env, com);
4123 rc = lfsck_layout_reset(env, com, false);
4125 rc = lfsck_set_param(env, com->lc_lfsck,
4126 lsp->lsp_start, true);
4132 rc = lfsck_layout_prep(env, com, lsp->lsp_start);
4136 rc = lfsck_start_assistant(env, com, lsp);
4141 CDEBUG(D_LFSCK, "%s: layout LFSCK master prep done, start pos ["
4142 LPU64"]\n", lfsck_lfsck2name(com->lc_lfsck),
4143 com->lc_pos_start.lp_oit_cookie);
4148 /* Pre-fetch the attribute for each stripe in the given layout EA. */
4149 static int lfsck_layout_scan_stripes(const struct lu_env *env,
4150 struct lfsck_component *com,
4151 struct dt_object *parent,
4152 struct lov_mds_md_v1 *lmm)
4154 struct lfsck_thread_info *info = lfsck_env_info(env);
4155 struct lfsck_instance *lfsck = com->lc_lfsck;
4156 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
4157 struct lfsck_layout *lo = com->lc_file_ram;
4158 struct lfsck_assistant_data *lad = com->lc_data;
4159 struct lfsck_assistant_object *lso = NULL;
4160 struct lov_ost_data_v1 *objs;
4161 struct lfsck_tgt_descs *ltds = &lfsck->li_ost_descs;
4162 struct ptlrpc_thread *mthread = &lfsck->li_thread;
4163 struct ptlrpc_thread *athread = &lad->lad_thread;
4164 struct l_wait_info lwi = { 0 };
4172 lfsck_buf_init(&buf, &info->lti_old_pfid,
4173 sizeof(struct filter_fid_old));
4174 count = le16_to_cpu(lmm->lmm_stripe_count);
4175 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
4176 * been verified in lfsck_layout_verify_header() already. If some
4177 * new magic introduced in the future, then layout LFSCK needs to
4178 * be updated also. */
4179 magic = le32_to_cpu(lmm->lmm_magic);
4180 if (magic == LOV_MAGIC_V1) {
4181 objs = &lmm->lmm_objects[0];
4183 LASSERT(magic == LOV_MAGIC_V3);
4184 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
4187 for (i = 0; i < count; i++, objs++) {
4188 struct lu_fid *fid = &info->lti_fid;
4189 struct ost_id *oi = &info->lti_oi;
4190 struct lfsck_layout_req *llr;
4191 struct lfsck_tgt_desc *tgt = NULL;
4192 struct dt_object *cobj = NULL;
4194 bool wakeup = false;
4196 if (unlikely(lovea_slot_is_dummy(objs)))
4199 l_wait_event(mthread->t_ctl_waitq,
4200 lad->lad_prefetched < bk->lb_async_windows ||
4201 !thread_is_running(mthread) ||
4202 thread_is_stopped(athread),
4205 if (unlikely(!thread_is_running(mthread)) ||
4206 thread_is_stopped(athread))
4209 if (unlikely(lfsck_is_dead_obj(parent)))
4212 ostid_le_to_cpu(&objs->l_ost_oi, oi);
4213 index = le32_to_cpu(objs->l_ost_idx);
4214 rc = ostid_to_fid(fid, oi, index);
4216 CDEBUG(D_LFSCK, "%s: get invalid layout EA for "DFID
4217 ": "DOSTID", idx:%u\n", lfsck_lfsck2name(lfsck),
4218 PFID(lfsck_dto2fid(parent)), POSTID(oi), index);
4222 tgt = lfsck_tgt_get(ltds, index);
4223 if (unlikely(tgt == NULL)) {
4224 CDEBUG(D_LFSCK, "%s: cannot talk with OST %x which "
4225 "did not join the layout LFSCK\n",
4226 lfsck_lfsck2name(lfsck), index);
4227 lfsck_lad_set_bitmap(env, com, index);
4231 /* There is potential deadlock race condition between object
4232 * destroy and layout LFSCK. Consider the following scenario:
4234 * 1) The LFSCK thread obtained the parent object firstly, at
4235 * that time, the parent object has not been destroyed yet.
4237 * 2) One RPC service thread destroyed the parent and all its
4238 * children objects. Because the LFSCK is referencing the
4239 * parent object, then the parent object will be marked as
4240 * dying in RAM. On the other hand, the parent object is
4241 * referencing all its children objects, then all children
4242 * objects will be marked as dying in RAM also.
4244 * 3) The LFSCK thread tries to find some child object with
4245 * the parent object referenced. Then it will find that the
4246 * child object is dying. According to the object visibility
4247 * rules: the object with dying flag cannot be returned to
4248 * others. So the LFSCK thread has to wait until the dying
4249 * object has been purged from RAM, then it can allocate a
4250 * new object (with the same FID) in RAM. Unfortunately, the
4251 * LFSCK thread itself is referencing the parent object, and
4252 * cause the parent object cannot be purged, then cause the
4253 * child object cannot be purged also. So the LFSCK thread
4254 * will fall into deadlock.
4256 * We introduce non-blocked version lu_object_find() to allow
4257 * the LFSCK thread to return failure immediately (instead of
4258 * wait) when it finds dying (child) object, then the LFSCK
4259 * thread can check whether the parent object is dying or not.
4260 * So avoid above deadlock. LU-5395 */
4261 cobj = lfsck_object_find_by_dev_nowait(env, tgt->ltd_tgt, fid);
4263 if (lfsck_is_dead_obj(parent)) {
4273 rc = dt_declare_attr_get(env, cobj);
4277 rc = dt_declare_xattr_get(env, cobj, &buf, XATTR_NAME_FID);
4282 struct lu_attr *attr = &info->lti_la;
4284 rc = dt_attr_get(env, parent, attr);
4290 lso = lfsck_assistant_object_init(env,
4291 lfsck_dto2fid(parent), attr,
4292 lfsck->li_pos_current.lp_oit_cookie, false);
4301 llr = lfsck_layout_assistant_req_init(lso, cobj, index, i);
4308 spin_lock(&lad->lad_lock);
4309 if (lad->lad_assistant_status < 0) {
4310 spin_unlock(&lad->lad_lock);
4311 lfsck_layout_assistant_req_fini(env, &llr->llr_lar);
4313 RETURN(lad->lad_assistant_status);
4316 list_add_tail(&llr->llr_lar.lar_list, &lad->lad_req_list);
4317 if (lad->lad_prefetched == 0)
4320 lad->lad_prefetched++;
4321 spin_unlock(&lad->lad_lock);
4323 wake_up_all(&athread->t_ctl_waitq);
4326 down_write(&com->lc_sem);
4327 com->lc_new_checked++;
4329 lfsck_layout_record_failure(env, lfsck, lo);
4330 up_write(&com->lc_sem);
4332 if (cobj != NULL && !IS_ERR(cobj))
4333 lfsck_object_put(env, cobj);
4335 if (likely(tgt != NULL))
4338 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
4346 lfsck_assistant_object_put(env, lso);
4351 /* For the given object, read its layout EA locally. For each stripe, pre-fetch
4352 * the OST-object's attribute and generate an structure lfsck_layout_req on the
4353 * list ::lad_req_list.
4355 * For each request on above list, the lfsck_layout_assistant thread compares
4356 * the OST side attribute with local attribute, if inconsistent, then repair it.
4358 * All above processing is async mode with pipeline. */
4359 static int lfsck_layout_master_exec_oit(const struct lu_env *env,
4360 struct lfsck_component *com,
4361 struct dt_object *obj)
4363 struct lfsck_thread_info *info = lfsck_env_info(env);
4364 struct ost_id *oi = &info->lti_oi;
4365 struct lfsck_layout *lo = com->lc_file_ram;
4366 struct lfsck_assistant_data *lad = com->lc_data;
4367 struct lfsck_instance *lfsck = com->lc_lfsck;
4368 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
4369 struct thandle *handle = NULL;
4370 struct lu_buf *buf = &info->lti_big_buf;
4371 struct lov_mds_md_v1 *lmm = NULL;
4372 struct dt_device *dev = lfsck_obj2dev(obj);
4373 struct lustre_handle lh = { 0 };
4374 struct lu_buf ea_buf = { NULL };
4377 bool locked = false;
4378 bool stripe = false;
4379 bool bad_oi = false;
4382 if (!S_ISREG(lfsck_object_type(obj)))
4385 if (lad->lad_assistant_status < 0)
4386 GOTO(out, rc = -ESRCH);
4388 fid_to_lmm_oi(lfsck_dto2fid(obj), oi);
4389 lmm_oi_cpu_to_le(oi, oi);
4390 dt_read_lock(env, obj, 0);
4394 if (dt_object_exists(obj) == 0 ||
4395 lfsck_is_dead_obj(obj))
4398 rc = lfsck_layout_get_lovea(env, obj, buf);
4404 rc = lfsck_layout_verify_header(lmm);
4405 /* If the LOV EA crashed, then it is possible to be rebuilt later
4406 * when handle orphan OST-objects. */
4410 if (memcmp(oi, &lmm->lmm_oi, sizeof(*oi)) == 0)
4411 GOTO(out, stripe = true);
4413 /* Inconsistent lmm_oi, should be repaired. */
4417 if (bk->lb_param & LPF_DRYRUN) {
4418 lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4420 GOTO(out, stripe = true);
4423 if (!lustre_handle_is_used(&lh)) {
4424 dt_read_unlock(env, obj);
4426 rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
4427 MDS_INODELOCK_LAYOUT |
4428 MDS_INODELOCK_XATTR, LCK_EX);
4432 handle = dt_trans_create(env, dev);
4434 GOTO(out, rc = PTR_ERR(handle));
4436 lfsck_buf_init(&ea_buf, lmm, size);
4437 rc = dt_declare_xattr_set(env, obj, &ea_buf, XATTR_NAME_LOV,
4438 LU_XATTR_REPLACE, handle);
4442 rc = dt_trans_start_local(env, dev, handle);
4446 dt_write_lock(env, obj, 0);
4452 rc = dt_xattr_set(env, obj, &ea_buf, XATTR_NAME_LOV,
4453 LU_XATTR_REPLACE, handle);
4457 lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4459 GOTO(out, stripe = true);
4463 if (lustre_handle_is_used(&lh))
4464 dt_write_unlock(env, obj);
4466 dt_read_unlock(env, obj);
4469 if (handle != NULL && !IS_ERR(handle))
4470 dt_trans_stop(env, dev, handle);
4472 lfsck_ibits_unlock(&lh, LCK_EX);
4475 CDEBUG(D_LFSCK, "%s: layout LFSCK master %s bad lmm_oi for "
4476 DFID": rc = %d\n", lfsck_lfsck2name(lfsck),
4477 bk->lb_param & LPF_DRYRUN ? "found" : "repaired",
4478 PFID(lfsck_dto2fid(obj)), rc);
4481 rc = lfsck_layout_scan_stripes(env, com, obj, lmm);
4483 down_write(&com->lc_sem);
4484 com->lc_new_checked++;
4486 lfsck_layout_record_failure(env, lfsck, lo);
4487 up_write(&com->lc_sem);
4493 static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
4494 struct lfsck_component *com,
4495 struct dt_object *obj)
4497 struct lfsck_instance *lfsck = com->lc_lfsck;
4498 struct lfsck_layout *lo = com->lc_file_ram;
4499 const struct lu_fid *fid = lfsck_dto2fid(obj);
4500 struct lfsck_layout_slave_data *llsd = com->lc_data;
4501 struct lfsck_layout_seq *lls;
4507 LASSERT(llsd != NULL);
4509 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY5) &&
4510 cfs_fail_val == lfsck_dev_idx(lfsck)) {
4511 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(1),
4513 struct ptlrpc_thread *thread = &lfsck->li_thread;
4515 l_wait_event(thread->t_ctl_waitq,
4516 !thread_is_running(thread),
4520 lfsck_rbtree_update_bitmap(env, com, fid, false);
4522 down_write(&com->lc_sem);
4523 if (fid_is_idif(fid))
4525 else if (!fid_is_norm(fid) ||
4526 !fid_is_for_ostobj(env, lfsck, obj, fid))
4527 GOTO(unlock, rc = 0);
4530 com->lc_new_checked++;
4532 lls = lfsck_layout_seq_lookup(llsd, seq);
4535 if (unlikely(lls == NULL))
4536 GOTO(unlock, rc = -ENOMEM);
4538 INIT_LIST_HEAD(&lls->lls_list);
4540 rc = lfsck_layout_lastid_load(env, com, lls);
4542 CDEBUG(D_LFSCK, "%s: layout LFSCK failed to "
4543 "load LAST_ID for "LPX64": rc = %d\n",
4544 lfsck_lfsck2name(com->lc_lfsck), seq, rc);
4545 lo->ll_objs_failed_phase1++;
4550 lfsck_layout_seq_insert(llsd, lls);
4553 if (unlikely(fid_is_last_id(fid)))
4554 GOTO(unlock, rc = 0);
4556 if (fid_is_idif(fid))
4557 oid = fid_idif_id(fid_seq(fid), fid_oid(fid), fid_ver(fid));
4561 if (oid > lls->lls_lastid_known)
4562 lls->lls_lastid_known = oid;
4564 if (oid > lls->lls_lastid) {
4565 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
4566 /* OFD may create new objects during LFSCK scanning. */
4567 rc = lfsck_layout_lastid_reload(env, com, lls);
4568 if (unlikely(rc != 0)) {
4569 CDEBUG(D_LFSCK, "%s: layout LFSCK failed to "
4570 "reload LAST_ID for "LPX64": rc = %d\n",
4571 lfsck_lfsck2name(com->lc_lfsck),
4577 if (oid <= lls->lls_lastid ||
4578 lo->ll_flags & LF_CRASHED_LASTID)
4579 GOTO(unlock, rc = 0);
4581 LASSERT(lfsck->li_out_notify != NULL);
4583 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4584 LE_LASTID_REBUILDING);
4585 lo->ll_flags |= LF_CRASHED_LASTID;
4587 CDEBUG(D_LFSCK, "%s: layout LFSCK finds crashed "
4588 "LAST_ID file (2) for the sequence "LPX64
4589 ", old value "LPU64", known value "LPU64"\n",
4590 lfsck_lfsck2name(lfsck), lls->lls_seq,
4591 lls->lls_lastid, oid);
4594 lls->lls_lastid = oid;
4598 GOTO(unlock, rc = 0);
4601 up_write(&com->lc_sem);
4606 static int lfsck_layout_exec_dir(const struct lu_env *env,
4607 struct lfsck_component *com,
4608 struct lfsck_assistant_object *lso,
4609 struct lu_dirent *ent, __u16 type)
4614 static int lfsck_layout_master_post(const struct lu_env *env,
4615 struct lfsck_component *com,
4616 int result, bool init)
4618 struct lfsck_instance *lfsck = com->lc_lfsck;
4619 struct lfsck_layout *lo = com->lc_file_ram;
4623 lfsck_post_generic(env, com, &result);
4625 down_write(&com->lc_sem);
4626 spin_lock(&lfsck->li_lock);
4628 lo->ll_pos_last_checkpoint =
4629 lfsck->li_pos_checkpoint.lp_oit_cookie;
4632 if (lo->ll_flags & LF_INCOMPLETE)
4633 lo->ll_status = LS_PARTIAL;
4635 lo->ll_status = LS_SCANNING_PHASE2;
4636 lo->ll_flags |= LF_SCANNED_ONCE;
4637 lo->ll_flags &= ~LF_UPGRADE;
4638 list_move_tail(&com->lc_link, &lfsck->li_list_double_scan);
4639 } else if (result == 0) {
4640 if (lfsck->li_status != 0)
4641 lo->ll_status = lfsck->li_status;
4643 lo->ll_status = LS_STOPPED;
4644 if (lo->ll_status != LS_PAUSED)
4645 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4647 lo->ll_status = LS_FAILED;
4648 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4650 spin_unlock(&lfsck->li_lock);
4653 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4654 HALF_SEC - lfsck->li_time_last_checkpoint);
4655 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4656 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4657 com->lc_new_checked = 0;
4660 rc = lfsck_layout_store(env, com);
4661 up_write(&com->lc_sem);
4663 CDEBUG(D_LFSCK, "%s: layout LFSCK master post done: rc = %d\n",
4664 lfsck_lfsck2name(lfsck), rc);
4669 static int lfsck_layout_slave_post(const struct lu_env *env,
4670 struct lfsck_component *com,
4671 int result, bool init)
4673 struct lfsck_instance *lfsck = com->lc_lfsck;
4674 struct lfsck_layout *lo = com->lc_file_ram;
4678 rc = lfsck_layout_lastid_store(env, com);
4682 LASSERT(lfsck->li_out_notify != NULL);
4684 down_write(&com->lc_sem);
4685 spin_lock(&lfsck->li_lock);
4687 lo->ll_pos_last_checkpoint =
4688 lfsck->li_pos_checkpoint.lp_oit_cookie;
4691 lo->ll_status = LS_SCANNING_PHASE2;
4692 lo->ll_flags |= LF_SCANNED_ONCE;
4693 if (lo->ll_flags & LF_CRASHED_LASTID) {
4695 lo->ll_flags &= ~LF_CRASHED_LASTID;
4697 CDEBUG(D_LFSCK, "%s: layout LFSCK has rebuilt "
4698 "crashed LAST_ID files successfully\n",
4699 lfsck_lfsck2name(lfsck));
4701 lo->ll_flags &= ~LF_UPGRADE;
4702 list_move_tail(&com->lc_link, &lfsck->li_list_double_scan);
4703 } else if (result == 0) {
4704 if (lfsck->li_status != 0)
4705 lo->ll_status = lfsck->li_status;
4707 lo->ll_status = LS_STOPPED;
4708 if (lo->ll_status != LS_PAUSED)
4709 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4711 lo->ll_status = LS_FAILED;
4712 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4714 spin_unlock(&lfsck->li_lock);
4717 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4721 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4722 HALF_SEC - lfsck->li_time_last_checkpoint);
4723 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4724 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4725 com->lc_new_checked = 0;
4728 rc = lfsck_layout_store(env, com);
4729 up_write(&com->lc_sem);
4731 lfsck_layout_slave_notify_master(env, com, LE_PHASE1_DONE, result);
4733 CDEBUG(D_LFSCK, "%s: layout LFSCK slave post done: rc = %d\n",
4734 lfsck_lfsck2name(lfsck), rc);
4739 static int lfsck_layout_dump(const struct lu_env *env,
4740 struct lfsck_component *com, struct seq_file *m)
4742 struct lfsck_instance *lfsck = com->lc_lfsck;
4743 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
4744 struct lfsck_layout *lo = com->lc_file_ram;
4747 down_read(&com->lc_sem);
4748 seq_printf(m, "name: lfsck_layout\n"
4754 lfsck_status2names(lo->ll_status));
4756 rc = lfsck_bits_dump(m, lo->ll_flags, lfsck_flags_names, "flags");
4760 rc = lfsck_bits_dump(m, bk->lb_param, lfsck_param_names, "param");
4764 rc = lfsck_time_dump(m, lo->ll_time_last_complete,
4769 rc = lfsck_time_dump(m, lo->ll_time_latest_start,
4774 rc = lfsck_time_dump(m, lo->ll_time_last_checkpoint,
4779 seq_printf(m, "latest_start_position: "LPU64"\n"
4780 "last_checkpoint_position: "LPU64"\n"
4781 "first_failure_position: "LPU64"\n",
4782 lo->ll_pos_latest_start,
4783 lo->ll_pos_last_checkpoint,
4784 lo->ll_pos_first_inconsistent);
4786 seq_printf(m, "success_count: %u\n"
4787 "repaired_dangling: "LPU64"\n"
4788 "repaired_unmatched_pair: "LPU64"\n"
4789 "repaired_multiple_referenced: "LPU64"\n"
4790 "repaired_orphan: "LPU64"\n"
4791 "repaired_inconsistent_owner: "LPU64"\n"
4792 "repaired_others: "LPU64"\n"
4793 "skipped: "LPU64"\n"
4794 "failed_phase1: "LPU64"\n"
4795 "failed_phase2: "LPU64"\n",
4796 lo->ll_success_count,
4797 lo->ll_objs_repaired[LLIT_DANGLING - 1],
4798 lo->ll_objs_repaired[LLIT_UNMATCHED_PAIR - 1],
4799 lo->ll_objs_repaired[LLIT_MULTIPLE_REFERENCED - 1],
4800 lo->ll_objs_repaired[LLIT_ORPHAN - 1],
4801 lo->ll_objs_repaired[LLIT_INCONSISTENT_OWNER - 1],
4802 lo->ll_objs_repaired[LLIT_OTHERS - 1],
4803 lo->ll_objs_skipped,
4804 lo->ll_objs_failed_phase1,
4805 lo->ll_objs_failed_phase2);
4807 if (lo->ll_status == LS_SCANNING_PHASE1) {
4809 const struct dt_it_ops *iops;
4810 cfs_duration_t duration = cfs_time_current() -
4811 lfsck->li_time_last_checkpoint;
4812 __u64 checked = lo->ll_objs_checked_phase1 +
4813 com->lc_new_checked;
4814 __u64 speed = checked;
4815 __u64 new_checked = com->lc_new_checked *
4816 msecs_to_jiffies(MSEC_PER_SEC);
4817 __u32 rtime = lo->ll_run_time_phase1 +
4818 cfs_duration_sec(duration + HALF_SEC);
4821 do_div(new_checked, duration);
4823 do_div(speed, rtime);
4824 seq_printf(m, "checked_phase1: "LPU64"\n"
4825 "checked_phase2: "LPU64"\n"
4826 "run_time_phase1: %u seconds\n"
4827 "run_time_phase2: %u seconds\n"
4828 "average_speed_phase1: "LPU64" items/sec\n"
4829 "average_speed_phase2: N/A\n"
4830 "real-time_speed_phase1: "LPU64" items/sec\n"
4831 "real-time_speed_phase2: N/A\n",
4833 lo->ll_objs_checked_phase2,
4835 lo->ll_run_time_phase2,
4839 LASSERT(lfsck->li_di_oit != NULL);
4841 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
4843 /* The low layer otable-based iteration position may NOT
4844 * exactly match the layout-based directory traversal
4845 * cookie. Generally, it is not a serious issue. But the
4846 * caller should NOT make assumption on that. */
4847 pos = iops->store(env, lfsck->li_di_oit);
4848 if (!lfsck->li_current_oit_processed)
4850 seq_printf(m, "current_position: "LPU64"\n", pos);
4852 } else if (lo->ll_status == LS_SCANNING_PHASE2) {
4853 cfs_duration_t duration = cfs_time_current() -
4854 com->lc_time_last_checkpoint;
4855 __u64 checked = lo->ll_objs_checked_phase2 +
4856 com->lc_new_checked;
4857 __u64 speed1 = lo->ll_objs_checked_phase1;
4858 __u64 speed2 = checked;
4859 __u64 new_checked = com->lc_new_checked *
4860 msecs_to_jiffies(MSEC_PER_SEC);
4861 __u32 rtime = lo->ll_run_time_phase2 +
4862 cfs_duration_sec(duration + HALF_SEC);
4865 do_div(new_checked, duration);
4866 if (lo->ll_run_time_phase1 != 0)
4867 do_div(speed1, lo->ll_run_time_phase1);
4869 do_div(speed2, rtime);
4870 rc = seq_printf(m, "checked_phase1: "LPU64"\n"
4871 "checked_phase2: "LPU64"\n"
4872 "run_time_phase1: %u seconds\n"
4873 "run_time_phase2: %u seconds\n"
4874 "average_speed_phase1: "LPU64" items/sec\n"
4875 "average_speed_phase2: "LPU64" items/sec\n"
4876 "real-time_speed_phase1: N/A\n"
4877 "real-time_speed_phase2: "LPU64" items/sec\n"
4878 "current_position: "DFID"\n",
4879 lo->ll_objs_checked_phase1,
4881 lo->ll_run_time_phase1,
4886 PFID(&com->lc_fid_latest_scanned_phase2));
4891 __u64 speed1 = lo->ll_objs_checked_phase1;
4892 __u64 speed2 = lo->ll_objs_checked_phase2;
4894 if (lo->ll_run_time_phase1 != 0)
4895 do_div(speed1, lo->ll_run_time_phase1);
4896 if (lo->ll_run_time_phase2 != 0)
4897 do_div(speed2, lo->ll_run_time_phase2);
4898 seq_printf(m, "checked_phase1: "LPU64"\n"
4899 "checked_phase2: "LPU64"\n"
4900 "run_time_phase1: %u seconds\n"
4901 "run_time_phase2: %u seconds\n"
4902 "average_speed_phase1: "LPU64" items/sec\n"
4903 "average_speed_phase2: "LPU64" objs/sec\n"
4904 "real-time_speed_phase1: N/A\n"
4905 "real-time_speed_phase2: N/A\n"
4906 "current_position: N/A\n",
4907 lo->ll_objs_checked_phase1,
4908 lo->ll_objs_checked_phase2,
4909 lo->ll_run_time_phase1,
4910 lo->ll_run_time_phase2,
4915 up_read(&com->lc_sem);
4920 static int lfsck_layout_master_double_scan(const struct lu_env *env,
4921 struct lfsck_component *com)
4923 struct lfsck_layout *lo = com->lc_file_ram;
4924 struct lfsck_assistant_data *lad = com->lc_data;
4925 struct lfsck_instance *lfsck = com->lc_lfsck;
4926 struct lfsck_tgt_descs *ltds;
4927 struct lfsck_tgt_desc *ltd;
4928 struct lfsck_tgt_desc *next;
4931 rc = lfsck_double_scan_generic(env, com, lo->ll_status);
4933 if (thread_is_stopped(&lad->lad_thread)) {
4934 LASSERT(list_empty(&lad->lad_req_list));
4935 LASSERT(list_empty(&lad->lad_ost_phase1_list));
4936 LASSERT(list_empty(&lad->lad_mdt_phase1_list));
4938 ltds = &lfsck->li_ost_descs;
4939 spin_lock(<ds->ltd_lock);
4940 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase2_list,
4941 ltd_layout_phase_list) {
4942 list_del_init(<d->ltd_layout_phase_list);
4944 spin_unlock(<ds->ltd_lock);
4946 ltds = &lfsck->li_mdt_descs;
4947 spin_lock(<ds->ltd_lock);
4948 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
4949 ltd_layout_phase_list) {
4950 list_del_init(<d->ltd_layout_phase_list);
4952 spin_unlock(<ds->ltd_lock);
4958 static int lfsck_layout_slave_double_scan(const struct lu_env *env,
4959 struct lfsck_component *com)
4961 struct lfsck_instance *lfsck = com->lc_lfsck;
4962 struct lfsck_layout_slave_data *llsd = com->lc_data;
4963 struct lfsck_layout *lo = com->lc_file_ram;
4964 struct ptlrpc_thread *thread = &lfsck->li_thread;
4968 CDEBUG(D_LFSCK, "%s: layout LFSCK slave phase2 scan start\n",
4969 lfsck_lfsck2name(lfsck));
4971 atomic_inc(&lfsck->li_double_scan_count);
4973 if (lo->ll_flags & LF_INCOMPLETE)
4976 com->lc_new_checked = 0;
4977 com->lc_new_scanned = 0;
4978 com->lc_time_last_checkpoint = cfs_time_current();
4979 com->lc_time_next_checkpoint = com->lc_time_last_checkpoint +
4980 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
4983 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(30),
4986 rc = lfsck_layout_slave_query_master(env, com);
4987 if (list_empty(&llsd->llsd_master_list)) {
4988 if (unlikely(!thread_is_running(thread)))
4999 rc = l_wait_event(thread->t_ctl_waitq,
5000 !thread_is_running(thread) ||
5001 lo->ll_flags & LF_INCOMPLETE ||
5002 list_empty(&llsd->llsd_master_list),
5004 if (unlikely(!thread_is_running(thread)))
5007 if (lo->ll_flags & LF_INCOMPLETE)
5010 if (rc == -ETIMEDOUT)
5013 GOTO(done, rc = (rc < 0 ? rc : 1));
5017 rc = lfsck_layout_double_scan_result(env, com, rc);
5018 lfsck_layout_slave_notify_master(env, com, LE_PHASE2_DONE,
5019 (rc > 0 && lo->ll_flags & LF_INCOMPLETE) ? 0 : rc);
5020 lfsck_layout_slave_quit(env, com);
5021 if (atomic_dec_and_test(&lfsck->li_double_scan_count))
5022 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5024 CDEBUG(D_LFSCK, "%s: layout LFSCK slave phase2 scan finished, "
5025 "status %d: rc = %d\n",
5026 lfsck_lfsck2name(lfsck), lo->ll_status, rc);
5031 static void lfsck_layout_master_data_release(const struct lu_env *env,
5032 struct lfsck_component *com)
5034 struct lfsck_assistant_data *lad = com->lc_data;
5035 struct lfsck_instance *lfsck = com->lc_lfsck;
5036 struct lfsck_tgt_descs *ltds;
5037 struct lfsck_tgt_desc *ltd;
5038 struct lfsck_tgt_desc *next;
5040 LASSERT(lad != NULL);
5041 LASSERT(thread_is_init(&lad->lad_thread) ||
5042 thread_is_stopped(&lad->lad_thread));
5043 LASSERT(list_empty(&lad->lad_req_list));
5045 com->lc_data = NULL;
5047 ltds = &lfsck->li_ost_descs;
5048 spin_lock(<ds->ltd_lock);
5049 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase1_list,
5050 ltd_layout_phase_list) {
5051 list_del_init(<d->ltd_layout_phase_list);
5053 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase2_list,
5054 ltd_layout_phase_list) {
5055 list_del_init(<d->ltd_layout_phase_list);
5057 list_for_each_entry_safe(ltd, next, &lad->lad_ost_list,
5059 list_del_init(<d->ltd_layout_list);
5061 spin_unlock(<ds->ltd_lock);
5063 ltds = &lfsck->li_mdt_descs;
5064 spin_lock(<ds->ltd_lock);
5065 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase1_list,
5066 ltd_layout_phase_list) {
5067 list_del_init(<d->ltd_layout_phase_list);
5069 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
5070 ltd_layout_phase_list) {
5071 list_del_init(<d->ltd_layout_phase_list);
5073 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_list,
5075 list_del_init(<d->ltd_layout_list);
5077 spin_unlock(<ds->ltd_lock);
5079 if (likely(lad->lad_bitmap != NULL))
5080 CFS_FREE_BITMAP(lad->lad_bitmap);
5085 static void lfsck_layout_slave_data_release(const struct lu_env *env,
5086 struct lfsck_component *com)
5088 struct lfsck_layout_slave_data *llsd = com->lc_data;
5090 lfsck_layout_slave_quit(env, com);
5091 com->lc_data = NULL;
5095 static void lfsck_layout_master_quit(const struct lu_env *env,
5096 struct lfsck_component *com)
5098 struct lfsck_assistant_data *lad = com->lc_data;
5099 struct lfsck_instance *lfsck = com->lc_lfsck;
5100 struct lfsck_tgt_descs *ltds;
5101 struct lfsck_tgt_desc *ltd;
5102 struct lfsck_tgt_desc *next;
5104 LASSERT(lad != NULL);
5106 lfsck_quit_generic(env, com);
5108 LASSERT(thread_is_init(&lad->lad_thread) ||
5109 thread_is_stopped(&lad->lad_thread));
5110 LASSERT(list_empty(&lad->lad_req_list));
5112 ltds = &lfsck->li_ost_descs;
5113 spin_lock(<ds->ltd_lock);
5114 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase1_list,
5115 ltd_layout_phase_list) {
5116 list_del_init(<d->ltd_layout_phase_list);
5118 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase2_list,
5119 ltd_layout_phase_list) {
5120 list_del_init(<d->ltd_layout_phase_list);
5122 spin_unlock(<ds->ltd_lock);
5124 ltds = &lfsck->li_mdt_descs;
5125 spin_lock(<ds->ltd_lock);
5126 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase1_list,
5127 ltd_layout_phase_list) {
5128 list_del_init(<d->ltd_layout_phase_list);
5130 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
5131 ltd_layout_phase_list) {
5132 list_del_init(<d->ltd_layout_phase_list);
5134 spin_unlock(<ds->ltd_lock);
5137 static void lfsck_layout_slave_quit(const struct lu_env *env,
5138 struct lfsck_component *com)
5140 struct lfsck_layout_slave_data *llsd = com->lc_data;
5141 struct lfsck_layout_seq *lls;
5142 struct lfsck_layout_seq *next;
5143 struct lfsck_layout_slave_target *llst;
5145 LASSERT(llsd != NULL);
5147 list_for_each_entry_safe(lls, next, &llsd->llsd_seq_list,
5149 list_del_init(&lls->lls_list);
5150 lfsck_object_put(env, lls->lls_lastid_obj);
5154 spin_lock(&llsd->llsd_lock);
5155 while (!list_empty(&llsd->llsd_master_list)) {
5156 llst = list_entry(llsd->llsd_master_list.next,
5157 struct lfsck_layout_slave_target, llst_list);
5158 list_del_init(&llst->llst_list);
5159 spin_unlock(&llsd->llsd_lock);
5160 lfsck_layout_llst_put(llst);
5161 spin_lock(&llsd->llsd_lock);
5163 spin_unlock(&llsd->llsd_lock);
5165 lfsck_rbtree_cleanup(env, com);
5168 static int lfsck_layout_master_in_notify(const struct lu_env *env,
5169 struct lfsck_component *com,
5170 struct lfsck_request *lr,
5173 struct lfsck_instance *lfsck = com->lc_lfsck;
5174 struct lfsck_layout *lo = com->lc_file_ram;
5175 struct lfsck_assistant_data *lad = com->lc_data;
5176 struct lfsck_tgt_descs *ltds;
5177 struct lfsck_tgt_desc *ltd;
5181 if (lr->lr_event == LE_PAIRS_VERIFY) {
5184 rc = lfsck_layout_master_check_pairs(env, com, &lr->lr_fid,
5190 CDEBUG(D_LFSCK, "%s: layout LFSCK master handles notify %u "
5191 "from %s %x, status %d, flags %x, flags2 %x\n",
5192 lfsck_lfsck2name(lfsck), lr->lr_event,
5193 (lr->lr_flags & LEF_FROM_OST) ? "OST" : "MDT",
5194 lr->lr_index, lr->lr_status, lr->lr_flags, lr->lr_flags2);
5196 if (lr->lr_event != LE_PHASE1_DONE &&
5197 lr->lr_event != LE_PHASE2_DONE &&
5198 lr->lr_event != LE_PEER_EXIT)
5201 if (lr->lr_flags & LEF_FROM_OST)
5202 ltds = &lfsck->li_ost_descs;
5204 ltds = &lfsck->li_mdt_descs;
5205 spin_lock(<ds->ltd_lock);
5206 ltd = LTD_TGT(ltds, lr->lr_index);
5208 spin_unlock(<ds->ltd_lock);
5213 list_del_init(<d->ltd_layout_phase_list);
5214 switch (lr->lr_event) {
5215 case LE_PHASE1_DONE:
5216 if (lr->lr_status <= 0 || lr->lr_flags2 & LF_INCOMPLETE) {
5217 if (lr->lr_flags2 & LF_INCOMPLETE) {
5218 if (lr->lr_flags & LEF_FROM_OST)
5219 lfsck_lad_set_bitmap(env, com,
5222 lo->ll_flags |= LF_INCOMPLETE;
5224 ltd->ltd_layout_done = 1;
5225 list_del_init(<d->ltd_layout_list);
5230 if (lr->lr_flags & LEF_FROM_OST) {
5231 if (list_empty(<d->ltd_layout_list))
5232 list_add_tail(<d->ltd_layout_list,
5233 &lad->lad_ost_list);
5234 list_add_tail(<d->ltd_layout_phase_list,
5235 &lad->lad_ost_phase2_list);
5237 if (list_empty(<d->ltd_layout_list))
5238 list_add_tail(<d->ltd_layout_list,
5239 &lad->lad_mdt_list);
5240 list_add_tail(<d->ltd_layout_phase_list,
5241 &lad->lad_mdt_phase2_list);
5244 case LE_PHASE2_DONE:
5245 ltd->ltd_layout_done = 1;
5246 if (!list_empty(<d->ltd_layout_list)) {
5247 list_del_init(<d->ltd_layout_list);
5248 if (lr->lr_flags2 & LF_INCOMPLETE) {
5249 lfsck_lad_set_bitmap(env, com, ltd->ltd_index);
5257 ltd->ltd_layout_done = 1;
5258 list_del_init(<d->ltd_layout_list);
5259 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) &&
5260 !(lr->lr_flags & LEF_FROM_OST))
5261 lo->ll_flags |= LF_INCOMPLETE;
5266 spin_unlock(<ds->ltd_lock);
5268 if (fail && lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
5269 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5271 memset(stop, 0, sizeof(*stop));
5272 stop->ls_status = lr->lr_status;
5273 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5274 lfsck_stop(env, lfsck->li_bottom, stop);
5275 } else if (lfsck_phase2_next_ready(lad)) {
5276 wake_up_all(&lad->lad_thread.t_ctl_waitq);
5282 static int lfsck_layout_slave_in_notify(const struct lu_env *env,
5283 struct lfsck_component *com,
5284 struct lfsck_request *lr,
5287 struct lfsck_instance *lfsck = com->lc_lfsck;
5288 struct lfsck_layout_slave_data *llsd = com->lc_data;
5289 struct lfsck_layout_slave_target *llst;
5293 switch (lr->lr_event) {
5294 case LE_FID_ACCESSED:
5295 lfsck_rbtree_update_bitmap(env, com, &lr->lr_fid, true);
5297 case LE_CONDITIONAL_DESTROY:
5298 rc = lfsck_layout_slave_conditional_destroy(env, com, lr);
5300 case LE_PAIRS_VERIFY: {
5301 lr->lr_status = LPVS_INIT;
5302 /* Firstly, if the MDT-object which is claimed via OST-object
5303 * local stored PFID xattr recognizes the OST-object, then it
5304 * must be that the client given PFID is wrong. */
5305 rc = lfsck_layout_slave_check_pairs(env, com, &lr->lr_fid,
5310 lr->lr_status = LPVS_INCONSISTENT;
5311 /* The OST-object local stored PFID xattr is stale. We need to
5312 * check whether the MDT-object that is claimed via the client
5313 * given PFID information recognizes the OST-object or not. If
5314 * matches, then need to update the OST-object's PFID xattr. */
5315 rc = lfsck_layout_slave_check_pairs(env, com, &lr->lr_fid,
5318 * We are not sure whether the client given PFID information
5319 * is correct or not, do nothing to avoid improper fixing.
5322 * The client given PFID information is also invalid, we can
5323 * NOT fix the OST-object inconsistency.
5328 lr->lr_status = LPVS_INCONSISTENT_TOFIX;
5329 rc = lfsck_layout_slave_repair_pfid(env, com, lr);
5333 case LE_PHASE1_DONE: {
5334 if (lr->lr_flags2 & LF_INCOMPLETE) {
5335 struct lfsck_layout *lo = com->lc_file_ram;
5337 lo->ll_flags |= LF_INCOMPLETE;
5338 llst = lfsck_layout_llst_find_and_del(llsd,
5342 lfsck_layout_llst_put(llst);
5343 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5349 case LE_PHASE2_DONE:
5351 CDEBUG(D_LFSCK, "%s: layout LFSCK slave handle notify %u "
5352 "from MDT %x, status %d\n", lfsck_lfsck2name(lfsck),
5353 lr->lr_event, lr->lr_index, lr->lr_status);
5359 llst = lfsck_layout_llst_find_and_del(llsd, lr->lr_index, true);
5363 lfsck_layout_llst_put(llst);
5364 if (list_empty(&llsd->llsd_master_list))
5365 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5367 if (lr->lr_event == LE_PEER_EXIT &&
5368 (lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT ||
5369 (list_empty(&llsd->llsd_master_list) &&
5370 (lr->lr_status == LS_STOPPED ||
5371 lr->lr_status == LS_CO_STOPPED)))) {
5372 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5374 memset(stop, 0, sizeof(*stop));
5375 stop->ls_status = lr->lr_status;
5376 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5377 lfsck_stop(env, lfsck->li_bottom, stop);
5383 static int lfsck_layout_query(const struct lu_env *env,
5384 struct lfsck_component *com)
5386 struct lfsck_layout *lo = com->lc_file_ram;
5388 return lo->ll_status;
5391 /* with lfsck::li_lock held */
5392 static int lfsck_layout_slave_join(const struct lu_env *env,
5393 struct lfsck_component *com,
5394 struct lfsck_start_param *lsp)
5396 struct lfsck_instance *lfsck = com->lc_lfsck;
5397 struct lfsck_layout_slave_data *llsd = com->lc_data;
5398 struct lfsck_layout_slave_target *llst;
5399 struct lfsck_start *start = lsp->lsp_start;
5403 if (start == NULL || !(start->ls_flags & LPF_OST_ORPHAN))
5406 if (!lsp->lsp_index_valid)
5409 /* If someone is running the LFSCK without orphan handling,
5410 * it will not maintain the object accessing rbtree. So we
5411 * cannot join it for orphan handling. */
5412 if (!llsd->llsd_rbtree_valid)
5415 spin_unlock(&lfsck->li_lock);
5416 rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
5417 spin_lock(&lfsck->li_lock);
5418 if (rc == 0 && !thread_is_running(&lfsck->li_thread)) {
5419 spin_unlock(&lfsck->li_lock);
5420 llst = lfsck_layout_llst_find_and_del(llsd, lsp->lsp_index,
5423 lfsck_layout_llst_put(llst);
5424 spin_lock(&lfsck->li_lock);
5431 static struct lfsck_operations lfsck_layout_master_ops = {
5432 .lfsck_reset = lfsck_layout_reset,
5433 .lfsck_fail = lfsck_layout_fail,
5434 .lfsck_checkpoint = lfsck_layout_master_checkpoint,
5435 .lfsck_prep = lfsck_layout_master_prep,
5436 .lfsck_exec_oit = lfsck_layout_master_exec_oit,
5437 .lfsck_exec_dir = lfsck_layout_exec_dir,
5438 .lfsck_post = lfsck_layout_master_post,
5439 .lfsck_dump = lfsck_layout_dump,
5440 .lfsck_double_scan = lfsck_layout_master_double_scan,
5441 .lfsck_data_release = lfsck_layout_master_data_release,
5442 .lfsck_quit = lfsck_layout_master_quit,
5443 .lfsck_in_notify = lfsck_layout_master_in_notify,
5444 .lfsck_query = lfsck_layout_query,
5447 static struct lfsck_operations lfsck_layout_slave_ops = {
5448 .lfsck_reset = lfsck_layout_reset,
5449 .lfsck_fail = lfsck_layout_fail,
5450 .lfsck_checkpoint = lfsck_layout_slave_checkpoint,
5451 .lfsck_prep = lfsck_layout_slave_prep,
5452 .lfsck_exec_oit = lfsck_layout_slave_exec_oit,
5453 .lfsck_exec_dir = lfsck_layout_exec_dir,
5454 .lfsck_post = lfsck_layout_slave_post,
5455 .lfsck_dump = lfsck_layout_dump,
5456 .lfsck_double_scan = lfsck_layout_slave_double_scan,
5457 .lfsck_data_release = lfsck_layout_slave_data_release,
5458 .lfsck_quit = lfsck_layout_slave_quit,
5459 .lfsck_in_notify = lfsck_layout_slave_in_notify,
5460 .lfsck_query = lfsck_layout_query,
5461 .lfsck_join = lfsck_layout_slave_join,
5464 static void lfsck_layout_assistant_fill_pos(const struct lu_env *env,
5465 struct lfsck_component *com,
5466 struct lfsck_position *pos)
5468 struct lfsck_assistant_data *lad = com->lc_data;
5469 struct lfsck_layout_req *llr;
5471 if (list_empty(&lad->lad_req_list))
5474 llr = list_entry(lad->lad_req_list.next,
5475 struct lfsck_layout_req,
5477 pos->lp_oit_cookie = llr->llr_lar.lar_parent->lso_oit_cookie - 1;
5480 struct lfsck_assistant_operations lfsck_layout_assistant_ops = {
5481 .la_handler_p1 = lfsck_layout_assistant_handler_p1,
5482 .la_handler_p2 = lfsck_layout_assistant_handler_p2,
5483 .la_fill_pos = lfsck_layout_assistant_fill_pos,
5484 .la_double_scan_result = lfsck_layout_double_scan_result,
5485 .la_req_fini = lfsck_layout_assistant_req_fini,
5486 .la_sync_failures = lfsck_layout_assistant_sync_failures,
5489 int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
5491 struct lfsck_component *com;
5492 struct lfsck_layout *lo;
5493 struct dt_object *root = NULL;
5494 struct dt_object *obj;
5502 INIT_LIST_HEAD(&com->lc_link);
5503 INIT_LIST_HEAD(&com->lc_link_dir);
5504 init_rwsem(&com->lc_sem);
5505 atomic_set(&com->lc_ref, 1);
5506 com->lc_lfsck = lfsck;
5507 com->lc_type = LFSCK_TYPE_LAYOUT;
5508 if (lfsck->li_master) {
5509 com->lc_ops = &lfsck_layout_master_ops;
5510 com->lc_data = lfsck_assistant_data_init(
5511 &lfsck_layout_assistant_ops,
5513 if (com->lc_data == NULL)
5514 GOTO(out, rc = -ENOMEM);
5516 struct lfsck_layout_slave_data *llsd;
5518 com->lc_ops = &lfsck_layout_slave_ops;
5519 OBD_ALLOC_PTR(llsd);
5521 GOTO(out, rc = -ENOMEM);
5523 INIT_LIST_HEAD(&llsd->llsd_seq_list);
5524 INIT_LIST_HEAD(&llsd->llsd_master_list);
5525 spin_lock_init(&llsd->llsd_lock);
5526 llsd->llsd_rb_root = RB_ROOT;
5527 rwlock_init(&llsd->llsd_rb_lock);
5528 com->lc_data = llsd;
5530 com->lc_file_size = sizeof(*lo);
5531 OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
5532 if (com->lc_file_ram == NULL)
5533 GOTO(out, rc = -ENOMEM);
5535 OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
5536 if (com->lc_file_disk == NULL)
5537 GOTO(out, rc = -ENOMEM);
5539 root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
5541 GOTO(out, rc = PTR_ERR(root));
5543 if (unlikely(!dt_try_as_dir(env, root)))
5544 GOTO(out, rc = -ENOTDIR);
5546 obj = local_file_find_or_create(env, lfsck->li_los, root,
5548 S_IFREG | S_IRUGO | S_IWUSR);
5550 GOTO(out, rc = PTR_ERR(obj));
5553 rc = lfsck_layout_load(env, com);
5555 rc = lfsck_layout_reset(env, com, true);
5556 else if (rc == -ENOENT)
5557 rc = lfsck_layout_init(env, com);
5562 lo = com->lc_file_ram;
5563 switch (lo->ll_status) {
5569 spin_lock(&lfsck->li_lock);
5570 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
5571 spin_unlock(&lfsck->li_lock);
5574 CERROR("%s: unknown lfsck_layout status %d\n",
5575 lfsck_lfsck2name(lfsck), lo->ll_status);
5577 case LS_SCANNING_PHASE1:
5578 case LS_SCANNING_PHASE2:
5579 /* No need to store the status to disk right now.
5580 * If the system crashed before the status stored,
5581 * it will be loaded back when next time. */
5582 lo->ll_status = LS_CRASHED;
5583 if (!lfsck->li_master)
5584 lo->ll_flags |= LF_INCOMPLETE;
5591 spin_lock(&lfsck->li_lock);
5592 list_add_tail(&com->lc_link, &lfsck->li_list_scan);
5593 spin_unlock(&lfsck->li_lock);
5597 if (lo->ll_flags & LF_CRASHED_LASTID) {
5598 LASSERT(lfsck->li_out_notify != NULL);
5600 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
5601 LE_LASTID_REBUILDING);
5607 if (root != NULL && !IS_ERR(root))
5608 lfsck_object_put(env, root);
5611 lfsck_component_cleanup(env, com);
5612 CERROR("%s: fail to init layout LFSCK component: rc = %d\n",
5613 lfsck_lfsck2name(lfsck), rc);
5619 struct lfsck_orphan_it {
5620 struct lfsck_component *loi_com;
5621 struct lfsck_rbtree_node *loi_lrn;
5622 struct lfsck_layout_slave_target *loi_llst;
5623 struct lu_fid loi_key;
5624 struct lu_orphan_rec loi_rec;
5626 unsigned int loi_over:1;
5629 static int lfsck_fid_match_idx(const struct lu_env *env,
5630 struct lfsck_instance *lfsck,
5631 const struct lu_fid *fid, int idx)
5633 struct seq_server_site *ss;
5634 struct lu_server_fld *sf;
5635 struct lu_seq_range *range = &lfsck_env_info(env)->lti_range;
5638 /* All abnormal cases will be returned to MDT0. */
5639 if (!fid_is_norm(fid)) {
5646 ss = lfsck_dev_site(lfsck);
5647 if (unlikely(ss == NULL))
5650 sf = ss->ss_server_fld;
5651 LASSERT(sf != NULL);
5653 fld_range_set_any(range);
5654 rc = fld_server_lookup(env, sf, fid_seq(fid), range);
5658 if (!fld_range_is_mdt(range))
5661 if (range->lsr_index == idx)
5667 static void lfsck_layout_destroy_orphan(const struct lu_env *env,
5668 struct dt_object *obj)
5670 struct dt_device *dev = lfsck_obj2dev(obj);
5671 struct thandle *handle;
5675 handle = dt_trans_create(env, dev);
5679 rc = dt_declare_ref_del(env, obj, handle);
5683 rc = dt_declare_destroy(env, obj, handle);
5687 rc = dt_trans_start_local(env, dev, handle);
5691 dt_write_lock(env, obj, 0);
5692 rc = dt_ref_del(env, obj, handle);
5694 rc = dt_destroy(env, obj, handle);
5695 dt_write_unlock(env, obj);
5700 dt_trans_stop(env, dev, handle);
5702 CDEBUG(D_LFSCK, "destroy orphan OST-object "DFID": rc = %d\n",
5703 PFID(lfsck_dto2fid(obj)), rc);
5708 static int lfsck_orphan_index_lookup(const struct lu_env *env,
5709 struct dt_object *dt,
5711 const struct dt_key *key)
5716 static int lfsck_orphan_index_declare_insert(const struct lu_env *env,
5717 struct dt_object *dt,
5718 const struct dt_rec *rec,
5719 const struct dt_key *key,
5720 struct thandle *handle)
5725 static int lfsck_orphan_index_insert(const struct lu_env *env,
5726 struct dt_object *dt,
5727 const struct dt_rec *rec,
5728 const struct dt_key *key,
5729 struct thandle *handle,
5735 static int lfsck_orphan_index_declare_delete(const struct lu_env *env,
5736 struct dt_object *dt,
5737 const struct dt_key *key,
5738 struct thandle *handle)
5743 static int lfsck_orphan_index_delete(const struct lu_env *env,
5744 struct dt_object *dt,
5745 const struct dt_key *key,
5746 struct thandle *handle)
5751 static struct dt_it *lfsck_orphan_it_init(const struct lu_env *env,
5752 struct dt_object *dt,
5755 struct dt_device *dev = lu2dt_dev(dt->do_lu.lo_dev);
5756 struct lfsck_instance *lfsck;
5757 struct lfsck_component *com = NULL;
5758 struct lfsck_layout_slave_data *llsd;
5759 struct lfsck_orphan_it *it = NULL;
5760 struct lfsck_layout *lo;
5764 lfsck = lfsck_instance_find(dev, true, false);
5765 if (unlikely(lfsck == NULL))
5766 RETURN(ERR_PTR(-ENXIO));
5768 com = lfsck_component_find(lfsck, LFSCK_TYPE_LAYOUT);
5769 if (unlikely(com == NULL))
5770 GOTO(out, rc = -ENOENT);
5772 lo = com->lc_file_ram;
5773 if (lo->ll_flags & LF_INCOMPLETE)
5774 GOTO(out, rc = -ESRCH);
5776 llsd = com->lc_data;
5777 if (!llsd->llsd_rbtree_valid)
5778 GOTO(out, rc = -ESRCH);
5782 GOTO(out, rc = -ENOMEM);
5784 it->loi_llst = lfsck_layout_llst_find_and_del(llsd, attr, false);
5785 if (it->loi_llst == NULL)
5786 GOTO(out, rc = -ENXIO);
5788 if (dev->dd_record_fid_accessed) {
5789 /* The first iteration against the rbtree, scan the whole rbtree
5790 * to remove the nodes which do NOT need to be handled. */
5791 write_lock(&llsd->llsd_rb_lock);
5792 if (dev->dd_record_fid_accessed) {
5793 struct rb_node *node;
5794 struct rb_node *next;
5795 struct lfsck_rbtree_node *lrn;
5797 /* No need to record the fid accessing anymore. */
5798 dev->dd_record_fid_accessed = 0;
5800 node = rb_first(&llsd->llsd_rb_root);
5801 while (node != NULL) {
5802 next = rb_next(node);
5803 lrn = rb_entry(node, struct lfsck_rbtree_node,
5805 if (atomic_read(&lrn->lrn_known_count) <=
5806 atomic_read(&lrn->lrn_accessed_count)) {
5807 rb_erase(node, &llsd->llsd_rb_root);
5808 lfsck_rbtree_free(lrn);
5813 write_unlock(&llsd->llsd_rb_lock);
5816 /* read lock the rbtree when init, and unlock when fini */
5817 read_lock(&llsd->llsd_rb_lock);
5825 lfsck_component_put(env, com);
5827 CDEBUG(D_LFSCK, "%s: init the orphan iteration: rc = %d\n",
5828 lfsck_lfsck2name(lfsck), rc);
5830 lfsck_instance_put(env, lfsck);
5835 it = (struct lfsck_orphan_it *)ERR_PTR(rc);
5838 return (struct dt_it *)it;
5841 static void lfsck_orphan_it_fini(const struct lu_env *env,
5844 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5845 struct lfsck_component *com = it->loi_com;
5846 struct lfsck_layout_slave_data *llsd;
5847 struct lfsck_layout_slave_target *llst;
5850 CDEBUG(D_LFSCK, "%s: fini the orphan iteration\n",
5851 lfsck_lfsck2name(com->lc_lfsck));
5853 llsd = com->lc_data;
5854 read_unlock(&llsd->llsd_rb_lock);
5855 llst = it->loi_llst;
5856 LASSERT(llst != NULL);
5858 /* Save the key and hash for iterate next. */
5859 llst->llst_fid = it->loi_key;
5860 llst->llst_hash = it->loi_hash;
5861 lfsck_layout_llst_put(llst);
5862 lfsck_component_put(env, com);
5868 * \retval +1: the iteration finished
5869 * \retval 0: on success, not finished
5870 * \retval -ve: on error
5872 static int lfsck_orphan_it_next(const struct lu_env *env,
5875 struct lfsck_thread_info *info = lfsck_env_info(env);
5876 struct filter_fid_old *pfid = &info->lti_old_pfid;
5877 struct lu_attr *la = &info->lti_la;
5878 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5879 struct lu_fid *key = &it->loi_key;
5880 struct lu_orphan_rec *rec = &it->loi_rec;
5881 struct lfsck_component *com = it->loi_com;
5882 struct lfsck_instance *lfsck = com->lc_lfsck;
5883 struct lfsck_layout_slave_data *llsd = com->lc_data;
5884 struct dt_object *obj;
5885 struct lfsck_rbtree_node *lrn;
5889 __u32 idx = it->loi_llst->llst_index;
5899 lrn = lfsck_rbtree_search(llsd, key, &exact);
5907 key->f_seq = lrn->lrn_seq;
5908 key->f_oid = lrn->lrn_first_oid;
5913 if (unlikely(key->f_oid == 0)) {
5920 lrn->lrn_first_oid + LFSCK_RBTREE_BITMAP_WIDTH) {
5926 if (unlikely(atomic_read(&lrn->lrn_known_count) <=
5927 atomic_read(&lrn->lrn_accessed_count))) {
5928 struct rb_node *next = rb_next(&lrn->lrn_node);
5930 while (next != NULL) {
5931 lrn = rb_entry(next, struct lfsck_rbtree_node,
5933 if (atomic_read(&lrn->lrn_known_count) >
5934 atomic_read(&lrn->lrn_accessed_count))
5936 next = rb_next(next);
5945 key->f_seq = lrn->lrn_seq;
5946 key->f_oid = lrn->lrn_first_oid;
5950 pos = key->f_oid - lrn->lrn_first_oid;
5953 pos = find_next_bit(lrn->lrn_known_bitmap,
5954 LFSCK_RBTREE_BITMAP_WIDTH, pos);
5955 if (pos >= LFSCK_RBTREE_BITMAP_WIDTH) {
5956 key->f_oid = lrn->lrn_first_oid + pos;
5957 if (unlikely(key->f_oid < lrn->lrn_first_oid)) {
5965 if (test_bit(pos, lrn->lrn_accessed_bitmap)) {
5970 key->f_oid = lrn->lrn_first_oid + pos;
5971 obj = lfsck_object_find_bottom(env, lfsck, key);
5974 if (rc == -ENOENT) {
5981 dt_read_lock(env, obj, 0);
5982 if (dt_object_exists(obj) == 0 ||
5983 lfsck_is_dead_obj(obj)) {
5984 dt_read_unlock(env, obj);
5985 lfsck_object_put(env, obj);
5990 rc = dt_attr_get(env, obj, la);
5994 rc = dt_xattr_get(env, obj, lfsck_buf_get(env, pfid, sizeof(*pfid)),
5996 if (rc == -ENODATA) {
5997 /* For the pre-created OST-object, update the bitmap to avoid
5998 * others LFSCK (second phase) iteration to touch it again. */
5999 if (la->la_ctime == 0) {
6000 if (!test_and_set_bit(pos, lrn->lrn_accessed_bitmap))
6001 atomic_inc(&lrn->lrn_accessed_count);
6003 /* For the race between repairing dangling referenced
6004 * MDT-object and unlink the file, it may left orphan
6005 * OST-object there. Destroy it now! */
6006 if (unlikely(!(la->la_mode & S_ISUID))) {
6007 dt_read_unlock(env, obj);
6008 lfsck_layout_destroy_orphan(env, obj);
6009 lfsck_object_put(env, obj);
6013 } else if (idx == 0) {
6014 /* If the orphan OST-object has no parent information,
6015 * regard it as referenced by the MDT-object on MDT0. */
6016 fid_zero(&rec->lor_fid);
6017 rec->lor_uid = la->la_uid;
6018 rec->lor_gid = la->la_gid;
6022 dt_read_unlock(env, obj);
6023 lfsck_object_put(env, obj);
6031 if (rc != sizeof(struct filter_fid) &&
6032 rc != sizeof(struct filter_fid_old))
6033 GOTO(out, rc = -EINVAL);
6035 fid_le_to_cpu(&rec->lor_fid, &pfid->ff_parent);
6036 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
6037 * MDT-object's FID::f_ver, instead it is the OST-object index in its
6038 * parent MDT-object's layout EA. */
6039 save = rec->lor_fid.f_stripe_idx;
6040 rec->lor_fid.f_ver = 0;
6041 rc = lfsck_fid_match_idx(env, lfsck, &rec->lor_fid, idx);
6042 /* If the orphan OST-object does not claim the MDT, then next.
6044 * If we do not know whether it matches or not, then return it
6045 * to the MDT for further check. */
6047 dt_read_unlock(env, obj);
6048 lfsck_object_put(env, obj);
6053 rec->lor_fid.f_stripe_idx = save;
6054 rec->lor_uid = la->la_uid;
6055 rec->lor_gid = la->la_gid;
6057 CDEBUG(D_LFSCK, "%s: return orphan "DFID", PFID "DFID", owner %u:%u\n",
6058 lfsck_lfsck2name(com->lc_lfsck), PFID(key), PFID(&rec->lor_fid),
6059 rec->lor_uid, rec->lor_gid);
6064 dt_read_unlock(env, obj);
6065 lfsck_object_put(env, obj);
6073 * \retval +1: locate to the exactly position
6074 * \retval 0: cannot locate to the exactly position,
6075 * call next() to move to a valid position.
6076 * \retval -ve: on error
6078 static int lfsck_orphan_it_get(const struct lu_env *env,
6080 const struct dt_key *key)
6082 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6085 it->loi_key = *(struct lu_fid *)key;
6086 rc = lfsck_orphan_it_next(env, di);
6096 static void lfsck_orphan_it_put(const struct lu_env *env,
6101 static struct dt_key *lfsck_orphan_it_key(const struct lu_env *env,
6102 const struct dt_it *di)
6104 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6106 return (struct dt_key *)&it->loi_key;
6109 static int lfsck_orphan_it_key_size(const struct lu_env *env,
6110 const struct dt_it *di)
6112 return sizeof(struct lu_fid);
6115 static int lfsck_orphan_it_rec(const struct lu_env *env,
6116 const struct dt_it *di,
6120 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6122 *(struct lu_orphan_rec *)rec = it->loi_rec;
6127 static __u64 lfsck_orphan_it_store(const struct lu_env *env,
6128 const struct dt_it *di)
6130 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6132 return it->loi_hash;
6136 * \retval +1: locate to the exactly position
6137 * \retval 0: cannot locate to the exactly position,
6138 * call next() to move to a valid position.
6139 * \retval -ve: on error
6141 static int lfsck_orphan_it_load(const struct lu_env *env,
6142 const struct dt_it *di,
6145 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6146 struct lfsck_layout_slave_target *llst = it->loi_llst;
6149 LASSERT(llst != NULL);
6151 if (hash != llst->llst_hash) {
6152 CDEBUG(D_LFSCK, "%s: the given hash "LPU64" for orphan "
6153 "iteration does not match the one when fini "
6154 LPU64", to be reset.\n",
6155 lfsck_lfsck2name(it->loi_com->lc_lfsck), hash,
6157 fid_zero(&llst->llst_fid);
6158 llst->llst_hash = 0;
6161 it->loi_key = llst->llst_fid;
6162 it->loi_hash = llst->llst_hash;
6163 rc = lfsck_orphan_it_next(env, (struct dt_it *)di);
6173 static int lfsck_orphan_it_key_rec(const struct lu_env *env,
6174 const struct dt_it *di,
6180 const struct dt_index_operations lfsck_orphan_index_ops = {
6181 .dio_lookup = lfsck_orphan_index_lookup,
6182 .dio_declare_insert = lfsck_orphan_index_declare_insert,
6183 .dio_insert = lfsck_orphan_index_insert,
6184 .dio_declare_delete = lfsck_orphan_index_declare_delete,
6185 .dio_delete = lfsck_orphan_index_delete,
6187 .init = lfsck_orphan_it_init,
6188 .fini = lfsck_orphan_it_fini,
6189 .get = lfsck_orphan_it_get,
6190 .put = lfsck_orphan_it_put,
6191 .next = lfsck_orphan_it_next,
6192 .key = lfsck_orphan_it_key,
6193 .key_size = lfsck_orphan_it_key_size,
6194 .rec = lfsck_orphan_it_rec,
6195 .store = lfsck_orphan_it_store,
6196 .load = lfsck_orphan_it_load,
6197 .key_rec = lfsck_orphan_it_key_rec,