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 = lfsck_ltd2tgt(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 /* XXX: If there is something worng or it needs to repair nothing,
1384 * then notify the lower to stop the modification. Currently,
1385 * we use th_result for such purpose, that may be replaced by
1386 * some rollback mechanism in the future. */
1387 handle->th_result = result;
1388 rc = dt_trans_stop(env, dev, handle);
1390 return result > 0 ? 0 : result;
1392 return rc == 0 ? 1 : rc;
1396 * Get the system default stripe size.
1398 * \param[in] env pointer to the thread context
1399 * \param[in] lfsck pointer to the lfsck instance
1400 * \param[out] size pointer to the default stripe size
1402 * \retval 0 for success
1403 * \retval negative error number on failure
1405 static int lfsck_layout_get_def_stripesize(const struct lu_env *env,
1406 struct lfsck_instance *lfsck,
1409 struct lov_user_md *lum = &lfsck_env_info(env)->lti_lum;
1410 struct dt_object *root;
1413 root = dt_locate(env, lfsck->li_next, &lfsck->li_local_root_fid);
1415 return PTR_ERR(root);
1417 /* Get the default stripe size via xattr_get on the backend root. */
1418 rc = dt_xattr_get(env, root, lfsck_buf_get(env, lum, sizeof(*lum)),
1421 /* The lum->lmm_stripe_size is LE mode. The *size also
1422 * should be LE mode. So it is unnecessary to convert. */
1423 *size = lum->lmm_stripe_size;
1425 } else if (unlikely(rc == 0)) {
1429 lfsck_object_put(env, root);
1435 * \retval +1: repaired
1436 * \retval 0: did nothing
1437 * \retval -ve: on error
1439 static int lfsck_layout_refill_lovea(const struct lu_env *env,
1440 struct thandle *handle,
1441 struct dt_object *parent,
1442 struct lu_fid *cfid,
1444 struct lov_ost_data_v1 *slot,
1445 int fl, __u32 ost_idx)
1447 struct ost_id *oi = &lfsck_env_info(env)->lti_oi;
1448 struct lov_mds_md_v1 *lmm = buf->lb_buf;
1449 struct lu_buf ea_buf;
1455 magic = le32_to_cpu(lmm->lmm_magic);
1456 count = le16_to_cpu(lmm->lmm_stripe_count);
1458 fid_to_ostid(cfid, oi);
1459 ostid_cpu_to_le(oi, &slot->l_ost_oi);
1460 slot->l_ost_gen = cpu_to_le32(0);
1461 slot->l_ost_idx = cpu_to_le32(ost_idx);
1463 if (le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_HOLE) {
1464 struct lov_ost_data_v1 *objs;
1467 if (magic == LOV_MAGIC_V1)
1468 objs = &lmm->lmm_objects[0];
1470 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1471 for (i = 0; i < count; i++, objs++) {
1472 if (objs != slot && lovea_slot_is_dummy(objs))
1476 /* If the @slot is the last dummy slot to be refilled,
1477 * then drop LOV_PATTERN_F_HOLE from lmm::lmm_pattern. */
1479 lmm->lmm_pattern &= ~cpu_to_le32(LOV_PATTERN_F_HOLE);
1482 lfsck_buf_init(&ea_buf, lmm, lov_mds_md_size(count, magic));
1483 rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV, fl, handle);
1491 * \retval +1: repaired
1492 * \retval 0: did nothing
1493 * \retval -ve: on error
1495 static int lfsck_layout_extend_lovea(const struct lu_env *env,
1496 struct lfsck_instance *lfsck,
1497 struct thandle *handle,
1498 struct dt_object *parent,
1499 struct lu_fid *cfid,
1500 struct lu_buf *buf, int fl,
1501 __u32 ost_idx, __u32 ea_off, bool reset)
1503 struct lov_mds_md_v1 *lmm = buf->lb_buf;
1504 struct lov_ost_data_v1 *objs;
1510 if (fl == LU_XATTR_CREATE || reset) {
1511 __u32 pattern = LOV_PATTERN_RAID0;
1514 LASSERT(buf->lb_len >= lov_mds_md_size(count, LOV_MAGIC_V1));
1516 if (ea_off != 0 || reset) {
1517 pattern |= LOV_PATTERN_F_HOLE;
1521 memset(lmm, 0, buf->lb_len);
1522 lmm->lmm_magic = cpu_to_le32(LOV_MAGIC_V1);
1523 lmm->lmm_pattern = cpu_to_le32(pattern);
1524 fid_to_lmm_oi(lfsck_dto2fid(parent), &lmm->lmm_oi);
1525 lmm_oi_cpu_to_le(&lmm->lmm_oi, &lmm->lmm_oi);
1527 rc = lfsck_layout_get_def_stripesize(env, lfsck,
1528 &lmm->lmm_stripe_size);
1532 objs = &lmm->lmm_objects[ea_off];
1534 __u32 magic = le32_to_cpu(lmm->lmm_magic);
1537 count = le16_to_cpu(lmm->lmm_stripe_count);
1538 if (magic == LOV_MAGIC_V1)
1539 objs = &lmm->lmm_objects[count];
1541 objs = &((struct lov_mds_md_v3 *)lmm)->
1544 gap = ea_off - count;
1547 LASSERT(buf->lb_len >= lov_mds_md_size(count, magic));
1550 memset(objs, 0, gap * sizeof(*objs));
1551 lmm->lmm_pattern |= cpu_to_le32(LOV_PATTERN_F_HOLE);
1555 lmm->lmm_layout_gen =
1556 cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
1560 lmm->lmm_stripe_count = cpu_to_le16(count);
1561 rc = lfsck_layout_refill_lovea(env, handle, parent, cfid, buf, objs,
1564 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant extend layout EA for "
1565 DFID": parent "DFID", OST-index %u, stripe-index %u, fl %d, "
1566 "reset %s, %s LOV EA hole: rc = %d\n",
1567 lfsck_lfsck2name(lfsck), PFID(cfid), PFID(lfsck_dto2fid(parent)),
1568 ost_idx, ea_off, fl, reset ? "yes" : "no",
1569 hole ? "with" : "without", rc);
1574 static int __lfsck_layout_update_pfid(const struct lu_env *env,
1575 struct dt_object *child,
1576 const struct lu_fid *pfid, __u32 offset)
1578 struct dt_device *dev = lfsck_obj2dev(child);
1579 struct filter_fid *ff = &lfsck_env_info(env)->lti_new_pfid;
1580 struct thandle *handle;
1581 struct lu_buf buf = { NULL };
1584 ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
1585 ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
1586 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
1587 * MDT-object's FID::f_ver, instead it is the OST-object index in its
1588 * parent MDT-object's layout EA. */
1589 ff->ff_parent.f_stripe_idx = cpu_to_le32(offset);
1590 lfsck_buf_init(&buf, ff, sizeof(struct filter_fid));
1592 handle = dt_trans_create(env, dev);
1594 RETURN(PTR_ERR(handle));
1596 rc = dt_declare_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
1600 rc = dt_trans_start_local(env, dev, handle);
1604 rc = dt_xattr_set(env, child, &buf, XATTR_NAME_FID, 0, handle);
1609 dt_trans_stop(env, dev, handle);
1615 * \retval +1: repaired
1616 * \retval 0: did nothing
1617 * \retval -ve: on error
1619 static int lfsck_layout_update_pfid(const struct lu_env *env,
1620 struct lfsck_component *com,
1621 struct dt_object *parent,
1622 struct lu_fid *cfid,
1623 struct dt_device *cdev, __u32 ea_off)
1625 struct dt_object *child;
1629 child = lfsck_object_find_by_dev(env, cdev, cfid);
1631 RETURN(PTR_ERR(child));
1633 rc = __lfsck_layout_update_pfid(env, child,
1634 lu_object_fid(&parent->do_lu), ea_off);
1635 lfsck_object_put(env, child);
1637 RETURN(rc == 0 ? 1 : rc);
1641 * This function will create the MDT-object with the given (partial) LOV EA.
1643 * Under some data corruption cases, the MDT-object of the file may be lost,
1644 * but its OST-objects, or some of them are there. The layout LFSCK needs to
1645 * re-create the MDT-object with the orphan OST-object(s) information.
1647 * On the other hand, the LFSCK may has created some OST-object for repairing
1648 * dangling LOV EA reference, but as the LFSCK processing, it may find that
1649 * the old OST-object is there and should replace the former new created OST
1650 * object. Unfortunately, some others have modified such newly created object.
1651 * To keep the data (both new and old), the LFSCK will create MDT-object with
1652 * new FID to reference the original OST-object.
1654 * \param[in] env pointer to the thread context
1655 * \param[in] com pointer to the lfsck component
1656 * \param[in] ltd pointer to target device descriptor
1657 * \param[in] rec pointer to the record for the orphan OST-object
1658 * \param[in] cfid pointer to FID for the orphan OST-object
1659 * \param[in] infix additional information, such as the FID for original
1660 * MDT-object and the stripe offset in the LOV EA
1661 * \param[in] type the type for describing why the orphan MDT-object is
1662 * created. The rules are as following:
1664 * type "C": Multiple OST-objects claim the same MDT-object and the
1665 * same slot in the layout EA. Then the LFSCK will create
1666 * new MDT-object(s) to hold the conflict OST-object(s).
1668 * type "N": The orphan OST-object does not know which one was the
1669 * real parent MDT-object, so the LFSCK uses new FID for
1670 * its parent MDT-object.
1672 * type "R": The orphan OST-object knows its parent MDT-object FID,
1673 * but does not know the position (the file name) in the
1676 * type "D": The MDT-object is a directory, it may knows its parent
1677 * but because there is no valid linkEA, the LFSCK cannot
1678 * know where to put it back to the namespace.
1679 * type "O": The MDT-object has no linkEA, and there is no name
1680 * entry that references the MDT-object.
1682 * type "P": The orphan object to be created was a parent directory
1683 * of some MDT-object which linkEA shows that the @orphan
1684 * object is missing.
1686 * The orphan name will be like:
1687 * ${FID}-${infix}-${type}-${conflict_version}
1689 * \param[in] ea_off the stripe offset in the LOV EA
1691 * \retval positive on repaired something
1692 * \retval 0 if needs to repair nothing
1693 * \retval negative error number on failure
1695 static int lfsck_layout_recreate_parent(const struct lu_env *env,
1696 struct lfsck_component *com,
1697 struct lfsck_tgt_desc *ltd,
1698 struct lu_orphan_rec *rec,
1699 struct lu_fid *cfid,
1704 struct lfsck_thread_info *info = lfsck_env_info(env);
1705 struct dt_insert_rec *dtrec = &info->lti_dt_rec;
1706 char *name = info->lti_key;
1707 struct lu_attr *la = &info->lti_la2;
1708 struct dt_object_format *dof = &info->lti_dof;
1709 struct lfsck_instance *lfsck = com->lc_lfsck;
1710 struct lu_fid *pfid = &rec->lor_fid;
1711 struct lu_fid *tfid = &info->lti_fid3;
1712 struct dt_device *dev = lfsck->li_bottom;
1713 struct dt_object *lpf = lfsck->li_lpf_obj;
1714 struct dt_object *pobj = NULL;
1715 struct dt_object *cobj = NULL;
1716 struct thandle *th = NULL;
1717 struct lu_buf *ea_buf = &info->lti_big_buf;
1718 struct lu_buf lov_buf;
1719 struct lfsck_lock_handle *llh = &info->lti_llh;
1720 struct linkea_data ldata = { NULL };
1721 struct lu_buf linkea_buf;
1722 const struct lu_name *pname;
1728 if (unlikely(lpf == NULL))
1729 GOTO(log, rc = -ENXIO);
1731 /* We use two separated transactions to repair the inconsistency.
1733 * 1) create the MDT-object locally.
1734 * 2) update the OST-object's PFID EA if necessary.
1736 * If 1) succeed, but 2) failed, then the OST-object's PFID EA will be
1737 * updated when the layout LFSCK run next time.
1739 * If 1) failed, but 2) succeed, then such MDT-object will be re-created
1740 * when the layout LFSCK run next time. */
1742 if (fid_is_zero(pfid)) {
1743 rc = lfsck_fid_alloc(env, lfsck, pfid, false);
1747 cobj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, cfid);
1749 GOTO(log, rc = PTR_ERR(cobj));
1752 pobj = lfsck_object_find_by_dev(env, dev, pfid);
1754 GOTO(log, rc = PTR_ERR(pobj));
1756 LASSERT(infix != NULL);
1757 LASSERT(type != NULL);
1759 memset(la, 0, sizeof(*la));
1760 la->la_uid = rec->lor_uid;
1761 la->la_gid = rec->lor_gid;
1762 la->la_mode = S_IFREG | S_IRUSR;
1763 la->la_valid = LA_MODE | LA_UID | LA_GID;
1765 memset(dof, 0, sizeof(*dof));
1766 dof->dof_type = dt_mode_to_dft(S_IFREG);
1767 /* Because the dof->dof_reg.striped = 0, the LOD will not create
1768 * the stripe(s). The LFSCK will specify the LOV EA via
1769 * lfsck_layout_extend_lovea(). */
1771 size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
1772 if (ea_buf->lb_len < size) {
1773 lu_buf_realloc(ea_buf, size);
1774 if (ea_buf->lb_buf == NULL)
1775 GOTO(log, rc = -ENOMEM);
1780 snprintf(name, NAME_MAX, DFID"%s-%s-%d", PFID(pfid), infix,
1782 rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
1783 (const struct dt_key *)name);
1784 if (rc != 0 && rc != -ENOENT)
1788 rc = lfsck_lock(env, lfsck, lfsck->li_lpf_obj, name, llh,
1789 MDS_INODELOCK_UPDATE, LCK_PW);
1793 /* Re-check whether the name conflict with othrs after taken
1795 rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
1796 (const struct dt_key *)name);
1797 if (unlikely(rc == 0)) {
1805 rc = linkea_data_new(&ldata,
1806 &lfsck_env_info(env)->lti_linkea_buf);
1810 pname = lfsck_name_get_const(env, name, strlen(name));
1811 rc = linkea_add_buf(&ldata, pname, lfsck_dto2fid(lfsck->li_lpf_obj));
1815 /* The 1st transaction. */
1816 th = dt_trans_create(env, dev);
1818 GOTO(unlock, rc = PTR_ERR(th));
1820 rc = dt_declare_create(env, pobj, la, NULL, dof, th);
1824 lfsck_buf_init(&lov_buf, ea_buf->lb_buf, size);
1825 rc = dt_declare_xattr_set(env, pobj, &lov_buf, XATTR_NAME_LOV,
1826 LU_XATTR_REPLACE, th);
1830 dtrec->rec_fid = pfid;
1831 dtrec->rec_type = S_IFREG;
1832 rc = dt_declare_insert(env, lpf,
1833 (const struct dt_rec *)dtrec,
1834 (const struct dt_key *)name, th);
1838 lfsck_buf_init(&linkea_buf, ldata.ld_buf->lb_buf,
1839 ldata.ld_leh->leh_len);
1840 rc = dt_declare_xattr_set(env, pobj, &linkea_buf,
1841 XATTR_NAME_LINK, 0, th);
1845 rc = dt_trans_start_local(env, dev, th);
1849 dt_write_lock(env, pobj, 0);
1850 rc = dt_create(env, pobj, la, NULL, dof, th);
1852 rc = lfsck_layout_extend_lovea(env, lfsck, th, pobj, cfid,
1853 &lov_buf, 0, ltd->ltd_index, ea_off, true);
1854 dt_write_unlock(env, pobj);
1858 rc = dt_insert(env, lpf, (const struct dt_rec *)dtrec,
1859 (const struct dt_key *)name, th, 1);
1863 rc = dt_xattr_set(env, pobj, &linkea_buf, XATTR_NAME_LINK, 0, th);
1864 if (rc == 0 && cobj != NULL) {
1865 dt_trans_stop(env, dev, th);
1868 /* The 2nd transaction. */
1869 rc = __lfsck_layout_update_pfid(env, cobj, pfid, ea_off);
1876 dt_trans_stop(env, dev, th);
1882 if (cobj != NULL && !IS_ERR(cobj))
1883 lfsck_object_put(env, cobj);
1884 if (pobj != NULL && !IS_ERR(pobj))
1885 lfsck_object_put(env, pobj);
1888 CDEBUG(D_LFSCK, "%s layout LFSCK assistant failed to "
1889 "recreate the lost MDT-object: parent "DFID
1890 ", child "DFID", OST-index %u, stripe-index %u, "
1891 "infix %s, type %s: rc = %d\n",
1892 lfsck_lfsck2name(lfsck), PFID(pfid), PFID(cfid),
1893 ltd->ltd_index, ea_off, infix, type, rc);
1895 return rc >= 0 ? 1 : rc;
1898 static int lfsck_layout_master_conditional_destroy(const struct lu_env *env,
1899 struct lfsck_component *com,
1900 const struct lu_fid *fid,
1903 struct lfsck_thread_info *info = lfsck_env_info(env);
1904 struct lfsck_request *lr = &info->lti_lr;
1905 struct lfsck_instance *lfsck = com->lc_lfsck;
1906 struct lfsck_tgt_desc *ltd;
1907 struct ptlrpc_request *req;
1908 struct lfsck_request *tmp;
1909 struct obd_export *exp;
1913 ltd = lfsck_tgt_get(&lfsck->li_ost_descs, index);
1914 if (unlikely(ltd == NULL))
1918 if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
1919 GOTO(put, rc = -EOPNOTSUPP);
1921 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
1923 GOTO(put, rc = -ENOMEM);
1925 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
1927 ptlrpc_request_free(req);
1932 memset(lr, 0, sizeof(*lr));
1933 lr->lr_event = LE_CONDITIONAL_DESTROY;
1934 lr->lr_active = LFSCK_TYPE_LAYOUT;
1937 tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
1939 ptlrpc_request_set_replen(req);
1941 rc = ptlrpc_queue_wait(req);
1942 ptlrpc_req_finished(req);
1952 static int lfsck_layout_slave_conditional_destroy(const struct lu_env *env,
1953 struct lfsck_component *com,
1954 struct lfsck_request *lr)
1956 struct lfsck_thread_info *info = lfsck_env_info(env);
1957 struct lu_attr *la = &info->lti_la;
1958 ldlm_policy_data_t *policy = &info->lti_policy;
1959 struct ldlm_res_id *resid = &info->lti_resid;
1960 struct lfsck_instance *lfsck = com->lc_lfsck;
1961 struct dt_device *dev = lfsck->li_bottom;
1962 struct lu_fid *fid = &lr->lr_fid;
1963 struct dt_object *obj;
1964 struct thandle *th = NULL;
1965 struct lustre_handle lh = { 0 };
1970 obj = lfsck_object_find_by_dev(env, dev, fid);
1972 RETURN(PTR_ERR(obj));
1974 dt_read_lock(env, obj, 0);
1975 if (dt_object_exists(obj) == 0 ||
1976 lfsck_is_dead_obj(obj)) {
1977 dt_read_unlock(env, obj);
1979 GOTO(put, rc = -ENOENT);
1982 /* Get obj's attr without lock firstly. */
1983 rc = dt_attr_get(env, obj, la);
1984 dt_read_unlock(env, obj);
1988 if (likely(la->la_ctime != 0 || la->la_mode & S_ISUID))
1989 GOTO(put, rc = -ETXTBSY);
1991 /* Acquire extent lock on [0, EOF] to sync with all possible written. */
1992 LASSERT(lfsck->li_namespace != NULL);
1994 memset(policy, 0, sizeof(*policy));
1995 policy->l_extent.end = OBD_OBJECT_EOF;
1996 ost_fid_build_resid(fid, resid);
1997 rc = ldlm_cli_enqueue_local(lfsck->li_namespace, resid, LDLM_EXTENT,
1998 policy, LCK_EX, &flags, ldlm_blocking_ast,
1999 ldlm_completion_ast, NULL, NULL, 0,
2000 LVB_T_NONE, NULL, &lh);
2002 GOTO(put, rc = -EIO);
2004 dt_write_lock(env, obj, 0);
2005 /* Get obj's attr within lock again. */
2006 rc = dt_attr_get(env, obj, la);
2010 if (la->la_ctime != 0)
2011 GOTO(unlock, rc = -ETXTBSY);
2013 th = dt_trans_create(env, dev);
2015 GOTO(unlock, rc = PTR_ERR(th));
2017 rc = dt_declare_ref_del(env, obj, th);
2021 rc = dt_declare_destroy(env, obj, th);
2025 rc = dt_trans_start_local(env, dev, th);
2029 rc = dt_ref_del(env, obj, th);
2033 rc = dt_destroy(env, obj, th);
2035 CDEBUG(D_LFSCK, "%s: layout LFSCK destroyed the empty "
2036 "OST-object "DFID" that was created for reparing "
2037 "dangling referenced case. But the original missing "
2038 "OST-object is found now.\n",
2039 lfsck_lfsck2name(lfsck), PFID(fid));
2044 dt_trans_stop(env, dev, th);
2047 dt_write_unlock(env, obj);
2048 ldlm_lock_decref(&lh, LCK_EX);
2051 lfsck_object_put(env, obj);
2057 * Some OST-object has occupied the specified layout EA slot.
2058 * Such OST-object may be generated by the LFSCK when repair
2059 * dangling referenced MDT-object, which can be indicated by
2060 * attr::la_ctime == 0 but without S_ISUID in la_mode. If it
2061 * is true and such OST-object has not been modified yet, we
2062 * will replace it with the orphan OST-object; otherwise the
2063 * LFSCK will create new MDT-object to reference the orphan.
2065 * \retval +1: repaired
2066 * \retval 0: did nothing
2067 * \retval -ve: on error
2069 static int lfsck_layout_conflict_create(const struct lu_env *env,
2070 struct lfsck_component *com,
2071 struct lfsck_tgt_desc *ltd,
2072 struct lu_orphan_rec *rec,
2073 struct dt_object *parent,
2074 struct lu_fid *cfid,
2075 struct lu_buf *ea_buf,
2076 struct lov_ost_data_v1 *slot,
2079 struct lfsck_thread_info *info = lfsck_env_info(env);
2080 struct lu_fid *cfid2 = &info->lti_fid2;
2081 struct ost_id *oi = &info->lti_oi;
2082 struct lov_mds_md_v1 *lmm = ea_buf->lb_buf;
2083 struct dt_device *dev = lfsck_obj2dev(parent);
2084 struct thandle *th = NULL;
2085 struct lustre_handle lh = { 0 };
2086 __u32 ost_idx2 = le32_to_cpu(slot->l_ost_idx);
2090 ostid_le_to_cpu(&slot->l_ost_oi, oi);
2091 rc = ostid_to_fid(cfid2, oi, ost_idx2);
2095 rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2096 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2101 rc = lfsck_layout_master_conditional_destroy(env, com, cfid2, ost_idx2);
2103 /* If the conflict OST-obejct is not created for fixing dangling
2104 * referenced MDT-object in former LFSCK check/repair, or it has
2105 * been modified by others, then we cannot destroy it. Re-create
2106 * a new MDT-object for the orphan OST-object. */
2107 if (rc == -ETXTBSY) {
2108 /* No need the layout lock on the original parent. */
2109 lfsck_ibits_unlock(&lh, LCK_EX);
2111 fid_zero(&rec->lor_fid);
2112 snprintf(info->lti_tmpbuf, sizeof(info->lti_tmpbuf),
2113 "-"DFID"-%x", PFID(lu_object_fid(&parent->do_lu)),
2115 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2116 info->lti_tmpbuf, "C", ea_off);
2121 if (rc != 0 && rc != -ENOENT)
2124 th = dt_trans_create(env, dev);
2126 GOTO(unlock, rc = PTR_ERR(th));
2128 rc = dt_declare_xattr_set(env, parent, ea_buf, XATTR_NAME_LOV,
2129 LU_XATTR_REPLACE, th);
2133 rc = dt_trans_start_local(env, dev, th);
2137 dt_write_lock(env, parent, 0);
2138 lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2139 rc = lfsck_layout_refill_lovea(env, th, parent, cfid, ea_buf, slot,
2140 LU_XATTR_REPLACE, ltd->ltd_index);
2141 dt_write_unlock(env, parent);
2146 dt_trans_stop(env, dev, th);
2149 lfsck_ibits_unlock(&lh, LCK_EX);
2152 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant replaced the conflict "
2153 "OST-object "DFID" on the OST %x with the orphan "DFID" on "
2154 "the OST %x: parent "DFID", stripe-index %u: rc = %d\n",
2155 lfsck_lfsck2name(com->lc_lfsck), PFID(cfid2), ost_idx2,
2156 PFID(cfid), ltd->ltd_index, PFID(lfsck_dto2fid(parent)),
2159 return rc >= 0 ? 1 : rc;
2163 * \retval +1: repaired
2164 * \retval 0: did nothing
2165 * \retval -ve: on error
2167 static int lfsck_layout_recreate_lovea(const struct lu_env *env,
2168 struct lfsck_component *com,
2169 struct lfsck_tgt_desc *ltd,
2170 struct lu_orphan_rec *rec,
2171 struct dt_object *parent,
2172 struct lu_fid *cfid,
2173 __u32 ost_idx, __u32 ea_off)
2175 struct lfsck_thread_info *info = lfsck_env_info(env);
2176 struct lu_buf *buf = &info->lti_big_buf;
2177 struct lu_fid *fid = &info->lti_fid2;
2178 struct ost_id *oi = &info->lti_oi;
2179 struct lfsck_instance *lfsck = com->lc_lfsck;
2180 struct dt_device *dt = lfsck_obj2dev(parent);
2181 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
2182 struct thandle *handle = NULL;
2184 struct lov_mds_md_v1 *lmm;
2185 struct lov_ost_data_v1 *objs;
2186 struct lustre_handle lh = { 0 };
2193 bool locked = false;
2196 rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2197 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2200 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant failed to recreate "
2201 "LOV EA for "DFID": parent "DFID", OST-index %u, "
2202 "stripe-index %u: rc = %d\n",
2203 lfsck_lfsck2name(lfsck), PFID(cfid),
2204 PFID(lfsck_dto2fid(parent)), ost_idx, ea_off, rc);
2211 dt_write_unlock(env, parent);
2215 if (handle != NULL) {
2216 dt_trans_stop(env, dt, handle);
2221 GOTO(unlock_layout, rc);
2224 if (buf->lb_len < lovea_size) {
2225 lu_buf_realloc(buf, lovea_size);
2226 if (buf->lb_buf == NULL)
2227 GOTO(unlock_layout, rc = -ENOMEM);
2230 if (!(bk->lb_param & LPF_DRYRUN)) {
2231 handle = dt_trans_create(env, dt);
2233 GOTO(unlock_layout, rc = PTR_ERR(handle));
2235 rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2240 rc = dt_trans_start_local(env, dt, handle);
2245 dt_write_lock(env, parent, 0);
2247 rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV);
2248 if (rc == -ERANGE) {
2249 rc = dt_xattr_get(env, parent, &LU_BUF_NULL, XATTR_NAME_LOV);
2252 } else if (rc == -ENODATA || rc == 0) {
2253 lovea_size = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
2254 /* If the declared is not big enough, re-try. */
2255 if (buf->lb_len < lovea_size) {
2259 fl = LU_XATTR_CREATE;
2260 } else if (rc < 0) {
2261 GOTO(unlock_parent, rc);
2262 } else if (unlikely(buf->lb_len == 0)) {
2265 fl = LU_XATTR_REPLACE;
2269 if (fl == LU_XATTR_CREATE) {
2270 if (bk->lb_param & LPF_DRYRUN)
2271 GOTO(unlock_parent, rc = 1);
2273 LASSERT(buf->lb_len >= lovea_size);
2275 rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
2276 buf, fl, ost_idx, ea_off, false);
2278 GOTO(unlock_parent, rc);
2282 rc1 = lfsck_layout_verify_header(lmm);
2284 /* If the LOV EA crashed, the rebuild it. */
2285 if (rc1 == -EINVAL) {
2286 if (bk->lb_param & LPF_DRYRUN)
2287 GOTO(unlock_parent, rc = 1);
2289 LASSERT(buf->lb_len >= lovea_size);
2291 rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
2292 buf, fl, ost_idx, ea_off, true);
2294 GOTO(unlock_parent, rc);
2297 /* For other unknown magic/pattern, keep the current LOV EA. */
2299 GOTO(unlock_parent, rc = rc1);
2301 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
2302 * been verified in lfsck_layout_verify_header() already. If some
2303 * new magic introduced in the future, then layout LFSCK needs to
2304 * be updated also. */
2305 magic = le32_to_cpu(lmm->lmm_magic);
2306 if (magic == LOV_MAGIC_V1) {
2307 objs = &lmm->lmm_objects[0];
2309 LASSERT(magic == LOV_MAGIC_V3);
2310 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
2313 count = le16_to_cpu(lmm->lmm_stripe_count);
2315 GOTO(unlock_parent, rc = -EINVAL);
2318 /* Exceed the current end of MDT-object layout EA. Then extend it. */
2319 if (count <= ea_off) {
2320 if (bk->lb_param & LPF_DRYRUN)
2321 GOTO(unlock_parent, rc = 1);
2323 lovea_size = lov_mds_md_size(ea_off + 1, magic);
2324 /* If the declared is not big enough, re-try. */
2325 if (buf->lb_len < lovea_size) {
2330 rc = lfsck_layout_extend_lovea(env, lfsck, handle, parent, cfid,
2331 buf, fl, ost_idx, ea_off, false);
2333 GOTO(unlock_parent, rc);
2336 LASSERTF(rc > 0, "invalid rc = %d\n", rc);
2338 for (i = 0; i < count; i++, objs++) {
2339 /* The MDT-object was created via lfsck_layout_recover_create()
2340 * by others before, and we fill the dummy layout EA. */
2341 if (lovea_slot_is_dummy(objs)) {
2345 if (bk->lb_param & LPF_DRYRUN)
2346 GOTO(unlock_parent, rc = 1);
2348 lmm->lmm_layout_gen =
2349 cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2350 rc = lfsck_layout_refill_lovea(env, handle, parent,
2351 cfid, buf, objs, fl,
2354 CDEBUG(D_LFSCK, "%s layout LFSCK assistant fill "
2355 "dummy layout slot for "DFID": parent "DFID
2356 ", OST-index %u, stripe-index %u: rc = %d\n",
2357 lfsck_lfsck2name(lfsck), PFID(cfid),
2358 PFID(lfsck_dto2fid(parent)), ost_idx, i, rc);
2360 GOTO(unlock_parent, rc);
2363 ostid_le_to_cpu(&objs->l_ost_oi, oi);
2364 rc = ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
2366 CDEBUG(D_LFSCK, "%s: the parent "DFID" contains "
2367 "invalid layout EA at the slot %d, index %u\n",
2368 lfsck_lfsck2name(lfsck),
2369 PFID(lfsck_dto2fid(parent)), i,
2370 le32_to_cpu(objs->l_ost_idx));
2372 GOTO(unlock_parent, rc);
2375 /* It should be rare case, the slot is there, but the LFSCK
2376 * does not handle it during the first-phase cycle scanning. */
2377 if (unlikely(lu_fid_eq(fid, cfid))) {
2379 GOTO(unlock_parent, rc = 0);
2381 /* Rare case that the OST-object index
2382 * does not match the parent MDT-object
2383 * layout EA. We trust the later one. */
2384 if (bk->lb_param & LPF_DRYRUN)
2385 GOTO(unlock_parent, rc = 1);
2387 dt_write_unlock(env, parent);
2389 dt_trans_stop(env, dt, handle);
2390 lfsck_ibits_unlock(&lh, LCK_EX);
2391 rc = lfsck_layout_update_pfid(env, com, parent,
2392 cfid, ltd->ltd_tgt, i);
2394 CDEBUG(D_LFSCK, "%s layout LFSCK assistant "
2395 "updated OST-object's pfid for "DFID
2396 ": parent "DFID", OST-index %u, "
2397 "stripe-index %u: rc = %d\n",
2398 lfsck_lfsck2name(lfsck), PFID(cfid),
2399 PFID(lfsck_dto2fid(parent)),
2400 ltd->ltd_index, i, rc);
2407 /* The MDT-object exists, but related layout EA slot is occupied
2409 if (bk->lb_param & LPF_DRYRUN)
2410 GOTO(unlock_parent, rc = 1);
2412 dt_write_unlock(env, parent);
2414 dt_trans_stop(env, dt, handle);
2415 lfsck_ibits_unlock(&lh, LCK_EX);
2416 if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1)
2417 objs = &lmm->lmm_objects[ea_off];
2419 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[ea_off];
2420 rc = lfsck_layout_conflict_create(env, com, ltd, rec, parent, cfid,
2427 dt_write_unlock(env, parent);
2431 dt_trans_stop(env, dt, handle);
2434 lfsck_ibits_unlock(&lh, LCK_EX);
2439 static int lfsck_layout_scan_orphan_one(const struct lu_env *env,
2440 struct lfsck_component *com,
2441 struct lfsck_tgt_desc *ltd,
2442 struct lu_orphan_rec *rec,
2443 struct lu_fid *cfid)
2445 struct lfsck_layout *lo = com->lc_file_ram;
2446 struct lu_fid *pfid = &rec->lor_fid;
2447 struct dt_object *parent = NULL;
2448 __u32 ea_off = pfid->f_stripe_idx;
2452 if (!fid_is_sane(cfid))
2453 GOTO(out, rc = -EINVAL);
2455 if (fid_is_zero(pfid)) {
2456 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2462 if (!fid_is_sane(pfid))
2463 GOTO(out, rc = -EINVAL);
2465 parent = lfsck_object_find_by_dev(env, com->lc_lfsck->li_bottom, pfid);
2467 GOTO(out, rc = PTR_ERR(parent));
2469 if (unlikely(dt_object_remote(parent) != 0))
2470 GOTO(put, rc = -EXDEV);
2472 if (dt_object_exists(parent) == 0) {
2473 lfsck_object_put(env, parent);
2474 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2479 if (!S_ISREG(lu_object_attr(&parent->do_lu)))
2480 GOTO(put, rc = -EISDIR);
2482 rc = lfsck_layout_recreate_lovea(env, com, ltd, rec, parent, cfid,
2483 ltd->ltd_index, ea_off);
2489 lfsck_object_put(env, parent);
2491 /* The layout EA is changed, need to be reloaded next time. */
2492 lu_object_put_nocache(env, &parent->do_lu);
2495 down_write(&com->lc_sem);
2496 com->lc_new_scanned++;
2497 com->lc_new_checked++;
2499 lo->ll_objs_repaired[LLIT_ORPHAN - 1]++;
2501 } else if (rc < 0) {
2502 lo->ll_objs_failed_phase2++;
2504 up_write(&com->lc_sem);
2509 static int lfsck_layout_scan_orphan(const struct lu_env *env,
2510 struct lfsck_component *com,
2511 struct lfsck_tgt_desc *ltd)
2513 struct lfsck_assistant_data *lad = com->lc_data;
2514 struct lfsck_instance *lfsck = com->lc_lfsck;
2515 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
2516 struct lfsck_thread_info *info = lfsck_env_info(env);
2517 struct ost_id *oi = &info->lti_oi;
2518 struct lu_fid *fid = &info->lti_fid;
2519 struct dt_object *obj;
2520 const struct dt_it_ops *iops;
2525 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant starts the orphan "
2526 "scanning for OST%04x\n",
2527 lfsck_lfsck2name(lfsck), ltd->ltd_index);
2529 if (cfs_bitmap_check(lad->lad_bitmap, ltd->ltd_index)) {
2530 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant skip the orphan "
2531 "scanning for OST%04x\n",
2532 lfsck_lfsck2name(lfsck), ltd->ltd_index);
2537 ostid_set_seq(oi, FID_SEQ_IDIF);
2538 ostid_set_id(oi, 0);
2539 rc = ostid_to_fid(fid, oi, ltd->ltd_index);
2543 obj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, fid);
2544 if (unlikely(IS_ERR(obj)))
2545 GOTO(log, rc = PTR_ERR(obj));
2547 rc = obj->do_ops->do_index_try(env, obj, &dt_lfsck_orphan_features);
2551 iops = &obj->do_index_ops->dio_it;
2552 di = iops->init(env, obj, 0);
2554 GOTO(put, rc = PTR_ERR(di));
2556 rc = iops->load(env, di, 0);
2558 /* -ESRCH means that the orphan OST-objects rbtree has been
2559 * cleanup because of the OSS server restart or other errors. */
2560 lfsck_lad_set_bitmap(env, com, ltd->ltd_index);
2565 rc = iops->next(env, di);
2577 struct lu_orphan_rec *rec = &info->lti_rec;
2579 if (CFS_FAIL_TIMEOUT(OBD_FAIL_LFSCK_DELAY3, cfs_fail_val) &&
2580 unlikely(!thread_is_running(&lfsck->li_thread)))
2583 key = iops->key(env, di);
2584 com->lc_fid_latest_scanned_phase2 = *(struct lu_fid *)key;
2585 rc = iops->rec(env, di, (struct dt_rec *)rec, 0);
2587 rc = lfsck_layout_scan_orphan_one(env, com, ltd, rec,
2588 &com->lc_fid_latest_scanned_phase2);
2589 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
2592 lfsck_control_speed_by_self(com);
2594 rc = iops->next(env, di);
2595 } while (rc < 0 && !(bk->lb_param & LPF_FAILOUT));
2602 iops->fini(env, di);
2604 lfsck_object_put(env, obj);
2607 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant finished the orphan "
2608 "scanning for OST%04x: rc = %d\n",
2609 lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
2611 return rc > 0 ? 0 : rc;
2614 /* For the MDT-object with dangling reference, we need to repare the
2615 * inconsistency according to the LFSCK sponsor's requirement:
2617 * 1) Keep the inconsistency there and report the inconsistency case,
2618 * then give the chance to the application to find related issues,
2619 * and the users can make the decision about how to handle it with
2620 * more human knownledge. (by default)
2622 * 2) Re-create the missing OST-object with the FID/owner information. */
2623 static int lfsck_layout_repair_dangling(const struct lu_env *env,
2624 struct lfsck_component *com,
2625 struct dt_object *parent,
2626 struct lfsck_layout_req *llr,
2629 struct lfsck_thread_info *info = lfsck_env_info(env);
2630 struct filter_fid *pfid = &info->lti_new_pfid;
2631 struct dt_object_format *dof = &info->lti_dof;
2632 struct dt_object *child = llr->llr_child;
2633 struct dt_device *dev = lfsck_obj2dev(child);
2634 const struct lu_fid *tfid = lu_object_fid(&parent->do_lu);
2635 struct thandle *handle;
2637 struct lustre_handle lh = { 0 };
2642 if (com->lc_lfsck->li_bookmark_ram.lb_param & LPF_CREATE_OSTOBJ)
2650 rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2651 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2656 rc = dt_attr_get(env, parent, la);
2660 la->la_mode = S_IFREG | 0666;
2661 la->la_atime = la->la_mtime = la->la_ctime = 0;
2662 la->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
2663 LA_ATIME | LA_MTIME | LA_CTIME;
2664 memset(dof, 0, sizeof(*dof));
2665 pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
2666 pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
2667 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
2668 * MDT-object's FID::f_ver, instead it is the OST-object index in its
2669 * parent MDT-object's layout EA. */
2670 pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
2671 buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
2673 handle = dt_trans_create(env, dev);
2675 GOTO(unlock1, rc = PTR_ERR(handle));
2677 rc = dt_declare_create(env, child, la, NULL, dof, handle);
2681 rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID,
2682 LU_XATTR_CREATE, handle);
2686 rc = dt_trans_start_local(env, dev, handle);
2690 dt_read_lock(env, parent, 0);
2691 if (unlikely(lfsck_is_dead_obj(parent)))
2692 GOTO(unlock2, rc = 1);
2694 rc = dt_create(env, child, la, NULL, dof, handle);
2698 rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, LU_XATTR_CREATE,
2704 dt_read_unlock(env, parent);
2707 rc = lfsck_layout_trans_stop(env, dev, handle, rc);
2710 lfsck_ibits_unlock(&lh, LCK_EX);
2714 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant found "
2715 "dangling reference for: parent "DFID", child "DFID
2716 ", OST-index %u, stripe-index %u, owner %u/%u. %s: "
2717 "rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
2718 PFID(lfsck_dto2fid(parent)), PFID(lfsck_dto2fid(child)),
2719 llr->llr_ost_idx, llr->llr_lov_idx,
2720 la->la_uid, la->la_gid,
2721 create ? "Create the lost OST-object as required" :
2722 "Keep the MDT-object there by default", rc);
2727 /* If the OST-object does not recognize the MDT-object as its parent, and
2728 * there is no other MDT-object claims as its parent, then just trust the
2729 * given MDT-object as its parent. So update the OST-object filter_fid. */
2730 static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
2731 struct lfsck_component *com,
2732 struct dt_object *parent,
2733 struct lfsck_layout_req *llr,
2736 struct lfsck_thread_info *info = lfsck_env_info(env);
2737 struct filter_fid *pfid = &info->lti_new_pfid;
2738 struct dt_object *child = llr->llr_child;
2739 struct dt_device *dev = lfsck_obj2dev(child);
2740 const struct lu_fid *tfid = lu_object_fid(&parent->do_lu);
2741 struct thandle *handle;
2743 struct lustre_handle lh = { 0 };
2747 rc = lfsck_ibits_lock(env, com->lc_lfsck, parent, &lh,
2748 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2753 pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
2754 pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
2755 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
2756 * MDT-object's FID::f_ver, instead it is the OST-object index in its
2757 * parent MDT-object's layout EA. */
2758 pfid->ff_parent.f_stripe_idx = cpu_to_le32(llr->llr_lov_idx);
2759 buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
2761 handle = dt_trans_create(env, dev);
2763 GOTO(unlock1, rc = PTR_ERR(handle));
2765 rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
2769 rc = dt_attr_get(env, parent, la);
2773 la->la_valid = LA_UID | LA_GID;
2774 rc = dt_declare_attr_set(env, child, la, handle);
2778 rc = dt_trans_start_local(env, dev, handle);
2782 dt_write_lock(env, parent, 0);
2783 if (unlikely(lfsck_is_dead_obj(parent)))
2784 GOTO(unlock2, rc = 1);
2786 rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
2790 /* Get the latest parent's owner. */
2791 rc = dt_attr_get(env, parent, la);
2795 la->la_valid = LA_UID | LA_GID;
2796 rc = dt_attr_set(env, child, la, handle);
2801 dt_write_unlock(env, parent);
2804 rc = lfsck_layout_trans_stop(env, dev, handle, rc);
2807 lfsck_ibits_unlock(&lh, LCK_EX);
2811 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired "
2812 "unmatched MDT-OST pair for: parent "DFID
2813 ", child "DFID", OST-index %u, stripe-index %u, "
2814 "owner %u/%u: rc = %d\n",
2815 lfsck_lfsck2name(com->lc_lfsck),
2816 PFID(lfsck_dto2fid(parent)),
2817 PFID(lfsck_dto2fid(child)),
2818 llr->llr_ost_idx, llr->llr_lov_idx,
2819 la->la_uid, la->la_gid, rc);
2824 /* If there are more than one MDT-objects claim as the OST-object's parent,
2825 * and the OST-object only recognizes one of them, then we need to generate
2826 * new OST-object(s) with new fid(s) for the non-recognized MDT-object(s). */
2827 static int lfsck_layout_repair_multiple_references(const struct lu_env *env,
2828 struct lfsck_component *com,
2829 struct dt_object *parent,
2830 struct lfsck_layout_req *llr,
2834 struct lfsck_thread_info *info = lfsck_env_info(env);
2835 struct dt_allocation_hint *hint = &info->lti_hint;
2836 struct dt_object_format *dof = &info->lti_dof;
2837 struct ost_id *oi = &info->lti_oi;
2838 struct lfsck_instance *lfsck = com->lc_lfsck;
2839 struct dt_device *dev;
2840 struct lu_device *d =
2841 &lfsck_obj2dev(llr->llr_child)->dd_lu_dev;
2842 struct lu_object *o;
2843 struct lu_object *n;
2844 struct dt_object *child = NULL;
2845 struct thandle *handle = NULL;
2846 struct lov_mds_md_v1 *lmm;
2847 struct lov_ost_data_v1 *objs;
2848 const struct lu_fid *pfid = lfsck_dto2fid(parent);
2850 struct lustre_handle lh = { 0 };
2851 struct lu_buf ea_buf;
2857 /* We use two separated transactions to repair the inconsistency.
2859 * 1) create the child (OST-object).
2860 * 2) update the parent LOV EA according to the child's FID.
2862 * If 1) succeed, but 2) failed or aborted, then such OST-object will be
2863 * handled as orphan when the layout LFSCK run next time.
2865 * If 1) failed, but 2) succeed, then such OST-object will be re-created
2866 * as dangling referened case when the layout LFSCK run next time. */
2868 /* The 1st transaction. */
2869 o = lu_object_anon(env, d, NULL);
2871 GOTO(log, rc = PTR_ERR(o));
2873 n = lu_object_locate(o->lo_header, d->ld_type);
2874 if (unlikely(n == NULL)) {
2875 lu_object_put_nocache(env, o);
2877 GOTO(log, rc = -EINVAL);
2880 child = container_of(n, struct dt_object, do_lu);
2881 memset(hint, 0, sizeof(*hint));
2882 rc = dt_attr_get(env, parent, la);
2886 la->la_valid = LA_UID | LA_GID;
2887 memset(dof, 0, sizeof(*dof));
2889 dev = lfsck_obj2dev(child);
2890 handle = dt_trans_create(env, dev);
2892 GOTO(log, rc = PTR_ERR(handle));
2894 rc = dt_declare_create(env, child, la, hint, dof, handle);
2898 rc = dt_trans_start_local(env, dev, handle);
2902 rc = dt_create(env, child, la, hint, dof, handle);
2903 dt_trans_stop(env, dev, handle);
2908 rc = lfsck_ibits_lock(env, lfsck, parent, &lh,
2909 MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR,
2914 /* The 2nd transaction. */
2916 /* XXX: Generally, we should use bottom device (OSD) to update parent
2917 * LOV EA. But because the LOD-object still references the wrong
2918 * OSP-object that should be detached after the parent's LOV EA
2919 * refreshed. Unfortunately, there is no suitable API for that.
2920 * So we have to make the LOD to re-load the OSP-object(s) via
2921 * replacing the LOV EA against the LOD-object.
2923 * Once the DNE2 patches have been landed, we can replace the
2924 * LOD device with the OSD device. LU-6230. */
2926 dev = lfsck->li_next;
2927 parent = lfsck_object_locate(dev, parent);
2929 GOTO(log, rc = PTR_ERR(parent));
2931 handle = dt_trans_create(env, dev);
2933 GOTO(log, rc = PTR_ERR(handle));
2935 rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2936 LU_XATTR_REPLACE, handle);
2940 rc = dt_trans_start_local(env, dev, handle);
2944 dt_write_lock(env, parent, 0);
2945 if (unlikely(lfsck_is_dead_obj(parent)))
2946 GOTO(unlock, rc = 0);
2948 rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV);
2949 if (unlikely(rc == 0 || rc == -ENODATA || rc == -ERANGE))
2950 GOTO(unlock, rc = 0);
2953 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
2954 * been verified in lfsck_layout_verify_header() already. If some
2955 * new magic introduced in the future, then layout LFSCK needs to
2956 * be updated also. */
2957 magic = le32_to_cpu(lmm->lmm_magic);
2958 if (magic == LOV_MAGIC_V1) {
2959 objs = &lmm->lmm_objects[llr->llr_lov_idx];
2961 LASSERT(magic == LOV_MAGIC_V3);
2963 &((struct lov_mds_md_v3 *)lmm)->lmm_objects[llr->llr_lov_idx];
2966 ostid_le_to_cpu(&objs->l_ost_oi, oi);
2967 index = le32_to_cpu(objs->l_ost_idx);
2968 rc = ostid_to_fid(&tfid, oi, index);
2969 /* Someone changed layout during the LFSCK, no need to repair then. */
2970 if (rc == 0 && !lu_fid_eq(&tfid, lu_object_fid(&llr->llr_child->do_lu)))
2971 GOTO(unlock, rc = 0);
2973 lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2974 fid_to_ostid(lu_object_fid(&child->do_lu), oi);
2975 ostid_cpu_to_le(oi, &objs->l_ost_oi);
2976 objs->l_ost_gen = cpu_to_le32(0);
2977 objs->l_ost_idx = cpu_to_le32(llr->llr_ost_idx);
2978 lfsck_buf_init(&ea_buf, lmm,
2979 lov_mds_md_size(le16_to_cpu(lmm->lmm_stripe_count),
2981 rc = dt_xattr_set(env, parent, &ea_buf, XATTR_NAME_LOV,
2982 LU_XATTR_REPLACE, handle);
2984 GOTO(unlock, rc = (rc == 0 ? 1 : rc));
2987 dt_write_unlock(env, parent);
2991 dt_trans_stop(env, dev, handle);
2994 lfsck_ibits_unlock(&lh, LCK_EX);
2996 lfsck_object_put(env, child);
2999 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired "
3000 "multiple references for: parent "DFID", OST-index %u, "
3001 "stripe-index %u, owner %u/%u: rc = %d\n",
3002 lfsck_lfsck2name(lfsck), PFID(pfid), llr->llr_ost_idx,
3003 llr->llr_lov_idx, la->la_uid, la->la_gid, rc);
3008 /* If the MDT-object and the OST-object have different owner information,
3009 * then trust the MDT-object, because the normal chown/chgrp handle order
3010 * is from MDT to OST, and it is possible that some chown/chgrp operation
3011 * is partly done. */
3012 static int lfsck_layout_repair_owner(const struct lu_env *env,
3013 struct lfsck_component *com,
3014 struct dt_object *parent,
3015 struct lfsck_layout_req *llr,
3016 struct lu_attr *pla,
3017 const struct lu_attr *cla)
3019 struct lfsck_thread_info *info = lfsck_env_info(env);
3020 struct lu_attr *tla = &info->lti_la2;
3021 struct dt_object *child = llr->llr_child;
3022 struct dt_device *dev = lfsck_obj2dev(child);
3023 struct thandle *handle;
3027 tla->la_uid = pla->la_uid;
3028 tla->la_gid = pla->la_gid;
3029 tla->la_valid = LA_UID | LA_GID;
3030 handle = dt_trans_create(env, dev);
3032 GOTO(log, rc = PTR_ERR(handle));
3034 rc = dt_declare_attr_set(env, child, tla, handle);
3038 rc = dt_trans_start_local(env, dev, handle);
3042 /* Use the dt_object lock to serialize with destroy and attr_set. */
3043 dt_read_lock(env, parent, 0);
3044 if (unlikely(lfsck_is_dead_obj(parent)))
3045 GOTO(unlock, rc = 1);
3047 /* Get the latest parent's owner. */
3048 rc = dt_attr_get(env, parent, pla);
3052 /* Some others chown/chgrp during the LFSCK, needs to do nothing. */
3053 if (unlikely(tla->la_uid != pla->la_uid ||
3054 tla->la_gid != pla->la_gid))
3057 rc = dt_attr_set(env, child, tla, handle);
3062 dt_read_unlock(env, parent);
3065 rc = lfsck_layout_trans_stop(env, dev, handle, rc);
3069 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant repaired "
3070 "inconsistent file owner for: parent "DFID", child "DFID
3071 ", OST-index %u, stripe-index %u, old owner %u/%u, "
3072 "new owner %u/%u: rc = %d\n",
3073 lfsck_lfsck2name(com->lc_lfsck),
3074 PFID(lfsck_dto2fid(parent)), PFID(lfsck_dto2fid(child)),
3075 llr->llr_ost_idx, llr->llr_lov_idx,
3076 cla->la_uid, cla->la_gid, tla->la_uid, tla->la_gid, rc);
3081 /* Check whether the OST-object correctly back points to the
3082 * MDT-object (@parent) via the XATTR_NAME_FID xattr (@pfid). */
3083 static int lfsck_layout_check_parent(const struct lu_env *env,
3084 struct lfsck_component *com,
3085 struct lfsck_assistant_object *lso,
3086 const struct lu_fid *pfid,
3087 const struct lu_fid *cfid,
3088 const struct lu_attr *cla,
3089 struct lfsck_layout_req *llr,
3090 struct lu_buf *lov_ea, __u32 idx)
3092 struct lfsck_thread_info *info = lfsck_env_info(env);
3093 struct lu_buf *buf = &info->lti_big_buf;
3094 struct dt_object *tobj;
3095 struct lov_mds_md_v1 *lmm;
3096 struct lov_ost_data_v1 *objs;
3097 struct lustre_handle lh = { 0 };
3104 if (unlikely(!fid_is_sane(pfid)))
3105 RETURN(LLIT_UNMATCHED_PAIR);
3107 if (lu_fid_eq(pfid, &lso->lso_fid)) {
3108 if (likely(llr->llr_lov_idx == idx))
3111 RETURN(LLIT_UNMATCHED_PAIR);
3114 tobj = lfsck_object_find_bottom(env, com->lc_lfsck, pfid);
3116 RETURN(PTR_ERR(tobj));
3118 if (dt_object_exists(tobj) == 0 || lfsck_is_dead_obj(tobj) ||
3119 !S_ISREG(lfsck_object_type(tobj)))
3120 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3122 /* Load the tobj's layout EA, in spite of it is a local MDT-object or
3123 * remote one on another MDT. Then check whether the given OST-object
3124 * is in such layout. If yes, it is multiple referenced, otherwise it
3125 * is unmatched referenced case. */
3126 rc = lfsck_layout_get_lovea(env, tobj, buf);
3127 if (rc == 0 || rc == -ENOENT)
3128 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3134 magic = le32_to_cpu(lmm->lmm_magic);
3135 if (magic == LOV_MAGIC_V1) {
3136 objs = &lmm->lmm_objects[0];
3138 LASSERT(magic == LOV_MAGIC_V3);
3139 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
3142 count = le16_to_cpu(lmm->lmm_stripe_count);
3143 for (i = 0; i < count; i++, objs++) {
3144 struct lu_fid *tfid = &info->lti_fid2;
3145 struct ost_id *oi = &info->lti_oi;
3148 if (lovea_slot_is_dummy(objs))
3151 ostid_le_to_cpu(&objs->l_ost_oi, oi);
3152 idx2 = le32_to_cpu(objs->l_ost_idx);
3153 rc = ostid_to_fid(tfid, oi, idx2);
3155 CDEBUG(D_LFSCK, "%s: the parent "DFID" contains "
3156 "invalid layout EA at the slot %d, index %u\n",
3157 lfsck_lfsck2name(com->lc_lfsck),
3158 PFID(pfid), i, idx2);
3160 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3163 if (lu_fid_eq(cfid, tfid)) {
3164 rc = lfsck_ibits_lock(env, com->lc_lfsck, tobj, &lh,
3165 MDS_INODELOCK_UPDATE |
3166 MDS_INODELOCK_LAYOUT |
3167 MDS_INODELOCK_XATTR,
3172 dt_read_lock(env, tobj, 0);
3174 /* For local MDT-object, re-check existence
3175 * after taken the lock. */
3176 if (!dt_object_remote(tobj)) {
3177 if (dt_object_exists(tobj) == 0 ||
3178 lfsck_is_dead_obj(tobj)) {
3179 rc = LLIT_UNMATCHED_PAIR;
3182 rc = LLIT_MULTIPLE_REFERENCED;
3188 /* For migration case, the new MDT-object and old
3189 * MDT-object may reference the same OST-object at
3190 * some migration internal time.
3192 * For remote MDT-object, the local MDT may not know
3193 * whether it has been removed or not. Try checking
3194 * for a non-existent xattr to check if this object
3195 * has been been removed or not. */
3196 rc = dt_xattr_get(env, tobj, &LU_BUF_NULL,
3198 if (unlikely(rc == -ENOENT || rc >= 0)) {
3199 rc = LLIT_UNMATCHED_PAIR;
3200 } else if (rc == -ENODATA) {
3202 rc = LLIT_MULTIPLE_REFERENCED;
3209 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3212 if (lustre_handle_is_used(&lh)) {
3213 dt_read_unlock(env, tobj);
3214 lfsck_ibits_unlock(&lh, LCK_EX);
3218 lfsck_object_put(env, tobj);
3223 static int lfsck_layout_assistant_handler_p1(const struct lu_env *env,
3224 struct lfsck_component *com,
3225 struct lfsck_assistant_req *lar)
3227 struct lfsck_layout_req *llr =
3228 container_of0(lar, struct lfsck_layout_req, llr_lar);
3229 struct lfsck_assistant_object *lso = lar->lar_parent;
3230 struct lfsck_layout *lo = com->lc_file_ram;
3231 struct lfsck_thread_info *info = lfsck_env_info(env);
3232 struct filter_fid_old *pea = &info->lti_old_pfid;
3233 struct lu_fid *pfid = &info->lti_fid;
3234 struct lu_buf buf = { NULL };
3235 struct dt_object *parent = NULL;
3236 struct dt_object *child = llr->llr_child;
3237 struct lu_attr *pla = &lso->lso_attr;
3238 struct lu_attr *cla = &info->lti_la;
3239 struct lfsck_instance *lfsck = com->lc_lfsck;
3240 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
3241 enum lfsck_layout_inconsistency_type type = LLIT_NONE;
3249 rc = dt_attr_get(env, child, cla);
3250 if (rc == -ENOENT) {
3251 parent = lfsck_assistant_object_load(env, lfsck, lso);
3252 if (IS_ERR(parent)) {
3253 rc = PTR_ERR(parent);
3255 RETURN(rc == -ENOENT ? 0 : rc);
3258 type = LLIT_DANGLING;
3265 lfsck_buf_init(&buf, pea, sizeof(struct filter_fid_old));
3266 rc = dt_xattr_get(env, child, &buf, XATTR_NAME_FID);
3267 if (unlikely(rc > 0 && rc != sizeof(struct filter_fid_old) &&
3268 rc != sizeof(struct filter_fid))) {
3269 type = LLIT_UNMATCHED_PAIR;
3273 if (rc < 0 && rc != -ENODATA)
3276 if (rc == 0 || rc == -ENODATA)
3277 GOTO(check_owner, rc = 0);
3279 fid_le_to_cpu(pfid, &pea->ff_parent);
3280 /* Currently, the filter_fid::ff_parent::f_ver is not the
3281 * real parent MDT-object's FID::f_ver, instead it is the
3282 * OST-object index in its parent MDT-object's layout EA. */
3283 idx = pfid->f_stripe_idx;
3285 rc = lfsck_layout_check_parent(env, com, lso, pfid,
3286 lu_object_fid(&child->do_lu),
3287 cla, llr, &buf, idx);
3297 /* Someone may has changed the owner after the parent attr pre-loaded.
3298 * It can be handled later inside the lfsck_layout_repair_owner(). */
3299 if (unlikely(cla->la_uid != pla->la_uid ||
3300 cla->la_gid != pla->la_gid)) {
3301 type = LLIT_INCONSISTENT_OWNER;
3306 if (type == LLIT_NONE)
3309 if (bk->lb_param & LPF_DRYRUN)
3312 if (parent == NULL) {
3313 parent = lfsck_assistant_object_load(env, lfsck, lso);
3314 if (IS_ERR(parent)) {
3315 rc = PTR_ERR(parent);
3326 rc = lfsck_layout_repair_dangling(env, com, parent, llr, pla);
3328 case LLIT_UNMATCHED_PAIR:
3329 rc = lfsck_layout_repair_unmatched_pair(env, com, parent,
3332 case LLIT_MULTIPLE_REFERENCED:
3333 rc = lfsck_layout_repair_multiple_references(env, com, parent,
3336 case LLIT_INCONSISTENT_OWNER:
3337 rc = lfsck_layout_repair_owner(env, com, parent, llr, pla, cla);
3347 down_write(&com->lc_sem);
3349 struct lfsck_assistant_data *lad = com->lc_data;
3351 if (unlikely(lad->lad_exit)) {
3353 } else if (rc == -ENOTCONN || rc == -ESHUTDOWN ||
3354 rc == -ETIMEDOUT || rc == -EHOSTDOWN ||
3355 rc == -EHOSTUNREACH) {
3356 /* If cannot touch the target server,
3357 * mark the LFSCK as INCOMPLETE. */
3358 CDEBUG(D_LFSCK, "%s: layout LFSCK assistant fail to "
3359 "talk with OST %x: rc = %d\n",
3360 lfsck_lfsck2name(lfsck), llr->llr_ost_idx, rc);
3361 lfsck_lad_set_bitmap(env, com, llr->llr_ost_idx);
3362 lo->ll_objs_skipped++;
3365 lfsck_layout_record_failure(env, lfsck, lo);
3367 } else if (rc > 0) {
3368 LASSERTF(type > LLIT_NONE && type <= LLIT_MAX,
3369 "unknown type = %d\n", type);
3371 lo->ll_objs_repaired[type - 1]++;
3372 if (bk->lb_param & LPF_DRYRUN &&
3373 unlikely(lo->ll_pos_first_inconsistent == 0))
3374 lo->ll_pos_first_inconsistent =
3375 lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
3378 up_write(&com->lc_sem);
3380 if (parent != NULL && !IS_ERR(parent))
3381 lfsck_object_put(env, parent);
3386 static int lfsck_layout_assistant_handler_p2(const struct lu_env *env,
3387 struct lfsck_component *com)
3389 struct lfsck_assistant_data *lad = com->lc_data;
3390 struct lfsck_instance *lfsck = com->lc_lfsck;
3391 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
3392 struct lfsck_tgt_descs *ltds = &lfsck->li_ost_descs;
3393 struct lfsck_tgt_desc *ltd;
3397 CDEBUG(D_LFSCK, "%s: layout LFSCK phase2 scan start\n",
3398 lfsck_lfsck2name(lfsck));
3400 spin_lock(<ds->ltd_lock);
3401 while (!list_empty(&lad->lad_ost_phase2_list)) {
3402 ltd = list_entry(lad->lad_ost_phase2_list.next,
3403 struct lfsck_tgt_desc,
3404 ltd_layout_phase_list);
3405 list_del_init(<d->ltd_layout_phase_list);
3406 if (bk->lb_param & LPF_ALL_TGT) {
3407 spin_unlock(<ds->ltd_lock);
3408 rc = lfsck_layout_scan_orphan(env, com, ltd);
3409 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
3412 if (unlikely(lad->lad_exit ||
3413 !thread_is_running(&lfsck->li_thread)))
3415 spin_lock(<ds->ltd_lock);
3419 if (list_empty(&lad->lad_ost_phase1_list))
3423 spin_unlock(<ds->ltd_lock);
3425 CDEBUG(D_LFSCK, "%s: layout LFSCK phase2 scan stop: rc = %d\n",
3426 lfsck_lfsck2name(lfsck), rc);
3432 lfsck_layout_slave_async_interpret(const struct lu_env *env,
3433 struct ptlrpc_request *req,
3436 struct lfsck_layout_slave_async_args *llsaa = args;
3437 struct obd_export *exp = llsaa->llsaa_exp;
3438 struct lfsck_component *com = llsaa->llsaa_com;
3439 struct lfsck_layout_slave_target *llst = llsaa->llsaa_llst;
3440 struct lfsck_layout_slave_data *llsd = com->lc_data;
3441 struct lfsck_reply *lr = NULL;
3445 /* It is quite probably caused by target crash,
3446 * to make the LFSCK can go ahead, assume that
3447 * the target finished the LFSCK prcoessing. */
3450 lr = req_capsule_server_get(&req->rq_pill, &RMF_LFSCK_REPLY);
3451 if (lr->lr_status != LS_SCANNING_PHASE1 &&
3452 lr->lr_status != LS_SCANNING_PHASE2)
3457 CDEBUG(D_LFSCK, "%s: layout LFSCK slave gets the MDT %x "
3458 "status %d\n", lfsck_lfsck2name(com->lc_lfsck),
3459 llst->llst_index, lr != NULL ? lr->lr_status : rc);
3461 lfsck_layout_llst_del(llsd, llst);
3464 lfsck_layout_llst_put(llst);
3465 lfsck_component_put(env, com);
3466 class_export_put(exp);
3471 static int lfsck_layout_async_query(const struct lu_env *env,
3472 struct lfsck_component *com,
3473 struct obd_export *exp,
3474 struct lfsck_layout_slave_target *llst,
3475 struct lfsck_request *lr,
3476 struct ptlrpc_request_set *set)
3478 struct lfsck_layout_slave_async_args *llsaa;
3479 struct ptlrpc_request *req;
3480 struct lfsck_request *tmp;
3484 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_QUERY);
3488 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_QUERY);
3490 ptlrpc_request_free(req);
3494 tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3496 ptlrpc_request_set_replen(req);
3498 llsaa = ptlrpc_req_async_args(req);
3499 llsaa->llsaa_exp = exp;
3500 llsaa->llsaa_com = lfsck_component_get(com);
3501 llsaa->llsaa_llst = llst;
3502 req->rq_interpret_reply = lfsck_layout_slave_async_interpret;
3503 ptlrpc_set_add_req(set, req);
3508 static int lfsck_layout_async_notify(const struct lu_env *env,
3509 struct obd_export *exp,
3510 struct lfsck_request *lr,
3511 struct ptlrpc_request_set *set)
3513 struct ptlrpc_request *req;
3514 struct lfsck_request *tmp;
3518 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
3522 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
3524 ptlrpc_request_free(req);
3528 tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3530 ptlrpc_request_set_replen(req);
3531 ptlrpc_set_add_req(set, req);
3537 lfsck_layout_slave_query_master(const struct lu_env *env,
3538 struct lfsck_component *com)
3540 struct lfsck_request *lr = &lfsck_env_info(env)->lti_lr;
3541 struct lfsck_instance *lfsck = com->lc_lfsck;
3542 struct lfsck_layout_slave_data *llsd = com->lc_data;
3543 struct lfsck_layout_slave_target *llst;
3544 struct obd_export *exp;
3545 struct ptlrpc_request_set *set;
3550 set = ptlrpc_prep_set();
3552 GOTO(log, rc = -ENOMEM);
3554 memset(lr, 0, sizeof(*lr));
3555 lr->lr_event = LE_QUERY;
3556 lr->lr_active = LFSCK_TYPE_LAYOUT;
3558 llsd->llsd_touch_gen++;
3559 spin_lock(&llsd->llsd_lock);
3560 while (!list_empty(&llsd->llsd_master_list)) {
3561 llst = list_entry(llsd->llsd_master_list.next,
3562 struct lfsck_layout_slave_target,
3564 if (llst->llst_gen == llsd->llsd_touch_gen)
3567 llst->llst_gen = llsd->llsd_touch_gen;
3568 list_move_tail(&llst->llst_list,
3569 &llsd->llsd_master_list);
3570 atomic_inc(&llst->llst_ref);
3571 spin_unlock(&llsd->llsd_lock);
3573 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3576 lfsck_layout_llst_del(llsd, llst);
3577 lfsck_layout_llst_put(llst);
3578 spin_lock(&llsd->llsd_lock);
3582 rc = lfsck_layout_async_query(env, com, exp, llst, lr, set);
3584 CDEBUG(D_LFSCK, "%s: layout LFSCK slave fail to "
3585 "query %s for layout: rc = %d\n",
3586 lfsck_lfsck2name(lfsck),
3587 exp->exp_obd->obd_name, rc);
3590 lfsck_layout_llst_put(llst);
3591 class_export_put(exp);
3593 spin_lock(&llsd->llsd_lock);
3595 spin_unlock(&llsd->llsd_lock);
3597 rc = ptlrpc_set_wait(set);
3598 ptlrpc_set_destroy(set);
3600 GOTO(log, rc = (rc1 != 0 ? rc1 : rc));
3603 CDEBUG(D_LFSCK, "%s: layout LFSCK slave queries master: rc = %d\n",
3604 lfsck_lfsck2name(com->lc_lfsck), rc);
3610 lfsck_layout_slave_notify_master(const struct lu_env *env,
3611 struct lfsck_component *com,
3612 enum lfsck_events event, int result)
3614 struct lfsck_layout *lo = com->lc_file_ram;
3615 struct lfsck_instance *lfsck = com->lc_lfsck;
3616 struct lfsck_layout_slave_data *llsd = com->lc_data;
3617 struct lfsck_request *lr = &lfsck_env_info(env)->lti_lr;
3618 struct lfsck_layout_slave_target *llst;
3619 struct obd_export *exp;
3620 struct ptlrpc_request_set *set;
3624 CDEBUG(D_LFSCK, "%s: layout LFSCK slave notifies master\n",
3625 lfsck_lfsck2name(com->lc_lfsck));
3627 set = ptlrpc_prep_set();
3631 memset(lr, 0, sizeof(*lr));
3632 lr->lr_event = event;
3633 lr->lr_flags = LEF_FROM_OST;
3634 lr->lr_status = result;
3635 lr->lr_index = lfsck_dev_idx(lfsck);
3636 lr->lr_active = LFSCK_TYPE_LAYOUT;
3637 lr->lr_flags2 = lo->ll_flags;
3638 llsd->llsd_touch_gen++;
3639 spin_lock(&llsd->llsd_lock);
3640 while (!list_empty(&llsd->llsd_master_list)) {
3641 llst = list_entry(llsd->llsd_master_list.next,
3642 struct lfsck_layout_slave_target,
3644 if (llst->llst_gen == llsd->llsd_touch_gen)
3647 llst->llst_gen = llsd->llsd_touch_gen;
3648 list_move_tail(&llst->llst_list,
3649 &llsd->llsd_master_list);
3650 atomic_inc(&llst->llst_ref);
3651 spin_unlock(&llsd->llsd_lock);
3653 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3656 lfsck_layout_llst_del(llsd, llst);
3657 lfsck_layout_llst_put(llst);
3658 spin_lock(&llsd->llsd_lock);
3662 rc = lfsck_layout_async_notify(env, exp, lr, set);
3664 CDEBUG(D_LFSCK, "%s: layout LFSCK slave fail to "
3665 "notify %s for layout: rc = %d\n",
3666 lfsck_lfsck2name(lfsck),
3667 exp->exp_obd->obd_name, rc);
3669 lfsck_layout_llst_put(llst);
3670 class_export_put(exp);
3671 spin_lock(&llsd->llsd_lock);
3673 spin_unlock(&llsd->llsd_lock);
3675 ptlrpc_set_wait(set);
3676 ptlrpc_set_destroy(set);
3682 * \ret -ENODATA: unrecognized stripe
3683 * \ret = 0 : recognized stripe
3684 * \ret < 0 : other failures
3686 static int lfsck_layout_master_check_pairs(const struct lu_env *env,
3687 struct lfsck_component *com,
3688 struct lu_fid *cfid,
3689 struct lu_fid *pfid)
3691 struct lfsck_thread_info *info = lfsck_env_info(env);
3692 struct lu_buf *buf = &info->lti_big_buf;
3693 struct ost_id *oi = &info->lti_oi;
3694 struct dt_object *obj;
3695 struct lov_mds_md_v1 *lmm;
3696 struct lov_ost_data_v1 *objs;
3697 __u32 idx = pfid->f_stripe_idx;
3705 obj = lfsck_object_find_bottom(env, com->lc_lfsck, pfid);
3707 RETURN(PTR_ERR(obj));
3709 dt_read_lock(env, obj, 0);
3710 if (unlikely(dt_object_exists(obj) == 0 ||
3711 lfsck_is_dead_obj(obj)))
3712 GOTO(unlock, rc = -ENOENT);
3714 if (!S_ISREG(lfsck_object_type(obj)))
3715 GOTO(unlock, rc = -ENODATA);
3717 rc = lfsck_layout_get_lovea(env, obj, buf);
3722 GOTO(unlock, rc = -ENODATA);
3725 rc = lfsck_layout_verify_header(lmm);
3729 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
3730 * been verified in lfsck_layout_verify_header() already. If some
3731 * new magic introduced in the future, then layout LFSCK needs to
3732 * be updated also. */
3733 magic = le32_to_cpu(lmm->lmm_magic);
3734 if (magic == LOV_MAGIC_V1) {
3735 objs = &lmm->lmm_objects[0];
3737 LASSERT(magic == LOV_MAGIC_V3);
3738 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
3741 fid_to_ostid(cfid, oi);
3742 count = le16_to_cpu(lmm->lmm_stripe_count);
3743 for (i = 0; i < count; i++, objs++) {
3746 ostid_le_to_cpu(&objs->l_ost_oi, &oi2);
3747 if (memcmp(oi, &oi2, sizeof(*oi)) == 0)
3748 GOTO(unlock, rc = (i != idx ? -ENODATA : 0));
3751 GOTO(unlock, rc = -ENODATA);
3754 dt_read_unlock(env, obj);
3755 lfsck_object_put(env, obj);
3761 * The LFSCK-on-OST will ask the LFSCK-on-MDT to check whether the given
3762 * MDT-object/OST-object pairs match or not to aviod transfer MDT-object
3763 * layout EA from MDT to OST. On one hand, the OST no need to understand
3764 * the layout EA structure; on the other hand, it may cause trouble when
3765 * transfer large layout EA from MDT to OST via normal OUT RPC.
3767 * \ret > 0: unrecognized stripe
3768 * \ret = 0: recognized stripe
3769 * \ret < 0: other failures
3771 static int lfsck_layout_slave_check_pairs(const struct lu_env *env,
3772 struct lfsck_component *com,
3773 struct lu_fid *cfid,
3774 struct lu_fid *pfid)
3776 struct lfsck_instance *lfsck = com->lc_lfsck;
3777 struct obd_device *obd = lfsck->li_obd;
3778 struct seq_server_site *ss = lfsck_dev_site(lfsck);
3779 struct obd_export *exp = NULL;
3780 struct ptlrpc_request *req = NULL;
3781 struct lfsck_request *lr;
3782 struct lu_seq_range *range = &lfsck_env_info(env)->lti_range;
3786 if (unlikely(fid_is_idif(pfid)))
3789 fld_range_set_any(range);
3790 rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(pfid), range);
3792 RETURN(rc == -ENOENT ? 1 : rc);
3794 if (unlikely(!fld_range_is_mdt(range)))
3797 exp = lustre_find_lwp_by_index(obd->obd_name, range->lsr_index);
3798 if (unlikely(exp == NULL))
3801 if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
3802 GOTO(out, rc = -EOPNOTSUPP);
3804 req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
3806 GOTO(out, rc = -ENOMEM);
3808 rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
3810 ptlrpc_request_free(req);
3815 lr = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3816 memset(lr, 0, sizeof(*lr));
3817 lr->lr_event = LE_PAIRS_VERIFY;
3818 lr->lr_active = LFSCK_TYPE_LAYOUT;
3819 lr->lr_fid = *cfid; /* OST-object itself FID. */
3820 lr->lr_fid2 = *pfid; /* The claimed parent FID. */
3822 ptlrpc_request_set_replen(req);
3823 rc = ptlrpc_queue_wait(req);
3824 ptlrpc_req_finished(req);
3826 if (rc == -ENOENT || rc == -ENODATA)
3833 class_export_put(exp);
3838 static int lfsck_layout_slave_repair_pfid(const struct lu_env *env,
3839 struct lfsck_component *com,
3840 struct lfsck_request *lr)
3842 struct dt_object *obj;
3846 obj = lfsck_object_find_bottom(env, com->lc_lfsck, &lr->lr_fid);
3848 GOTO(log, rc = PTR_ERR(obj));
3850 dt_write_lock(env, obj, 0);
3851 if (unlikely(dt_object_exists(obj) == 0 ||
3852 lfsck_is_dead_obj(obj)))
3853 GOTO(unlock, rc = 0);
3855 rc = __lfsck_layout_update_pfid(env, obj, &lr->lr_fid2,
3861 dt_write_unlock(env, obj);
3862 lfsck_object_put(env, obj);
3865 CDEBUG(D_LFSCK, "%s: layout LFSCK slave repaired pfid for "DFID
3866 ", parent "DFID": rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
3867 PFID(&lr->lr_fid), PFID(&lr->lr_fid2), rc);
3874 static void lfsck_layout_slave_quit(const struct lu_env *env,
3875 struct lfsck_component *com);
3877 static int lfsck_layout_reset(const struct lu_env *env,
3878 struct lfsck_component *com, bool init)
3880 struct lfsck_layout *lo = com->lc_file_ram;
3883 down_write(&com->lc_sem);
3885 memset(lo, 0, com->lc_file_size);
3887 __u32 count = lo->ll_success_count;
3888 __u64 last_time = lo->ll_time_last_complete;
3890 memset(lo, 0, com->lc_file_size);
3891 lo->ll_success_count = count;
3892 lo->ll_time_last_complete = last_time;
3895 lo->ll_magic = LFSCK_LAYOUT_MAGIC;
3896 lo->ll_status = LS_INIT;
3898 if (com->lc_lfsck->li_master) {
3899 struct lfsck_assistant_data *lad = com->lc_data;
3901 lad->lad_incomplete = 0;
3902 CFS_RESET_BITMAP(lad->lad_bitmap);
3905 rc = lfsck_layout_store(env, com);
3906 up_write(&com->lc_sem);
3908 CDEBUG(D_LFSCK, "%s: layout LFSCK reset: rc = %d\n",
3909 lfsck_lfsck2name(com->lc_lfsck), rc);
3914 static void lfsck_layout_fail(const struct lu_env *env,
3915 struct lfsck_component *com, bool new_checked)
3917 struct lfsck_layout *lo = com->lc_file_ram;
3919 down_write(&com->lc_sem);
3921 com->lc_new_checked++;
3922 lfsck_layout_record_failure(env, com->lc_lfsck, lo);
3923 up_write(&com->lc_sem);
3926 static int lfsck_layout_master_checkpoint(const struct lu_env *env,
3927 struct lfsck_component *com, bool init)
3929 struct lfsck_instance *lfsck = com->lc_lfsck;
3930 struct lfsck_layout *lo = com->lc_file_ram;
3934 rc = lfsck_checkpoint_generic(env, com);
3936 return rc > 0 ? 0 : rc;
3939 down_write(&com->lc_sem);
3941 lo->ll_pos_latest_start =
3942 lfsck->li_pos_checkpoint.lp_oit_cookie;
3944 lo->ll_pos_last_checkpoint =
3945 lfsck->li_pos_checkpoint.lp_oit_cookie;
3946 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3947 HALF_SEC - lfsck->li_time_last_checkpoint);
3948 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3949 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3950 com->lc_new_checked = 0;
3953 rc = lfsck_layout_store(env, com);
3954 up_write(&com->lc_sem);
3956 CDEBUG(D_LFSCK, "%s: layout LFSCK master checkpoint at the pos ["
3957 LPU64"]: rc = %d\n", lfsck_lfsck2name(lfsck),
3958 lfsck->li_pos_current.lp_oit_cookie, rc);
3963 static int lfsck_layout_slave_checkpoint(const struct lu_env *env,
3964 struct lfsck_component *com, bool init)
3966 struct lfsck_instance *lfsck = com->lc_lfsck;
3967 struct lfsck_layout *lo = com->lc_file_ram;
3970 if (com->lc_new_checked == 0 && !init)
3973 down_write(&com->lc_sem);
3975 lo->ll_pos_latest_start =
3976 lfsck->li_pos_checkpoint.lp_oit_cookie;
3978 lo->ll_pos_last_checkpoint =
3979 lfsck->li_pos_checkpoint.lp_oit_cookie;
3980 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3981 HALF_SEC - lfsck->li_time_last_checkpoint);
3982 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3983 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3984 com->lc_new_checked = 0;
3987 rc = lfsck_layout_store(env, com);
3988 up_write(&com->lc_sem);
3990 CDEBUG(D_LFSCK, "%s: layout LFSCK slave checkpoint at the pos ["
3991 LPU64"]: rc = %d\n", lfsck_lfsck2name(lfsck),
3992 lfsck->li_pos_current.lp_oit_cookie, rc);
3997 static int lfsck_layout_prep(const struct lu_env *env,
3998 struct lfsck_component *com,
3999 struct lfsck_start *start)
4001 struct lfsck_instance *lfsck = com->lc_lfsck;
4002 struct lfsck_layout *lo = com->lc_file_ram;
4003 struct lfsck_position *pos = &com->lc_pos_start;
4005 fid_zero(&pos->lp_dir_parent);
4006 pos->lp_dir_cookie = 0;
4007 if (lo->ll_status == LS_COMPLETED ||
4008 lo->ll_status == LS_PARTIAL ||
4009 /* To handle orphan, must scan from the beginning. */
4010 (start != NULL && start->ls_flags & LPF_OST_ORPHAN)) {
4013 rc = lfsck_layout_reset(env, com, false);
4015 rc = lfsck_set_param(env, lfsck, start, true);
4018 CDEBUG(D_LFSCK, "%s: layout LFSCK prep failed: "
4019 "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
4025 down_write(&com->lc_sem);
4026 lo->ll_time_latest_start = cfs_time_current_sec();
4027 spin_lock(&lfsck->li_lock);
4028 if (lo->ll_flags & LF_SCANNED_ONCE) {
4029 if (!lfsck->li_drop_dryrun ||
4030 lo->ll_pos_first_inconsistent == 0) {
4031 lo->ll_status = LS_SCANNING_PHASE2;
4032 list_move_tail(&com->lc_link,
4033 &lfsck->li_list_double_scan);
4034 pos->lp_oit_cookie = 0;
4038 lo->ll_status = LS_SCANNING_PHASE1;
4039 lo->ll_run_time_phase1 = 0;
4040 lo->ll_run_time_phase2 = 0;
4041 lo->ll_objs_checked_phase1 = 0;
4042 lo->ll_objs_checked_phase2 = 0;
4043 lo->ll_objs_failed_phase1 = 0;
4044 lo->ll_objs_failed_phase2 = 0;
4045 for (i = 0; i < LLIT_MAX; i++)
4046 lo->ll_objs_repaired[i] = 0;
4048 pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
4049 fid_zero(&com->lc_fid_latest_scanned_phase2);
4052 lo->ll_status = LS_SCANNING_PHASE1;
4053 if (!lfsck->li_drop_dryrun ||
4054 lo->ll_pos_first_inconsistent == 0)
4055 pos->lp_oit_cookie = lo->ll_pos_last_checkpoint + 1;
4057 pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
4059 spin_unlock(&lfsck->li_lock);
4060 up_write(&com->lc_sem);
4065 static int lfsck_layout_slave_prep(const struct lu_env *env,
4066 struct lfsck_component *com,
4067 struct lfsck_start_param *lsp)
4069 struct lfsck_layout_slave_data *llsd = com->lc_data;
4070 struct lfsck_instance *lfsck = com->lc_lfsck;
4071 struct lfsck_layout *lo = com->lc_file_ram;
4072 struct lfsck_start *start = lsp->lsp_start;
4075 rc = lfsck_layout_prep(env, com, start);
4079 if (lo->ll_flags & LF_CRASHED_LASTID &&
4080 list_empty(&llsd->llsd_master_list)) {
4081 LASSERT(lfsck->li_out_notify != NULL);
4083 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4084 LE_LASTID_REBUILDING);
4087 if (!lsp->lsp_index_valid)
4090 rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
4091 if (rc == 0 && start != NULL && start->ls_flags & LPF_OST_ORPHAN) {
4092 LASSERT(!llsd->llsd_rbtree_valid);
4094 write_lock(&llsd->llsd_rb_lock);
4095 rc = lfsck_rbtree_setup(env, com);
4096 write_unlock(&llsd->llsd_rb_lock);
4099 CDEBUG(D_LFSCK, "%s: layout LFSCK slave prep done, start pos ["
4100 LPU64"]\n", lfsck_lfsck2name(lfsck),
4101 com->lc_pos_start.lp_oit_cookie);
4106 static int lfsck_layout_master_prep(const struct lu_env *env,
4107 struct lfsck_component *com,
4108 struct lfsck_start_param *lsp)
4113 rc = lfsck_layout_load_bitmap(env, com);
4115 rc = lfsck_layout_reset(env, com, false);
4117 rc = lfsck_set_param(env, com->lc_lfsck,
4118 lsp->lsp_start, true);
4124 rc = lfsck_layout_prep(env, com, lsp->lsp_start);
4128 rc = lfsck_start_assistant(env, com, lsp);
4133 CDEBUG(D_LFSCK, "%s: layout LFSCK master prep done, start pos ["
4134 LPU64"]\n", lfsck_lfsck2name(com->lc_lfsck),
4135 com->lc_pos_start.lp_oit_cookie);
4140 /* Pre-fetch the attribute for each stripe in the given layout EA. */
4141 static int lfsck_layout_scan_stripes(const struct lu_env *env,
4142 struct lfsck_component *com,
4143 struct dt_object *parent,
4144 struct lov_mds_md_v1 *lmm)
4146 struct lfsck_thread_info *info = lfsck_env_info(env);
4147 struct lfsck_instance *lfsck = com->lc_lfsck;
4148 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
4149 struct lfsck_layout *lo = com->lc_file_ram;
4150 struct lfsck_assistant_data *lad = com->lc_data;
4151 struct lfsck_assistant_object *lso = NULL;
4152 struct lov_ost_data_v1 *objs;
4153 struct lfsck_tgt_descs *ltds = &lfsck->li_ost_descs;
4154 struct ptlrpc_thread *mthread = &lfsck->li_thread;
4155 struct ptlrpc_thread *athread = &lad->lad_thread;
4156 struct l_wait_info lwi = { 0 };
4164 lfsck_buf_init(&buf, &info->lti_old_pfid,
4165 sizeof(struct filter_fid_old));
4166 count = le16_to_cpu(lmm->lmm_stripe_count);
4167 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
4168 * been verified in lfsck_layout_verify_header() already. If some
4169 * new magic introduced in the future, then layout LFSCK needs to
4170 * be updated also. */
4171 magic = le32_to_cpu(lmm->lmm_magic);
4172 if (magic == LOV_MAGIC_V1) {
4173 objs = &lmm->lmm_objects[0];
4175 LASSERT(magic == LOV_MAGIC_V3);
4176 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
4179 for (i = 0; i < count; i++, objs++) {
4180 struct lu_fid *fid = &info->lti_fid;
4181 struct ost_id *oi = &info->lti_oi;
4182 struct lfsck_layout_req *llr;
4183 struct lfsck_tgt_desc *tgt = NULL;
4184 struct dt_object *cobj = NULL;
4186 bool wakeup = false;
4188 if (unlikely(lovea_slot_is_dummy(objs)))
4191 l_wait_event(mthread->t_ctl_waitq,
4192 lad->lad_prefetched < bk->lb_async_windows ||
4193 !thread_is_running(mthread) ||
4194 thread_is_stopped(athread),
4197 if (unlikely(!thread_is_running(mthread)) ||
4198 thread_is_stopped(athread))
4201 if (unlikely(lfsck_is_dead_obj(parent)))
4204 ostid_le_to_cpu(&objs->l_ost_oi, oi);
4205 index = le32_to_cpu(objs->l_ost_idx);
4206 rc = ostid_to_fid(fid, oi, index);
4208 CDEBUG(D_LFSCK, "%s: get invalid layout EA for "DFID
4209 ": "DOSTID", idx:%u\n", lfsck_lfsck2name(lfsck),
4210 PFID(lfsck_dto2fid(parent)), POSTID(oi), index);
4214 tgt = lfsck_tgt_get(ltds, index);
4215 if (unlikely(tgt == NULL)) {
4216 CDEBUG(D_LFSCK, "%s: cannot talk with OST %x which "
4217 "did not join the layout LFSCK\n",
4218 lfsck_lfsck2name(lfsck), index);
4219 lfsck_lad_set_bitmap(env, com, index);
4223 /* There is potential deadlock race condition between object
4224 * destroy and layout LFSCK. Consider the following scenario:
4226 * 1) The LFSCK thread obtained the parent object firstly, at
4227 * that time, the parent object has not been destroyed yet.
4229 * 2) One RPC service thread destroyed the parent and all its
4230 * children objects. Because the LFSCK is referencing the
4231 * parent object, then the parent object will be marked as
4232 * dying in RAM. On the other hand, the parent object is
4233 * referencing all its children objects, then all children
4234 * objects will be marked as dying in RAM also.
4236 * 3) The LFSCK thread tries to find some child object with
4237 * the parent object referenced. Then it will find that the
4238 * child object is dying. According to the object visibility
4239 * rules: the object with dying flag cannot be returned to
4240 * others. So the LFSCK thread has to wait until the dying
4241 * object has been purged from RAM, then it can allocate a
4242 * new object (with the same FID) in RAM. Unfortunately, the
4243 * LFSCK thread itself is referencing the parent object, and
4244 * cause the parent object cannot be purged, then cause the
4245 * child object cannot be purged also. So the LFSCK thread
4246 * will fall into deadlock.
4248 * We introduce non-blocked version lu_object_find() to allow
4249 * the LFSCK thread to return failure immediately (instead of
4250 * wait) when it finds dying (child) object, then the LFSCK
4251 * thread can check whether the parent object is dying or not.
4252 * So avoid above deadlock. LU-5395 */
4253 cobj = lfsck_object_find_by_dev_nowait(env, tgt->ltd_tgt, fid);
4255 if (lfsck_is_dead_obj(parent)) {
4265 rc = dt_declare_attr_get(env, cobj);
4269 rc = dt_declare_xattr_get(env, cobj, &buf, XATTR_NAME_FID);
4274 struct lu_attr *attr = &info->lti_la;
4276 rc = dt_attr_get(env, parent, attr);
4282 lso = lfsck_assistant_object_init(env,
4283 lfsck_dto2fid(parent), attr,
4284 lfsck->li_pos_current.lp_oit_cookie, false);
4293 llr = lfsck_layout_assistant_req_init(lso, cobj, index, i);
4300 spin_lock(&lad->lad_lock);
4301 if (lad->lad_assistant_status < 0) {
4302 spin_unlock(&lad->lad_lock);
4303 lfsck_layout_assistant_req_fini(env, &llr->llr_lar);
4305 RETURN(lad->lad_assistant_status);
4308 list_add_tail(&llr->llr_lar.lar_list, &lad->lad_req_list);
4309 if (lad->lad_prefetched == 0)
4312 lad->lad_prefetched++;
4313 spin_unlock(&lad->lad_lock);
4315 wake_up_all(&athread->t_ctl_waitq);
4318 down_write(&com->lc_sem);
4319 com->lc_new_checked++;
4321 lfsck_layout_record_failure(env, lfsck, lo);
4322 up_write(&com->lc_sem);
4324 if (cobj != NULL && !IS_ERR(cobj))
4325 lfsck_object_put(env, cobj);
4327 if (likely(tgt != NULL))
4330 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
4338 lfsck_assistant_object_put(env, lso);
4343 /* For the given object, read its layout EA locally. For each stripe, pre-fetch
4344 * the OST-object's attribute and generate an structure lfsck_layout_req on the
4345 * list ::lad_req_list.
4347 * For each request on above list, the lfsck_layout_assistant thread compares
4348 * the OST side attribute with local attribute, if inconsistent, then repair it.
4350 * All above processing is async mode with pipeline. */
4351 static int lfsck_layout_master_exec_oit(const struct lu_env *env,
4352 struct lfsck_component *com,
4353 struct dt_object *obj)
4355 struct lfsck_thread_info *info = lfsck_env_info(env);
4356 struct ost_id *oi = &info->lti_oi;
4357 struct lfsck_layout *lo = com->lc_file_ram;
4358 struct lfsck_assistant_data *lad = com->lc_data;
4359 struct lfsck_instance *lfsck = com->lc_lfsck;
4360 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
4361 struct thandle *handle = NULL;
4362 struct lu_buf *buf = &info->lti_big_buf;
4363 struct lov_mds_md_v1 *lmm = NULL;
4364 struct dt_device *dev = lfsck_obj2dev(obj);
4365 struct lustre_handle lh = { 0 };
4366 struct lu_buf ea_buf = { NULL };
4369 bool locked = false;
4370 bool stripe = false;
4371 bool bad_oi = false;
4374 if (!S_ISREG(lfsck_object_type(obj)))
4377 if (lad->lad_assistant_status < 0)
4378 GOTO(out, rc = -ESRCH);
4380 fid_to_lmm_oi(lfsck_dto2fid(obj), oi);
4381 lmm_oi_cpu_to_le(oi, oi);
4382 dt_read_lock(env, obj, 0);
4386 if (dt_object_exists(obj) == 0 ||
4387 lfsck_is_dead_obj(obj))
4390 rc = lfsck_layout_get_lovea(env, obj, buf);
4396 rc = lfsck_layout_verify_header(lmm);
4397 /* If the LOV EA crashed, then it is possible to be rebuilt later
4398 * when handle orphan OST-objects. */
4402 if (memcmp(oi, &lmm->lmm_oi, sizeof(*oi)) == 0)
4403 GOTO(out, stripe = true);
4405 /* Inconsistent lmm_oi, should be repaired. */
4409 if (bk->lb_param & LPF_DRYRUN) {
4410 lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4412 GOTO(out, stripe = true);
4415 if (!lustre_handle_is_used(&lh)) {
4416 dt_read_unlock(env, obj);
4418 rc = lfsck_ibits_lock(env, lfsck, obj, &lh,
4419 MDS_INODELOCK_LAYOUT |
4420 MDS_INODELOCK_XATTR, LCK_EX);
4424 handle = dt_trans_create(env, dev);
4426 GOTO(out, rc = PTR_ERR(handle));
4428 lfsck_buf_init(&ea_buf, lmm, size);
4429 rc = dt_declare_xattr_set(env, obj, &ea_buf, XATTR_NAME_LOV,
4430 LU_XATTR_REPLACE, handle);
4434 rc = dt_trans_start_local(env, dev, handle);
4438 dt_write_lock(env, obj, 0);
4444 rc = dt_xattr_set(env, obj, &ea_buf, XATTR_NAME_LOV,
4445 LU_XATTR_REPLACE, handle);
4449 lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4451 GOTO(out, stripe = true);
4455 if (lustre_handle_is_used(&lh))
4456 dt_write_unlock(env, obj);
4458 dt_read_unlock(env, obj);
4461 if (handle != NULL && !IS_ERR(handle))
4462 dt_trans_stop(env, dev, handle);
4464 lfsck_ibits_unlock(&lh, LCK_EX);
4467 CDEBUG(D_LFSCK, "%s: layout LFSCK master %s bad lmm_oi for "
4468 DFID": rc = %d\n", lfsck_lfsck2name(lfsck),
4469 bk->lb_param & LPF_DRYRUN ? "found" : "repaired",
4470 PFID(lfsck_dto2fid(obj)), rc);
4473 rc = lfsck_layout_scan_stripes(env, com, obj, lmm);
4475 down_write(&com->lc_sem);
4476 com->lc_new_checked++;
4478 lfsck_layout_record_failure(env, lfsck, lo);
4479 up_write(&com->lc_sem);
4485 static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
4486 struct lfsck_component *com,
4487 struct dt_object *obj)
4489 struct lfsck_instance *lfsck = com->lc_lfsck;
4490 struct lfsck_layout *lo = com->lc_file_ram;
4491 const struct lu_fid *fid = lfsck_dto2fid(obj);
4492 struct lfsck_layout_slave_data *llsd = com->lc_data;
4493 struct lfsck_layout_seq *lls;
4499 LASSERT(llsd != NULL);
4501 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY5) &&
4502 cfs_fail_val == lfsck_dev_idx(lfsck)) {
4503 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(1),
4505 struct ptlrpc_thread *thread = &lfsck->li_thread;
4507 l_wait_event(thread->t_ctl_waitq,
4508 !thread_is_running(thread),
4512 lfsck_rbtree_update_bitmap(env, com, fid, false);
4514 down_write(&com->lc_sem);
4515 if (fid_is_idif(fid))
4517 else if (!fid_is_norm(fid) ||
4518 !fid_is_for_ostobj(env, lfsck, obj, fid))
4519 GOTO(unlock, rc = 0);
4522 com->lc_new_checked++;
4524 lls = lfsck_layout_seq_lookup(llsd, seq);
4527 if (unlikely(lls == NULL))
4528 GOTO(unlock, rc = -ENOMEM);
4530 INIT_LIST_HEAD(&lls->lls_list);
4532 rc = lfsck_layout_lastid_load(env, com, lls);
4534 CDEBUG(D_LFSCK, "%s: layout LFSCK failed to "
4535 "load LAST_ID for "LPX64": rc = %d\n",
4536 lfsck_lfsck2name(com->lc_lfsck), seq, rc);
4537 lo->ll_objs_failed_phase1++;
4542 lfsck_layout_seq_insert(llsd, lls);
4545 if (unlikely(fid_is_last_id(fid)))
4546 GOTO(unlock, rc = 0);
4548 if (fid_is_idif(fid))
4549 oid = fid_idif_id(fid_seq(fid), fid_oid(fid), fid_ver(fid));
4553 if (oid > lls->lls_lastid_known)
4554 lls->lls_lastid_known = oid;
4556 if (oid > lls->lls_lastid) {
4557 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
4558 /* OFD may create new objects during LFSCK scanning. */
4559 rc = lfsck_layout_lastid_reload(env, com, lls);
4560 if (unlikely(rc != 0)) {
4561 CDEBUG(D_LFSCK, "%s: layout LFSCK failed to "
4562 "reload LAST_ID for "LPX64": rc = %d\n",
4563 lfsck_lfsck2name(com->lc_lfsck),
4569 if (oid <= lls->lls_lastid ||
4570 lo->ll_flags & LF_CRASHED_LASTID)
4571 GOTO(unlock, rc = 0);
4573 LASSERT(lfsck->li_out_notify != NULL);
4575 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4576 LE_LASTID_REBUILDING);
4577 lo->ll_flags |= LF_CRASHED_LASTID;
4579 CDEBUG(D_LFSCK, "%s: layout LFSCK finds crashed "
4580 "LAST_ID file (2) for the sequence "LPX64
4581 ", old value "LPU64", known value "LPU64"\n",
4582 lfsck_lfsck2name(lfsck), lls->lls_seq,
4583 lls->lls_lastid, oid);
4586 lls->lls_lastid = oid;
4590 GOTO(unlock, rc = 0);
4593 up_write(&com->lc_sem);
4598 static int lfsck_layout_exec_dir(const struct lu_env *env,
4599 struct lfsck_component *com,
4600 struct lfsck_assistant_object *lso,
4601 struct lu_dirent *ent, __u16 type)
4606 static int lfsck_layout_master_post(const struct lu_env *env,
4607 struct lfsck_component *com,
4608 int result, bool init)
4610 struct lfsck_instance *lfsck = com->lc_lfsck;
4611 struct lfsck_layout *lo = com->lc_file_ram;
4615 lfsck_post_generic(env, com, &result);
4617 down_write(&com->lc_sem);
4618 spin_lock(&lfsck->li_lock);
4620 lo->ll_pos_last_checkpoint =
4621 lfsck->li_pos_checkpoint.lp_oit_cookie;
4624 if (lo->ll_flags & LF_INCOMPLETE)
4625 lo->ll_status = LS_PARTIAL;
4627 lo->ll_status = LS_SCANNING_PHASE2;
4628 lo->ll_flags |= LF_SCANNED_ONCE;
4629 lo->ll_flags &= ~LF_UPGRADE;
4630 list_move_tail(&com->lc_link, &lfsck->li_list_double_scan);
4631 } else if (result == 0) {
4632 if (lfsck->li_status != 0)
4633 lo->ll_status = lfsck->li_status;
4635 lo->ll_status = LS_STOPPED;
4636 if (lo->ll_status != LS_PAUSED)
4637 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4639 lo->ll_status = LS_FAILED;
4640 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4642 spin_unlock(&lfsck->li_lock);
4645 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4646 HALF_SEC - lfsck->li_time_last_checkpoint);
4647 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4648 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4649 com->lc_new_checked = 0;
4652 rc = lfsck_layout_store(env, com);
4653 up_write(&com->lc_sem);
4655 CDEBUG(D_LFSCK, "%s: layout LFSCK master post done: rc = %d\n",
4656 lfsck_lfsck2name(lfsck), rc);
4661 static int lfsck_layout_slave_post(const struct lu_env *env,
4662 struct lfsck_component *com,
4663 int result, bool init)
4665 struct lfsck_instance *lfsck = com->lc_lfsck;
4666 struct lfsck_layout *lo = com->lc_file_ram;
4670 rc = lfsck_layout_lastid_store(env, com);
4674 LASSERT(lfsck->li_out_notify != NULL);
4676 down_write(&com->lc_sem);
4677 spin_lock(&lfsck->li_lock);
4679 lo->ll_pos_last_checkpoint =
4680 lfsck->li_pos_checkpoint.lp_oit_cookie;
4683 lo->ll_status = LS_SCANNING_PHASE2;
4684 lo->ll_flags |= LF_SCANNED_ONCE;
4685 if (lo->ll_flags & LF_CRASHED_LASTID) {
4687 lo->ll_flags &= ~LF_CRASHED_LASTID;
4689 CDEBUG(D_LFSCK, "%s: layout LFSCK has rebuilt "
4690 "crashed LAST_ID files successfully\n",
4691 lfsck_lfsck2name(lfsck));
4693 lo->ll_flags &= ~LF_UPGRADE;
4694 list_move_tail(&com->lc_link, &lfsck->li_list_double_scan);
4695 } else if (result == 0) {
4696 if (lfsck->li_status != 0)
4697 lo->ll_status = lfsck->li_status;
4699 lo->ll_status = LS_STOPPED;
4700 if (lo->ll_status != LS_PAUSED)
4701 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4703 lo->ll_status = LS_FAILED;
4704 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
4706 spin_unlock(&lfsck->li_lock);
4709 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4713 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4714 HALF_SEC - lfsck->li_time_last_checkpoint);
4715 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4716 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4717 com->lc_new_checked = 0;
4720 rc = lfsck_layout_store(env, com);
4721 up_write(&com->lc_sem);
4723 lfsck_layout_slave_notify_master(env, com, LE_PHASE1_DONE, result);
4725 CDEBUG(D_LFSCK, "%s: layout LFSCK slave post done: rc = %d\n",
4726 lfsck_lfsck2name(lfsck), rc);
4731 static int lfsck_layout_dump(const struct lu_env *env,
4732 struct lfsck_component *com, struct seq_file *m)
4734 struct lfsck_instance *lfsck = com->lc_lfsck;
4735 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
4736 struct lfsck_layout *lo = com->lc_file_ram;
4739 down_read(&com->lc_sem);
4740 seq_printf(m, "name: lfsck_layout\n"
4746 lfsck_status2names(lo->ll_status));
4748 rc = lfsck_bits_dump(m, lo->ll_flags, lfsck_flags_names, "flags");
4752 rc = lfsck_bits_dump(m, bk->lb_param, lfsck_param_names, "param");
4756 rc = lfsck_time_dump(m, lo->ll_time_last_complete,
4761 rc = lfsck_time_dump(m, lo->ll_time_latest_start,
4766 rc = lfsck_time_dump(m, lo->ll_time_last_checkpoint,
4771 seq_printf(m, "latest_start_position: "LPU64"\n"
4772 "last_checkpoint_position: "LPU64"\n"
4773 "first_failure_position: "LPU64"\n",
4774 lo->ll_pos_latest_start,
4775 lo->ll_pos_last_checkpoint,
4776 lo->ll_pos_first_inconsistent);
4778 seq_printf(m, "success_count: %u\n"
4779 "repaired_dangling: "LPU64"\n"
4780 "repaired_unmatched_pair: "LPU64"\n"
4781 "repaired_multiple_referenced: "LPU64"\n"
4782 "repaired_orphan: "LPU64"\n"
4783 "repaired_inconsistent_owner: "LPU64"\n"
4784 "repaired_others: "LPU64"\n"
4785 "skipped: "LPU64"\n"
4786 "failed_phase1: "LPU64"\n"
4787 "failed_phase2: "LPU64"\n",
4788 lo->ll_success_count,
4789 lo->ll_objs_repaired[LLIT_DANGLING - 1],
4790 lo->ll_objs_repaired[LLIT_UNMATCHED_PAIR - 1],
4791 lo->ll_objs_repaired[LLIT_MULTIPLE_REFERENCED - 1],
4792 lo->ll_objs_repaired[LLIT_ORPHAN - 1],
4793 lo->ll_objs_repaired[LLIT_INCONSISTENT_OWNER - 1],
4794 lo->ll_objs_repaired[LLIT_OTHERS - 1],
4795 lo->ll_objs_skipped,
4796 lo->ll_objs_failed_phase1,
4797 lo->ll_objs_failed_phase2);
4799 if (lo->ll_status == LS_SCANNING_PHASE1) {
4801 const struct dt_it_ops *iops;
4802 cfs_duration_t duration = cfs_time_current() -
4803 lfsck->li_time_last_checkpoint;
4804 __u64 checked = lo->ll_objs_checked_phase1 +
4805 com->lc_new_checked;
4806 __u64 speed = checked;
4807 __u64 new_checked = com->lc_new_checked *
4808 msecs_to_jiffies(MSEC_PER_SEC);
4809 __u32 rtime = lo->ll_run_time_phase1 +
4810 cfs_duration_sec(duration + HALF_SEC);
4813 do_div(new_checked, duration);
4815 do_div(speed, rtime);
4816 seq_printf(m, "checked_phase1: "LPU64"\n"
4817 "checked_phase2: "LPU64"\n"
4818 "run_time_phase1: %u seconds\n"
4819 "run_time_phase2: %u seconds\n"
4820 "average_speed_phase1: "LPU64" items/sec\n"
4821 "average_speed_phase2: N/A\n"
4822 "real-time_speed_phase1: "LPU64" items/sec\n"
4823 "real-time_speed_phase2: N/A\n",
4825 lo->ll_objs_checked_phase2,
4827 lo->ll_run_time_phase2,
4831 LASSERT(lfsck->li_di_oit != NULL);
4833 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
4835 /* The low layer otable-based iteration position may NOT
4836 * exactly match the layout-based directory traversal
4837 * cookie. Generally, it is not a serious issue. But the
4838 * caller should NOT make assumption on that. */
4839 pos = iops->store(env, lfsck->li_di_oit);
4840 if (!lfsck->li_current_oit_processed)
4842 seq_printf(m, "current_position: "LPU64"\n", pos);
4844 } else if (lo->ll_status == LS_SCANNING_PHASE2) {
4845 cfs_duration_t duration = cfs_time_current() -
4846 com->lc_time_last_checkpoint;
4847 __u64 checked = lo->ll_objs_checked_phase2 +
4848 com->lc_new_checked;
4849 __u64 speed1 = lo->ll_objs_checked_phase1;
4850 __u64 speed2 = checked;
4851 __u64 new_checked = com->lc_new_checked *
4852 msecs_to_jiffies(MSEC_PER_SEC);
4853 __u32 rtime = lo->ll_run_time_phase2 +
4854 cfs_duration_sec(duration + HALF_SEC);
4857 do_div(new_checked, duration);
4858 if (lo->ll_run_time_phase1 != 0)
4859 do_div(speed1, lo->ll_run_time_phase1);
4861 do_div(speed2, rtime);
4862 rc = seq_printf(m, "checked_phase1: "LPU64"\n"
4863 "checked_phase2: "LPU64"\n"
4864 "run_time_phase1: %u seconds\n"
4865 "run_time_phase2: %u seconds\n"
4866 "average_speed_phase1: "LPU64" items/sec\n"
4867 "average_speed_phase2: "LPU64" items/sec\n"
4868 "real-time_speed_phase1: N/A\n"
4869 "real-time_speed_phase2: "LPU64" items/sec\n"
4870 "current_position: "DFID"\n",
4871 lo->ll_objs_checked_phase1,
4873 lo->ll_run_time_phase1,
4878 PFID(&com->lc_fid_latest_scanned_phase2));
4883 __u64 speed1 = lo->ll_objs_checked_phase1;
4884 __u64 speed2 = lo->ll_objs_checked_phase2;
4886 if (lo->ll_run_time_phase1 != 0)
4887 do_div(speed1, lo->ll_run_time_phase1);
4888 if (lo->ll_run_time_phase2 != 0)
4889 do_div(speed2, lo->ll_run_time_phase2);
4890 seq_printf(m, "checked_phase1: "LPU64"\n"
4891 "checked_phase2: "LPU64"\n"
4892 "run_time_phase1: %u seconds\n"
4893 "run_time_phase2: %u seconds\n"
4894 "average_speed_phase1: "LPU64" items/sec\n"
4895 "average_speed_phase2: "LPU64" objs/sec\n"
4896 "real-time_speed_phase1: N/A\n"
4897 "real-time_speed_phase2: N/A\n"
4898 "current_position: N/A\n",
4899 lo->ll_objs_checked_phase1,
4900 lo->ll_objs_checked_phase2,
4901 lo->ll_run_time_phase1,
4902 lo->ll_run_time_phase2,
4907 up_read(&com->lc_sem);
4912 static int lfsck_layout_master_double_scan(const struct lu_env *env,
4913 struct lfsck_component *com)
4915 struct lfsck_layout *lo = com->lc_file_ram;
4916 struct lfsck_assistant_data *lad = com->lc_data;
4917 struct lfsck_instance *lfsck = com->lc_lfsck;
4918 struct lfsck_tgt_descs *ltds;
4919 struct lfsck_tgt_desc *ltd;
4920 struct lfsck_tgt_desc *next;
4923 rc = lfsck_double_scan_generic(env, com, lo->ll_status);
4925 if (thread_is_stopped(&lad->lad_thread)) {
4926 LASSERT(list_empty(&lad->lad_req_list));
4927 LASSERT(list_empty(&lad->lad_ost_phase1_list));
4928 LASSERT(list_empty(&lad->lad_mdt_phase1_list));
4930 ltds = &lfsck->li_ost_descs;
4931 spin_lock(<ds->ltd_lock);
4932 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase2_list,
4933 ltd_layout_phase_list) {
4934 list_del_init(<d->ltd_layout_phase_list);
4936 spin_unlock(<ds->ltd_lock);
4938 ltds = &lfsck->li_mdt_descs;
4939 spin_lock(<ds->ltd_lock);
4940 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
4941 ltd_layout_phase_list) {
4942 list_del_init(<d->ltd_layout_phase_list);
4944 spin_unlock(<ds->ltd_lock);
4950 static int lfsck_layout_slave_double_scan(const struct lu_env *env,
4951 struct lfsck_component *com)
4953 struct lfsck_instance *lfsck = com->lc_lfsck;
4954 struct lfsck_layout_slave_data *llsd = com->lc_data;
4955 struct lfsck_layout *lo = com->lc_file_ram;
4956 struct ptlrpc_thread *thread = &lfsck->li_thread;
4960 CDEBUG(D_LFSCK, "%s: layout LFSCK slave phase2 scan start\n",
4961 lfsck_lfsck2name(lfsck));
4963 atomic_inc(&lfsck->li_double_scan_count);
4965 if (lo->ll_flags & LF_INCOMPLETE)
4968 com->lc_new_checked = 0;
4969 com->lc_new_scanned = 0;
4970 com->lc_time_last_checkpoint = cfs_time_current();
4971 com->lc_time_next_checkpoint = com->lc_time_last_checkpoint +
4972 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
4975 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(30),
4978 rc = lfsck_layout_slave_query_master(env, com);
4979 if (list_empty(&llsd->llsd_master_list)) {
4980 if (unlikely(!thread_is_running(thread)))
4991 rc = l_wait_event(thread->t_ctl_waitq,
4992 !thread_is_running(thread) ||
4993 lo->ll_flags & LF_INCOMPLETE ||
4994 list_empty(&llsd->llsd_master_list),
4996 if (unlikely(!thread_is_running(thread)))
4999 if (lo->ll_flags & LF_INCOMPLETE)
5002 if (rc == -ETIMEDOUT)
5005 GOTO(done, rc = (rc < 0 ? rc : 1));
5009 rc = lfsck_layout_double_scan_result(env, com, rc);
5010 lfsck_layout_slave_notify_master(env, com, LE_PHASE2_DONE,
5011 (rc > 0 && lo->ll_flags & LF_INCOMPLETE) ? 0 : rc);
5012 lfsck_layout_slave_quit(env, com);
5013 if (atomic_dec_and_test(&lfsck->li_double_scan_count))
5014 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5016 CDEBUG(D_LFSCK, "%s: layout LFSCK slave phase2 scan finished, "
5017 "status %d: rc = %d\n",
5018 lfsck_lfsck2name(lfsck), lo->ll_status, rc);
5023 static void lfsck_layout_master_data_release(const struct lu_env *env,
5024 struct lfsck_component *com)
5026 struct lfsck_assistant_data *lad = com->lc_data;
5027 struct lfsck_instance *lfsck = com->lc_lfsck;
5028 struct lfsck_tgt_descs *ltds;
5029 struct lfsck_tgt_desc *ltd;
5030 struct lfsck_tgt_desc *next;
5032 LASSERT(lad != NULL);
5033 LASSERT(thread_is_init(&lad->lad_thread) ||
5034 thread_is_stopped(&lad->lad_thread));
5035 LASSERT(list_empty(&lad->lad_req_list));
5037 com->lc_data = NULL;
5039 ltds = &lfsck->li_ost_descs;
5040 spin_lock(<ds->ltd_lock);
5041 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase1_list,
5042 ltd_layout_phase_list) {
5043 list_del_init(<d->ltd_layout_phase_list);
5045 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase2_list,
5046 ltd_layout_phase_list) {
5047 list_del_init(<d->ltd_layout_phase_list);
5049 list_for_each_entry_safe(ltd, next, &lad->lad_ost_list,
5051 list_del_init(<d->ltd_layout_list);
5053 spin_unlock(<ds->ltd_lock);
5055 ltds = &lfsck->li_mdt_descs;
5056 spin_lock(<ds->ltd_lock);
5057 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase1_list,
5058 ltd_layout_phase_list) {
5059 list_del_init(<d->ltd_layout_phase_list);
5061 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
5062 ltd_layout_phase_list) {
5063 list_del_init(<d->ltd_layout_phase_list);
5065 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_list,
5067 list_del_init(<d->ltd_layout_list);
5069 spin_unlock(<ds->ltd_lock);
5071 if (likely(lad->lad_bitmap != NULL))
5072 CFS_FREE_BITMAP(lad->lad_bitmap);
5077 static void lfsck_layout_slave_data_release(const struct lu_env *env,
5078 struct lfsck_component *com)
5080 struct lfsck_layout_slave_data *llsd = com->lc_data;
5082 lfsck_layout_slave_quit(env, com);
5083 com->lc_data = NULL;
5087 static void lfsck_layout_master_quit(const struct lu_env *env,
5088 struct lfsck_component *com)
5090 struct lfsck_assistant_data *lad = com->lc_data;
5091 struct lfsck_instance *lfsck = com->lc_lfsck;
5092 struct lfsck_tgt_descs *ltds;
5093 struct lfsck_tgt_desc *ltd;
5094 struct lfsck_tgt_desc *next;
5096 LASSERT(lad != NULL);
5098 lfsck_quit_generic(env, com);
5100 LASSERT(thread_is_init(&lad->lad_thread) ||
5101 thread_is_stopped(&lad->lad_thread));
5102 LASSERT(list_empty(&lad->lad_req_list));
5104 ltds = &lfsck->li_ost_descs;
5105 spin_lock(<ds->ltd_lock);
5106 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase1_list,
5107 ltd_layout_phase_list) {
5108 list_del_init(<d->ltd_layout_phase_list);
5110 list_for_each_entry_safe(ltd, next, &lad->lad_ost_phase2_list,
5111 ltd_layout_phase_list) {
5112 list_del_init(<d->ltd_layout_phase_list);
5114 spin_unlock(<ds->ltd_lock);
5116 ltds = &lfsck->li_mdt_descs;
5117 spin_lock(<ds->ltd_lock);
5118 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase1_list,
5119 ltd_layout_phase_list) {
5120 list_del_init(<d->ltd_layout_phase_list);
5122 list_for_each_entry_safe(ltd, next, &lad->lad_mdt_phase2_list,
5123 ltd_layout_phase_list) {
5124 list_del_init(<d->ltd_layout_phase_list);
5126 spin_unlock(<ds->ltd_lock);
5129 static void lfsck_layout_slave_quit(const struct lu_env *env,
5130 struct lfsck_component *com)
5132 struct lfsck_layout_slave_data *llsd = com->lc_data;
5133 struct lfsck_layout_seq *lls;
5134 struct lfsck_layout_seq *next;
5135 struct lfsck_layout_slave_target *llst;
5137 LASSERT(llsd != NULL);
5139 list_for_each_entry_safe(lls, next, &llsd->llsd_seq_list,
5141 list_del_init(&lls->lls_list);
5142 lfsck_object_put(env, lls->lls_lastid_obj);
5146 spin_lock(&llsd->llsd_lock);
5147 while (!list_empty(&llsd->llsd_master_list)) {
5148 llst = list_entry(llsd->llsd_master_list.next,
5149 struct lfsck_layout_slave_target, llst_list);
5150 list_del_init(&llst->llst_list);
5151 spin_unlock(&llsd->llsd_lock);
5152 lfsck_layout_llst_put(llst);
5153 spin_lock(&llsd->llsd_lock);
5155 spin_unlock(&llsd->llsd_lock);
5157 lfsck_rbtree_cleanup(env, com);
5160 static int lfsck_layout_master_in_notify(const struct lu_env *env,
5161 struct lfsck_component *com,
5162 struct lfsck_request *lr,
5165 struct lfsck_instance *lfsck = com->lc_lfsck;
5166 struct lfsck_layout *lo = com->lc_file_ram;
5167 struct lfsck_assistant_data *lad = com->lc_data;
5168 struct lfsck_tgt_descs *ltds;
5169 struct lfsck_tgt_desc *ltd;
5173 if (lr->lr_event == LE_PAIRS_VERIFY) {
5176 rc = lfsck_layout_master_check_pairs(env, com, &lr->lr_fid,
5182 CDEBUG(D_LFSCK, "%s: layout LFSCK master handles notify %u "
5183 "from %s %x, status %d, flags %x, flags2 %x\n",
5184 lfsck_lfsck2name(lfsck), lr->lr_event,
5185 (lr->lr_flags & LEF_FROM_OST) ? "OST" : "MDT",
5186 lr->lr_index, lr->lr_status, lr->lr_flags, lr->lr_flags2);
5188 if (lr->lr_event != LE_PHASE1_DONE &&
5189 lr->lr_event != LE_PHASE2_DONE &&
5190 lr->lr_event != LE_PEER_EXIT)
5193 if (lr->lr_flags & LEF_FROM_OST)
5194 ltds = &lfsck->li_ost_descs;
5196 ltds = &lfsck->li_mdt_descs;
5197 spin_lock(<ds->ltd_lock);
5198 ltd = lfsck_ltd2tgt(ltds, lr->lr_index);
5200 spin_unlock(<ds->ltd_lock);
5205 list_del_init(<d->ltd_layout_phase_list);
5206 switch (lr->lr_event) {
5207 case LE_PHASE1_DONE:
5208 if (lr->lr_status <= 0 || lr->lr_flags2 & LF_INCOMPLETE) {
5209 if (lr->lr_flags2 & LF_INCOMPLETE) {
5210 if (lr->lr_flags & LEF_FROM_OST)
5211 lfsck_lad_set_bitmap(env, com,
5214 lo->ll_flags |= LF_INCOMPLETE;
5216 ltd->ltd_layout_done = 1;
5217 list_del_init(<d->ltd_layout_list);
5222 if (lr->lr_flags & LEF_FROM_OST) {
5223 if (list_empty(<d->ltd_layout_list))
5224 list_add_tail(<d->ltd_layout_list,
5225 &lad->lad_ost_list);
5226 list_add_tail(<d->ltd_layout_phase_list,
5227 &lad->lad_ost_phase2_list);
5229 if (list_empty(<d->ltd_layout_list))
5230 list_add_tail(<d->ltd_layout_list,
5231 &lad->lad_mdt_list);
5232 list_add_tail(<d->ltd_layout_phase_list,
5233 &lad->lad_mdt_phase2_list);
5236 case LE_PHASE2_DONE:
5237 ltd->ltd_layout_done = 1;
5238 if (!list_empty(<d->ltd_layout_list)) {
5239 list_del_init(<d->ltd_layout_list);
5240 if (lr->lr_flags2 & LF_INCOMPLETE) {
5241 lfsck_lad_set_bitmap(env, com, ltd->ltd_index);
5249 ltd->ltd_layout_done = 1;
5250 list_del_init(<d->ltd_layout_list);
5251 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) &&
5252 !(lr->lr_flags & LEF_FROM_OST))
5253 lo->ll_flags |= LF_INCOMPLETE;
5258 spin_unlock(<ds->ltd_lock);
5260 if (fail && lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
5261 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5263 memset(stop, 0, sizeof(*stop));
5264 stop->ls_status = lr->lr_status;
5265 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5266 lfsck_stop(env, lfsck->li_bottom, stop);
5267 } else if (lfsck_phase2_next_ready(lad)) {
5268 wake_up_all(&lad->lad_thread.t_ctl_waitq);
5274 static int lfsck_layout_slave_in_notify(const struct lu_env *env,
5275 struct lfsck_component *com,
5276 struct lfsck_request *lr,
5279 struct lfsck_instance *lfsck = com->lc_lfsck;
5280 struct lfsck_layout_slave_data *llsd = com->lc_data;
5281 struct lfsck_layout_slave_target *llst;
5285 switch (lr->lr_event) {
5286 case LE_FID_ACCESSED:
5287 lfsck_rbtree_update_bitmap(env, com, &lr->lr_fid, true);
5289 case LE_CONDITIONAL_DESTROY:
5290 rc = lfsck_layout_slave_conditional_destroy(env, com, lr);
5292 case LE_PAIRS_VERIFY: {
5293 lr->lr_status = LPVS_INIT;
5294 /* Firstly, if the MDT-object which is claimed via OST-object
5295 * local stored PFID xattr recognizes the OST-object, then it
5296 * must be that the client given PFID is wrong. */
5297 rc = lfsck_layout_slave_check_pairs(env, com, &lr->lr_fid,
5302 lr->lr_status = LPVS_INCONSISTENT;
5303 /* The OST-object local stored PFID xattr is stale. We need to
5304 * check whether the MDT-object that is claimed via the client
5305 * given PFID information recognizes the OST-object or not. If
5306 * matches, then need to update the OST-object's PFID xattr. */
5307 rc = lfsck_layout_slave_check_pairs(env, com, &lr->lr_fid,
5310 * We are not sure whether the client given PFID information
5311 * is correct or not, do nothing to avoid improper fixing.
5314 * The client given PFID information is also invalid, we can
5315 * NOT fix the OST-object inconsistency.
5320 lr->lr_status = LPVS_INCONSISTENT_TOFIX;
5321 rc = lfsck_layout_slave_repair_pfid(env, com, lr);
5325 case LE_PHASE1_DONE: {
5326 if (lr->lr_flags2 & LF_INCOMPLETE) {
5327 struct lfsck_layout *lo = com->lc_file_ram;
5329 lo->ll_flags |= LF_INCOMPLETE;
5330 llst = lfsck_layout_llst_find_and_del(llsd,
5334 lfsck_layout_llst_put(llst);
5335 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5341 case LE_PHASE2_DONE:
5343 CDEBUG(D_LFSCK, "%s: layout LFSCK slave handle notify %u "
5344 "from MDT %x, status %d\n", lfsck_lfsck2name(lfsck),
5345 lr->lr_event, lr->lr_index, lr->lr_status);
5351 llst = lfsck_layout_llst_find_and_del(llsd, lr->lr_index, true);
5355 lfsck_layout_llst_put(llst);
5356 if (list_empty(&llsd->llsd_master_list))
5357 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5359 if (lr->lr_event == LE_PEER_EXIT &&
5360 (lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT ||
5361 (list_empty(&llsd->llsd_master_list) &&
5362 (lr->lr_status == LS_STOPPED ||
5363 lr->lr_status == LS_CO_STOPPED)))) {
5364 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5366 memset(stop, 0, sizeof(*stop));
5367 stop->ls_status = lr->lr_status;
5368 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5369 lfsck_stop(env, lfsck->li_bottom, stop);
5375 static int lfsck_layout_query(const struct lu_env *env,
5376 struct lfsck_component *com)
5378 struct lfsck_layout *lo = com->lc_file_ram;
5380 return lo->ll_status;
5383 /* with lfsck::li_lock held */
5384 static int lfsck_layout_slave_join(const struct lu_env *env,
5385 struct lfsck_component *com,
5386 struct lfsck_start_param *lsp)
5388 struct lfsck_instance *lfsck = com->lc_lfsck;
5389 struct lfsck_layout_slave_data *llsd = com->lc_data;
5390 struct lfsck_layout_slave_target *llst;
5391 struct lfsck_start *start = lsp->lsp_start;
5395 if (start == NULL || !(start->ls_flags & LPF_OST_ORPHAN))
5398 if (!lsp->lsp_index_valid)
5401 /* If someone is running the LFSCK without orphan handling,
5402 * it will not maintain the object accessing rbtree. So we
5403 * cannot join it for orphan handling. */
5404 if (!llsd->llsd_rbtree_valid)
5407 spin_unlock(&lfsck->li_lock);
5408 rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
5409 spin_lock(&lfsck->li_lock);
5410 if (rc == 0 && !thread_is_running(&lfsck->li_thread)) {
5411 spin_unlock(&lfsck->li_lock);
5412 llst = lfsck_layout_llst_find_and_del(llsd, lsp->lsp_index,
5415 lfsck_layout_llst_put(llst);
5416 spin_lock(&lfsck->li_lock);
5423 static struct lfsck_operations lfsck_layout_master_ops = {
5424 .lfsck_reset = lfsck_layout_reset,
5425 .lfsck_fail = lfsck_layout_fail,
5426 .lfsck_checkpoint = lfsck_layout_master_checkpoint,
5427 .lfsck_prep = lfsck_layout_master_prep,
5428 .lfsck_exec_oit = lfsck_layout_master_exec_oit,
5429 .lfsck_exec_dir = lfsck_layout_exec_dir,
5430 .lfsck_post = lfsck_layout_master_post,
5431 .lfsck_dump = lfsck_layout_dump,
5432 .lfsck_double_scan = lfsck_layout_master_double_scan,
5433 .lfsck_data_release = lfsck_layout_master_data_release,
5434 .lfsck_quit = lfsck_layout_master_quit,
5435 .lfsck_in_notify = lfsck_layout_master_in_notify,
5436 .lfsck_query = lfsck_layout_query,
5439 static struct lfsck_operations lfsck_layout_slave_ops = {
5440 .lfsck_reset = lfsck_layout_reset,
5441 .lfsck_fail = lfsck_layout_fail,
5442 .lfsck_checkpoint = lfsck_layout_slave_checkpoint,
5443 .lfsck_prep = lfsck_layout_slave_prep,
5444 .lfsck_exec_oit = lfsck_layout_slave_exec_oit,
5445 .lfsck_exec_dir = lfsck_layout_exec_dir,
5446 .lfsck_post = lfsck_layout_slave_post,
5447 .lfsck_dump = lfsck_layout_dump,
5448 .lfsck_double_scan = lfsck_layout_slave_double_scan,
5449 .lfsck_data_release = lfsck_layout_slave_data_release,
5450 .lfsck_quit = lfsck_layout_slave_quit,
5451 .lfsck_in_notify = lfsck_layout_slave_in_notify,
5452 .lfsck_query = lfsck_layout_query,
5453 .lfsck_join = lfsck_layout_slave_join,
5456 static void lfsck_layout_assistant_fill_pos(const struct lu_env *env,
5457 struct lfsck_component *com,
5458 struct lfsck_position *pos)
5460 struct lfsck_assistant_data *lad = com->lc_data;
5461 struct lfsck_layout_req *llr;
5463 if (list_empty(&lad->lad_req_list))
5466 llr = list_entry(lad->lad_req_list.next,
5467 struct lfsck_layout_req,
5469 pos->lp_oit_cookie = llr->llr_lar.lar_parent->lso_oit_cookie - 1;
5472 struct lfsck_assistant_operations lfsck_layout_assistant_ops = {
5473 .la_handler_p1 = lfsck_layout_assistant_handler_p1,
5474 .la_handler_p2 = lfsck_layout_assistant_handler_p2,
5475 .la_fill_pos = lfsck_layout_assistant_fill_pos,
5476 .la_double_scan_result = lfsck_layout_double_scan_result,
5477 .la_req_fini = lfsck_layout_assistant_req_fini,
5478 .la_sync_failures = lfsck_layout_assistant_sync_failures,
5481 int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
5483 struct lfsck_component *com;
5484 struct lfsck_layout *lo;
5485 struct dt_object *root = NULL;
5486 struct dt_object *obj;
5494 INIT_LIST_HEAD(&com->lc_link);
5495 INIT_LIST_HEAD(&com->lc_link_dir);
5496 init_rwsem(&com->lc_sem);
5497 atomic_set(&com->lc_ref, 1);
5498 com->lc_lfsck = lfsck;
5499 com->lc_type = LFSCK_TYPE_LAYOUT;
5500 if (lfsck->li_master) {
5501 com->lc_ops = &lfsck_layout_master_ops;
5502 com->lc_data = lfsck_assistant_data_init(
5503 &lfsck_layout_assistant_ops,
5505 if (com->lc_data == NULL)
5506 GOTO(out, rc = -ENOMEM);
5508 struct lfsck_layout_slave_data *llsd;
5510 com->lc_ops = &lfsck_layout_slave_ops;
5511 OBD_ALLOC_PTR(llsd);
5513 GOTO(out, rc = -ENOMEM);
5515 INIT_LIST_HEAD(&llsd->llsd_seq_list);
5516 INIT_LIST_HEAD(&llsd->llsd_master_list);
5517 spin_lock_init(&llsd->llsd_lock);
5518 llsd->llsd_rb_root = RB_ROOT;
5519 rwlock_init(&llsd->llsd_rb_lock);
5520 com->lc_data = llsd;
5522 com->lc_file_size = sizeof(*lo);
5523 OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
5524 if (com->lc_file_ram == NULL)
5525 GOTO(out, rc = -ENOMEM);
5527 OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
5528 if (com->lc_file_disk == NULL)
5529 GOTO(out, rc = -ENOMEM);
5531 root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
5533 GOTO(out, rc = PTR_ERR(root));
5535 if (unlikely(!dt_try_as_dir(env, root)))
5536 GOTO(out, rc = -ENOTDIR);
5538 obj = local_file_find_or_create(env, lfsck->li_los, root,
5540 S_IFREG | S_IRUGO | S_IWUSR);
5542 GOTO(out, rc = PTR_ERR(obj));
5545 rc = lfsck_layout_load(env, com);
5547 rc = lfsck_layout_reset(env, com, true);
5548 else if (rc == -ENOENT)
5549 rc = lfsck_layout_init(env, com);
5554 lo = com->lc_file_ram;
5555 switch (lo->ll_status) {
5561 spin_lock(&lfsck->li_lock);
5562 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
5563 spin_unlock(&lfsck->li_lock);
5566 CERROR("%s: unknown lfsck_layout status %d\n",
5567 lfsck_lfsck2name(lfsck), lo->ll_status);
5569 case LS_SCANNING_PHASE1:
5570 case LS_SCANNING_PHASE2:
5571 /* No need to store the status to disk right now.
5572 * If the system crashed before the status stored,
5573 * it will be loaded back when next time. */
5574 lo->ll_status = LS_CRASHED;
5575 if (!lfsck->li_master)
5576 lo->ll_flags |= LF_INCOMPLETE;
5583 spin_lock(&lfsck->li_lock);
5584 list_add_tail(&com->lc_link, &lfsck->li_list_scan);
5585 spin_unlock(&lfsck->li_lock);
5589 if (lo->ll_flags & LF_CRASHED_LASTID) {
5590 LASSERT(lfsck->li_out_notify != NULL);
5592 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
5593 LE_LASTID_REBUILDING);
5599 if (root != NULL && !IS_ERR(root))
5600 lfsck_object_put(env, root);
5603 lfsck_component_cleanup(env, com);
5604 CERROR("%s: fail to init layout LFSCK component: rc = %d\n",
5605 lfsck_lfsck2name(lfsck), rc);
5611 struct lfsck_orphan_it {
5612 struct lfsck_component *loi_com;
5613 struct lfsck_rbtree_node *loi_lrn;
5614 struct lfsck_layout_slave_target *loi_llst;
5615 struct lu_fid loi_key;
5616 struct lu_orphan_rec loi_rec;
5618 unsigned int loi_over:1;
5621 static int lfsck_fid_match_idx(const struct lu_env *env,
5622 struct lfsck_instance *lfsck,
5623 const struct lu_fid *fid, int idx)
5625 struct seq_server_site *ss;
5626 struct lu_server_fld *sf;
5627 struct lu_seq_range *range = &lfsck_env_info(env)->lti_range;
5630 /* All abnormal cases will be returned to MDT0. */
5631 if (!fid_is_norm(fid)) {
5638 ss = lfsck_dev_site(lfsck);
5639 if (unlikely(ss == NULL))
5642 sf = ss->ss_server_fld;
5643 LASSERT(sf != NULL);
5645 fld_range_set_any(range);
5646 rc = fld_server_lookup(env, sf, fid_seq(fid), range);
5650 if (!fld_range_is_mdt(range))
5653 if (range->lsr_index == idx)
5659 static void lfsck_layout_destroy_orphan(const struct lu_env *env,
5660 struct dt_object *obj)
5662 struct dt_device *dev = lfsck_obj2dev(obj);
5663 struct thandle *handle;
5667 handle = dt_trans_create(env, dev);
5671 rc = dt_declare_ref_del(env, obj, handle);
5675 rc = dt_declare_destroy(env, obj, handle);
5679 rc = dt_trans_start_local(env, dev, handle);
5683 dt_write_lock(env, obj, 0);
5684 rc = dt_ref_del(env, obj, handle);
5686 rc = dt_destroy(env, obj, handle);
5687 dt_write_unlock(env, obj);
5692 dt_trans_stop(env, dev, handle);
5694 CDEBUG(D_LFSCK, "destroy orphan OST-object "DFID": rc = %d\n",
5695 PFID(lfsck_dto2fid(obj)), rc);
5700 static int lfsck_orphan_index_lookup(const struct lu_env *env,
5701 struct dt_object *dt,
5703 const struct dt_key *key)
5708 static int lfsck_orphan_index_declare_insert(const struct lu_env *env,
5709 struct dt_object *dt,
5710 const struct dt_rec *rec,
5711 const struct dt_key *key,
5712 struct thandle *handle)
5717 static int lfsck_orphan_index_insert(const struct lu_env *env,
5718 struct dt_object *dt,
5719 const struct dt_rec *rec,
5720 const struct dt_key *key,
5721 struct thandle *handle,
5727 static int lfsck_orphan_index_declare_delete(const struct lu_env *env,
5728 struct dt_object *dt,
5729 const struct dt_key *key,
5730 struct thandle *handle)
5735 static int lfsck_orphan_index_delete(const struct lu_env *env,
5736 struct dt_object *dt,
5737 const struct dt_key *key,
5738 struct thandle *handle)
5743 static struct dt_it *lfsck_orphan_it_init(const struct lu_env *env,
5744 struct dt_object *dt,
5747 struct dt_device *dev = lu2dt_dev(dt->do_lu.lo_dev);
5748 struct lfsck_instance *lfsck;
5749 struct lfsck_component *com = NULL;
5750 struct lfsck_layout_slave_data *llsd;
5751 struct lfsck_orphan_it *it = NULL;
5752 struct lfsck_layout *lo;
5756 lfsck = lfsck_instance_find(dev, true, false);
5757 if (unlikely(lfsck == NULL))
5758 RETURN(ERR_PTR(-ENXIO));
5760 com = lfsck_component_find(lfsck, LFSCK_TYPE_LAYOUT);
5761 if (unlikely(com == NULL))
5762 GOTO(out, rc = -ENOENT);
5764 lo = com->lc_file_ram;
5765 if (lo->ll_flags & LF_INCOMPLETE)
5766 GOTO(out, rc = -ESRCH);
5768 llsd = com->lc_data;
5769 if (!llsd->llsd_rbtree_valid)
5770 GOTO(out, rc = -ESRCH);
5774 GOTO(out, rc = -ENOMEM);
5776 it->loi_llst = lfsck_layout_llst_find_and_del(llsd, attr, false);
5777 if (it->loi_llst == NULL)
5778 GOTO(out, rc = -ENXIO);
5780 if (dev->dd_record_fid_accessed) {
5781 /* The first iteration against the rbtree, scan the whole rbtree
5782 * to remove the nodes which do NOT need to be handled. */
5783 write_lock(&llsd->llsd_rb_lock);
5784 if (dev->dd_record_fid_accessed) {
5785 struct rb_node *node;
5786 struct rb_node *next;
5787 struct lfsck_rbtree_node *lrn;
5789 /* No need to record the fid accessing anymore. */
5790 dev->dd_record_fid_accessed = 0;
5792 node = rb_first(&llsd->llsd_rb_root);
5793 while (node != NULL) {
5794 next = rb_next(node);
5795 lrn = rb_entry(node, struct lfsck_rbtree_node,
5797 if (atomic_read(&lrn->lrn_known_count) <=
5798 atomic_read(&lrn->lrn_accessed_count)) {
5799 rb_erase(node, &llsd->llsd_rb_root);
5800 lfsck_rbtree_free(lrn);
5805 write_unlock(&llsd->llsd_rb_lock);
5808 /* read lock the rbtree when init, and unlock when fini */
5809 read_lock(&llsd->llsd_rb_lock);
5817 lfsck_component_put(env, com);
5819 CDEBUG(D_LFSCK, "%s: init the orphan iteration: rc = %d\n",
5820 lfsck_lfsck2name(lfsck), rc);
5822 lfsck_instance_put(env, lfsck);
5827 it = (struct lfsck_orphan_it *)ERR_PTR(rc);
5830 return (struct dt_it *)it;
5833 static void lfsck_orphan_it_fini(const struct lu_env *env,
5836 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5837 struct lfsck_component *com = it->loi_com;
5838 struct lfsck_layout_slave_data *llsd;
5839 struct lfsck_layout_slave_target *llst;
5842 CDEBUG(D_LFSCK, "%s: fini the orphan iteration\n",
5843 lfsck_lfsck2name(com->lc_lfsck));
5845 llsd = com->lc_data;
5846 read_unlock(&llsd->llsd_rb_lock);
5847 llst = it->loi_llst;
5848 LASSERT(llst != NULL);
5850 /* Save the key and hash for iterate next. */
5851 llst->llst_fid = it->loi_key;
5852 llst->llst_hash = it->loi_hash;
5853 lfsck_layout_llst_put(llst);
5854 lfsck_component_put(env, com);
5860 * \retval +1: the iteration finished
5861 * \retval 0: on success, not finished
5862 * \retval -ve: on error
5864 static int lfsck_orphan_it_next(const struct lu_env *env,
5867 struct lfsck_thread_info *info = lfsck_env_info(env);
5868 struct filter_fid_old *pfid = &info->lti_old_pfid;
5869 struct lu_attr *la = &info->lti_la;
5870 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5871 struct lu_fid *key = &it->loi_key;
5872 struct lu_orphan_rec *rec = &it->loi_rec;
5873 struct lfsck_component *com = it->loi_com;
5874 struct lfsck_instance *lfsck = com->lc_lfsck;
5875 struct lfsck_layout_slave_data *llsd = com->lc_data;
5876 struct dt_object *obj;
5877 struct lfsck_rbtree_node *lrn;
5881 __u32 idx = it->loi_llst->llst_index;
5891 lrn = lfsck_rbtree_search(llsd, key, &exact);
5899 key->f_seq = lrn->lrn_seq;
5900 key->f_oid = lrn->lrn_first_oid;
5905 if (unlikely(key->f_oid == 0)) {
5912 lrn->lrn_first_oid + LFSCK_RBTREE_BITMAP_WIDTH) {
5918 if (unlikely(atomic_read(&lrn->lrn_known_count) <=
5919 atomic_read(&lrn->lrn_accessed_count))) {
5920 struct rb_node *next = rb_next(&lrn->lrn_node);
5922 while (next != NULL) {
5923 lrn = rb_entry(next, struct lfsck_rbtree_node,
5925 if (atomic_read(&lrn->lrn_known_count) >
5926 atomic_read(&lrn->lrn_accessed_count))
5928 next = rb_next(next);
5937 key->f_seq = lrn->lrn_seq;
5938 key->f_oid = lrn->lrn_first_oid;
5942 pos = key->f_oid - lrn->lrn_first_oid;
5945 pos = find_next_bit(lrn->lrn_known_bitmap,
5946 LFSCK_RBTREE_BITMAP_WIDTH, pos);
5947 if (pos >= LFSCK_RBTREE_BITMAP_WIDTH) {
5948 key->f_oid = lrn->lrn_first_oid + pos;
5949 if (unlikely(key->f_oid < lrn->lrn_first_oid)) {
5957 if (test_bit(pos, lrn->lrn_accessed_bitmap)) {
5962 key->f_oid = lrn->lrn_first_oid + pos;
5963 obj = lfsck_object_find_bottom(env, lfsck, key);
5966 if (rc == -ENOENT) {
5973 dt_read_lock(env, obj, 0);
5974 if (dt_object_exists(obj) == 0 ||
5975 lfsck_is_dead_obj(obj)) {
5976 dt_read_unlock(env, obj);
5977 lfsck_object_put(env, obj);
5982 rc = dt_attr_get(env, obj, la);
5986 rc = dt_xattr_get(env, obj, lfsck_buf_get(env, pfid, sizeof(*pfid)),
5988 if (rc == -ENODATA) {
5989 /* For the pre-created OST-object, update the bitmap to avoid
5990 * others LFSCK (second phase) iteration to touch it again. */
5991 if (la->la_ctime == 0) {
5992 if (!test_and_set_bit(pos, lrn->lrn_accessed_bitmap))
5993 atomic_inc(&lrn->lrn_accessed_count);
5995 /* For the race between repairing dangling referenced
5996 * MDT-object and unlink the file, it may left orphan
5997 * OST-object there. Destroy it now! */
5998 if (unlikely(!(la->la_mode & S_ISUID))) {
5999 dt_read_unlock(env, obj);
6000 lfsck_layout_destroy_orphan(env, obj);
6001 lfsck_object_put(env, obj);
6005 } else if (idx == 0) {
6006 /* If the orphan OST-object has no parent information,
6007 * regard it as referenced by the MDT-object on MDT0. */
6008 fid_zero(&rec->lor_fid);
6009 rec->lor_uid = la->la_uid;
6010 rec->lor_gid = la->la_gid;
6014 dt_read_unlock(env, obj);
6015 lfsck_object_put(env, obj);
6023 if (rc != sizeof(struct filter_fid) &&
6024 rc != sizeof(struct filter_fid_old))
6025 GOTO(out, rc = -EINVAL);
6027 fid_le_to_cpu(&rec->lor_fid, &pfid->ff_parent);
6028 /* Currently, the filter_fid::ff_parent::f_ver is not the real parent
6029 * MDT-object's FID::f_ver, instead it is the OST-object index in its
6030 * parent MDT-object's layout EA. */
6031 save = rec->lor_fid.f_stripe_idx;
6032 rec->lor_fid.f_ver = 0;
6033 rc = lfsck_fid_match_idx(env, lfsck, &rec->lor_fid, idx);
6034 /* If the orphan OST-object does not claim the MDT, then next.
6036 * If we do not know whether it matches or not, then return it
6037 * to the MDT for further check. */
6039 dt_read_unlock(env, obj);
6040 lfsck_object_put(env, obj);
6045 rec->lor_fid.f_stripe_idx = save;
6046 rec->lor_uid = la->la_uid;
6047 rec->lor_gid = la->la_gid;
6049 CDEBUG(D_LFSCK, "%s: return orphan "DFID", PFID "DFID", owner %u:%u\n",
6050 lfsck_lfsck2name(com->lc_lfsck), PFID(key), PFID(&rec->lor_fid),
6051 rec->lor_uid, rec->lor_gid);
6056 dt_read_unlock(env, obj);
6057 lfsck_object_put(env, obj);
6065 * \retval +1: locate to the exactly position
6066 * \retval 0: cannot locate to the exactly position,
6067 * call next() to move to a valid position.
6068 * \retval -ve: on error
6070 static int lfsck_orphan_it_get(const struct lu_env *env,
6072 const struct dt_key *key)
6074 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6077 it->loi_key = *(struct lu_fid *)key;
6078 rc = lfsck_orphan_it_next(env, di);
6088 static void lfsck_orphan_it_put(const struct lu_env *env,
6093 static struct dt_key *lfsck_orphan_it_key(const struct lu_env *env,
6094 const struct dt_it *di)
6096 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6098 return (struct dt_key *)&it->loi_key;
6101 static int lfsck_orphan_it_key_size(const struct lu_env *env,
6102 const struct dt_it *di)
6104 return sizeof(struct lu_fid);
6107 static int lfsck_orphan_it_rec(const struct lu_env *env,
6108 const struct dt_it *di,
6112 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6114 *(struct lu_orphan_rec *)rec = it->loi_rec;
6119 static __u64 lfsck_orphan_it_store(const struct lu_env *env,
6120 const struct dt_it *di)
6122 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6124 return it->loi_hash;
6128 * \retval +1: locate to the exactly position
6129 * \retval 0: cannot locate to the exactly position,
6130 * call next() to move to a valid position.
6131 * \retval -ve: on error
6133 static int lfsck_orphan_it_load(const struct lu_env *env,
6134 const struct dt_it *di,
6137 struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
6138 struct lfsck_layout_slave_target *llst = it->loi_llst;
6141 LASSERT(llst != NULL);
6143 if (hash != llst->llst_hash) {
6144 CDEBUG(D_LFSCK, "%s: the given hash "LPU64" for orphan "
6145 "iteration does not match the one when fini "
6146 LPU64", to be reset.\n",
6147 lfsck_lfsck2name(it->loi_com->lc_lfsck), hash,
6149 fid_zero(&llst->llst_fid);
6150 llst->llst_hash = 0;
6153 it->loi_key = llst->llst_fid;
6154 it->loi_hash = llst->llst_hash;
6155 rc = lfsck_orphan_it_next(env, (struct dt_it *)di);
6165 static int lfsck_orphan_it_key_rec(const struct lu_env *env,
6166 const struct dt_it *di,
6172 const struct dt_index_operations lfsck_orphan_index_ops = {
6173 .dio_lookup = lfsck_orphan_index_lookup,
6174 .dio_declare_insert = lfsck_orphan_index_declare_insert,
6175 .dio_insert = lfsck_orphan_index_insert,
6176 .dio_declare_delete = lfsck_orphan_index_declare_delete,
6177 .dio_delete = lfsck_orphan_index_delete,
6179 .init = lfsck_orphan_it_init,
6180 .fini = lfsck_orphan_it_fini,
6181 .get = lfsck_orphan_it_get,
6182 .put = lfsck_orphan_it_put,
6183 .next = lfsck_orphan_it_next,
6184 .key = lfsck_orphan_it_key,
6185 .key_size = lfsck_orphan_it_key_size,
6186 .rec = lfsck_orphan_it_rec,
6187 .store = lfsck_orphan_it_store,
6188 .load = lfsck_orphan_it_load,
6189 .key_rec = lfsck_orphan_it_key_rec,