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) 2012, 2013, Intel Corporation.
26 * lustre/lfsck/lfsck_engine.c
28 * Author: Fan, Yong <fan.yong@intel.com>
31 #define DEBUG_SUBSYSTEM S_LFSCK
33 #include <lu_object.h>
34 #include <dt_object.h>
35 #include <lustre_net.h>
36 #include <lustre_fid.h>
37 #include <obd_support.h>
38 #include <lustre_lib.h>
40 #include "lfsck_internal.h"
42 static void lfsck_unpack_ent(struct lu_dirent *ent, __u64 *cookie)
44 fid_le_to_cpu(&ent->lde_fid, &ent->lde_fid);
45 *cookie = le64_to_cpu(ent->lde_hash);
46 ent->lde_reclen = le16_to_cpu(ent->lde_reclen);
47 ent->lde_namelen = le16_to_cpu(ent->lde_namelen);
48 ent->lde_attrs = le32_to_cpu(ent->lde_attrs);
50 /* Make sure the name is terminated with '0'.
51 * The data (type) after ent::lde_name maybe
52 * broken, but we do not care. */
53 ent->lde_name[ent->lde_namelen] = 0;
56 static void lfsck_di_oit_put(const struct lu_env *env, struct lfsck_instance *lfsck)
58 const struct dt_it_ops *iops;
61 spin_lock(&lfsck->li_lock);
62 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
63 di = lfsck->li_di_oit;
64 lfsck->li_di_oit = NULL;
65 spin_unlock(&lfsck->li_lock);
69 static void lfsck_di_dir_put(const struct lu_env *env, struct lfsck_instance *lfsck)
71 const struct dt_it_ops *iops;
74 spin_lock(&lfsck->li_lock);
75 iops = &lfsck->li_obj_dir->do_index_ops->dio_it;
76 di = lfsck->li_di_dir;
77 lfsck->li_di_dir = NULL;
78 lfsck->li_cookie_dir = 0;
79 spin_unlock(&lfsck->li_lock);
83 static void lfsck_close_dir(const struct lu_env *env,
84 struct lfsck_instance *lfsck)
86 struct dt_object *dir_obj = lfsck->li_obj_dir;
87 const struct dt_it_ops *dir_iops = &dir_obj->do_index_ops->dio_it;
88 struct dt_it *dir_di = lfsck->li_di_dir;
90 lfsck_di_dir_put(env, lfsck);
91 dir_iops->fini(env, dir_di);
92 lfsck->li_obj_dir = NULL;
93 lfsck_object_put(env, dir_obj);
96 static int lfsck_update_lma(const struct lu_env *env,
97 struct lfsck_instance *lfsck, struct dt_object *obj)
99 struct lfsck_thread_info *info = lfsck_env_info(env);
100 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
101 struct dt_device *dt = lfsck->li_bottom;
102 struct lustre_mdt_attrs *lma = &info->lti_lma;
109 if (bk->lb_param & LPF_DRYRUN)
112 buf = lfsck_buf_get(env, info->lti_lma_old, LMA_OLD_SIZE);
113 rc = dt_xattr_get(env, obj, buf, XATTR_NAME_LMA, BYPASS_CAPA);
118 fl = LU_XATTR_CREATE;
119 lustre_lma_init(lma, lfsck_dto2fid(obj), LMAC_FID_ON_OST, 0);
121 if (rc != LMA_OLD_SIZE && rc != sizeof(struct lustre_mdt_attrs))
124 fl = LU_XATTR_REPLACE;
125 lustre_lma_swab(lma);
126 lustre_lma_init(lma, lfsck_dto2fid(obj),
127 lma->lma_compat | LMAC_FID_ON_OST,
130 lustre_lma_swab(lma);
132 th = dt_trans_create(env, dt);
136 buf = lfsck_buf_get(env, lma, sizeof(*lma));
137 rc = dt_declare_xattr_set(env, obj, buf, XATTR_NAME_LMA, fl, th);
141 rc = dt_trans_start(env, dt, th);
145 rc = dt_xattr_set(env, obj, buf, XATTR_NAME_LMA, fl, th, BYPASS_CAPA);
150 dt_trans_stop(env, dt, th);
154 static int lfsck_parent_fid(const struct lu_env *env, struct dt_object *obj,
157 if (unlikely(!S_ISDIR(lfsck_object_type(obj)) ||
158 !dt_try_as_dir(env, obj)))
161 return dt_lookup(env, obj, (struct dt_rec *)fid,
162 (const struct dt_key *)"..", BYPASS_CAPA);
165 static int lfsck_needs_scan_dir(const struct lu_env *env,
166 struct lfsck_instance *lfsck,
167 struct dt_object *obj)
169 struct lu_fid *fid = &lfsck_env_info(env)->lti_fid;
173 if (list_empty(&lfsck->li_list_dir) || !S_ISDIR(lfsck_object_type(obj)))
177 /* XXX: Currently, we do not scan the "/REMOTE_PARENT_DIR",
178 * which is the agent directory to manage the objects
179 * which name entries reside on remote MDTs. Related
180 * consistency verification will be processed in LFSCK
182 if (lu_fid_eq(lfsck_dto2fid(obj), &lfsck->li_global_root_fid)) {
184 lfsck_object_put(env, obj);
188 /* No need to check .lustre and its children. */
189 if (fid_seq_is_dot(fid_seq(lfsck_dto2fid(obj)))) {
191 lfsck_object_put(env, obj);
195 dt_read_lock(env, obj, MOR_TGT_CHILD);
196 if (unlikely(lfsck_is_dead_obj(obj))) {
197 dt_read_unlock(env, obj);
199 lfsck_object_put(env, obj);
203 rc = dt_xattr_get(env, obj,
204 lfsck_buf_get(env, NULL, 0), XATTR_NAME_LINK,
206 dt_read_unlock(env, obj);
209 lfsck_object_put(env, obj);
213 if (rc < 0 && rc != -ENODATA) {
215 lfsck_object_put(env, obj);
219 rc = lfsck_parent_fid(env, obj, fid);
221 lfsck_object_put(env, obj);
225 if (unlikely(lu_fid_eq(fid, &lfsck->li_local_root_fid)))
228 obj = lfsck_object_find(env, lfsck, fid);
232 if (!dt_object_exists(obj)) {
233 lfsck_object_put(env, obj);
237 if (dt_object_remote(obj)) {
238 /* .lustre/lost+found/MDTxxx can be remote directory. */
239 if (fid_seq_is_dot(fid_seq(lfsck_dto2fid(obj))))
242 /* Other remote directory should be client
243 * visible and need to be checked. */
245 lfsck_object_put(env, obj);
254 /* LFSCK wrap functions */
256 static void lfsck_fail(const struct lu_env *env, struct lfsck_instance *lfsck,
259 struct lfsck_component *com;
261 list_for_each_entry(com, &lfsck->li_list_scan, lc_link) {
262 com->lc_ops->lfsck_fail(env, com, new_checked);
266 static int lfsck_checkpoint(const struct lu_env *env,
267 struct lfsck_instance *lfsck)
269 struct lfsck_component *com;
273 if (likely(cfs_time_beforeq(cfs_time_current(),
274 lfsck->li_time_next_checkpoint)))
277 lfsck_pos_fill(env, lfsck, &lfsck->li_pos_current, false);
278 list_for_each_entry(com, &lfsck->li_list_scan, lc_link) {
279 rc = com->lc_ops->lfsck_checkpoint(env, com, false);
284 lfsck->li_time_last_checkpoint = cfs_time_current();
285 lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
286 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
287 return rc1 != 0 ? rc1 : rc;
290 static int lfsck_prep(const struct lu_env *env, struct lfsck_instance *lfsck,
291 struct lfsck_start_param *lsp)
293 struct dt_object *obj = NULL;
294 struct lfsck_component *com;
295 struct lfsck_component *next;
296 struct lfsck_position *pos = NULL;
297 const struct dt_it_ops *iops =
298 &lfsck->li_obj_oit->do_index_ops->dio_it;
303 LASSERT(lfsck->li_obj_dir == NULL);
304 LASSERT(lfsck->li_di_dir == NULL);
306 lfsck->li_current_oit_processed = 0;
307 list_for_each_entry_safe(com, next, &lfsck->li_list_scan, lc_link) {
308 com->lc_new_checked = 0;
309 if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
312 rc = com->lc_ops->lfsck_prep(env, com, lsp);
317 (!lfsck_pos_is_zero(&com->lc_pos_start) &&
318 lfsck_pos_is_eq(pos, &com->lc_pos_start) > 0))
319 pos = &com->lc_pos_start;
322 /* Init otable-based iterator. */
324 rc = iops->load(env, lfsck->li_di_oit, 0);
326 lfsck->li_oit_over = 1;
333 rc = iops->load(env, lfsck->li_di_oit, pos->lp_oit_cookie);
337 lfsck->li_oit_over = 1;
339 if (!lfsck->li_master || fid_is_zero(&pos->lp_dir_parent))
342 /* Find the directory for namespace-based traverse. */
343 obj = lfsck_object_find(env, lfsck, &pos->lp_dir_parent);
345 RETURN(PTR_ERR(obj));
347 /* XXX: Currently, skip remote object, the consistency for
348 * remote object will be processed in LFSCK phase III. */
349 if (!dt_object_exists(obj) || dt_object_remote(obj) ||
350 unlikely(!S_ISDIR(lfsck_object_type(obj))))
353 if (unlikely(!dt_try_as_dir(env, obj)))
354 GOTO(out, rc = -ENOTDIR);
356 /* Init the namespace-based directory traverse. */
357 iops = &obj->do_index_ops->dio_it;
358 di = iops->init(env, obj, lfsck->li_args_dir, BYPASS_CAPA);
360 GOTO(out, rc = PTR_ERR(di));
362 LASSERT(pos->lp_dir_cookie < MDS_DIR_END_OFF);
364 rc = iops->load(env, di, pos->lp_dir_cookie);
365 if ((rc == 0) || (rc > 0 && pos->lp_dir_cookie > 0))
366 rc = iops->next(env, di);
376 lfsck->li_obj_dir = lfsck_object_get(obj);
377 lfsck->li_cookie_dir = iops->store(env, di);
378 spin_lock(&lfsck->li_lock);
379 lfsck->li_di_dir = di;
380 spin_unlock(&lfsck->li_lock);
386 lfsck_object_put(env, obj);
389 list_for_each_entry_safe(com, next, &lfsck->li_list_scan,
391 com->lc_ops->lfsck_post(env, com, rc, true);
397 lfsck_pos_fill(env, lfsck, &lfsck->li_pos_current, true);
398 list_for_each_entry(com, &lfsck->li_list_scan, lc_link) {
399 rc = com->lc_ops->lfsck_checkpoint(env, com, true);
404 lfsck->li_time_last_checkpoint = cfs_time_current();
405 lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
406 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
410 static int lfsck_exec_oit(const struct lu_env *env,
411 struct lfsck_instance *lfsck, struct dt_object *obj)
413 struct lfsck_component *com;
414 const struct dt_it_ops *iops;
419 LASSERT(lfsck->li_obj_dir == NULL);
421 list_for_each_entry(com, &lfsck->li_list_scan, lc_link) {
422 rc = com->lc_ops->lfsck_exec_oit(env, com, obj);
427 rc = lfsck_needs_scan_dir(env, lfsck, obj);
431 if (unlikely(!dt_try_as_dir(env, obj)))
432 GOTO(out, rc = -ENOTDIR);
434 iops = &obj->do_index_ops->dio_it;
435 di = iops->init(env, obj, lfsck->li_args_dir, BYPASS_CAPA);
437 GOTO(out, rc = PTR_ERR(di));
439 rc = iops->load(env, di, 0);
441 rc = iops->next(env, di);
451 lfsck->li_obj_dir = lfsck_object_get(obj);
452 lfsck->li_cookie_dir = iops->store(env, di);
453 spin_lock(&lfsck->li_lock);
454 lfsck->li_di_dir = di;
455 spin_unlock(&lfsck->li_lock);
461 lfsck_fail(env, lfsck, false);
462 return (rc > 0 ? 0 : rc);
465 static int lfsck_exec_dir(const struct lu_env *env,
466 struct lfsck_instance *lfsck,
467 struct dt_object *obj, struct lu_dirent *ent)
469 struct lfsck_component *com;
472 list_for_each_entry(com, &lfsck->li_list_scan, lc_link) {
473 rc = com->lc_ops->lfsck_exec_dir(env, com, obj, ent);
480 static int lfsck_post(const struct lu_env *env, struct lfsck_instance *lfsck,
483 struct lfsck_component *com;
484 struct lfsck_component *next;
488 lfsck_pos_fill(env, lfsck, &lfsck->li_pos_current, false);
489 list_for_each_entry_safe(com, next, &lfsck->li_list_scan, lc_link) {
490 rc = com->lc_ops->lfsck_post(env, com, result, false);
495 lfsck->li_time_last_checkpoint = cfs_time_current();
496 lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
497 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
499 /* Ignore some component post failure to make other can go ahead. */
503 static int lfsck_double_scan(const struct lu_env *env,
504 struct lfsck_instance *lfsck)
506 struct lfsck_component *com;
507 struct lfsck_component *next;
508 struct l_wait_info lwi = { 0 };
512 list_for_each_entry(com, &lfsck->li_list_double_scan, lc_link) {
513 if (lfsck->li_bookmark_ram.lb_param & LPF_DRYRUN)
516 rc = com->lc_ops->lfsck_double_scan(env, com);
521 l_wait_event(lfsck->li_thread.t_ctl_waitq,
522 atomic_read(&lfsck->li_double_scan_count) == 0,
525 if (lfsck->li_status != LS_PAUSED &&
526 lfsck->li_status != LS_CO_PAUSED) {
527 list_for_each_entry_safe(com, next, &lfsck->li_list_double_scan,
529 spin_lock(&lfsck->li_lock);
530 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
531 spin_unlock(&lfsck->li_lock);
535 return rc1 != 0 ? rc1 : rc;
538 static void lfsck_quit(const struct lu_env *env, struct lfsck_instance *lfsck)
540 struct lfsck_component *com;
541 struct lfsck_component *next;
543 list_for_each_entry_safe(com, next, &lfsck->li_list_scan,
545 if (com->lc_ops->lfsck_quit != NULL)
546 com->lc_ops->lfsck_quit(env, com);
548 spin_lock(&lfsck->li_lock);
549 list_del_init(&com->lc_link_dir);
550 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
551 spin_unlock(&lfsck->li_lock);
554 list_for_each_entry_safe(com, next, &lfsck->li_list_double_scan,
556 if (com->lc_ops->lfsck_quit != NULL)
557 com->lc_ops->lfsck_quit(env, com);
559 spin_lock(&lfsck->li_lock);
560 list_move_tail(&com->lc_link, &lfsck->li_list_idle);
561 spin_unlock(&lfsck->li_lock);
567 static int lfsck_master_dir_engine(const struct lu_env *env,
568 struct lfsck_instance *lfsck)
570 struct lfsck_thread_info *info = lfsck_env_info(env);
571 struct dt_object *dir = lfsck->li_obj_dir;
572 const struct dt_it_ops *iops = &dir->do_index_ops->dio_it;
573 struct dt_it *di = lfsck->li_di_dir;
574 struct lu_dirent *ent =
575 (struct lu_dirent *)info->lti_key;
576 struct lu_fid *fid = &info->lti_fid;
577 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
578 struct ptlrpc_thread *thread = &lfsck->li_thread;
583 struct dt_object *child;
585 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY2) &&
587 struct l_wait_info lwi;
589 lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val),
591 l_wait_event(thread->t_ctl_waitq,
592 !thread_is_running(thread),
596 lfsck->li_new_scanned++;
597 rc = iops->rec(env, di, (struct dt_rec *)ent,
599 lfsck_unpack_ent(ent, &lfsck->li_cookie_dir);
601 CDEBUG(D_LFSCK, "%s: scan dir failed at rec(), "
602 "parent "DFID", cookie "LPX64": rc = %d\n",
603 lfsck_lfsck2name(lfsck),
604 PFID(lfsck_dto2fid(dir)),
605 lfsck->li_cookie_dir, rc);
606 lfsck_fail(env, lfsck, true);
607 if (bk->lb_param & LPF_FAILOUT)
613 if (ent->lde_attrs & LUDA_IGNORE)
617 child = lfsck_object_find(env, lfsck, fid);
619 CDEBUG(D_LFSCK, "%s: scan dir failed at find target, "
620 "parent "DFID", child %.*s "DFID": rc = %d\n",
621 lfsck_lfsck2name(lfsck),
622 PFID(lfsck_dto2fid(dir)),
623 ent->lde_namelen, ent->lde_name,
624 PFID(&ent->lde_fid), rc);
625 lfsck_fail(env, lfsck, true);
626 if (bk->lb_param & LPF_FAILOUT)
627 RETURN(PTR_ERR(child));
632 /* XXX: Currently, skip remote object, the consistency for
633 * remote object will be processed in LFSCK phase III. */
634 if (dt_object_exists(child) && !dt_object_remote(child))
635 rc = lfsck_exec_dir(env, lfsck, child, ent);
636 lfsck_object_put(env, child);
637 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
641 rc = lfsck_checkpoint(env, lfsck);
642 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
646 lfsck_control_speed(lfsck);
647 if (unlikely(!thread_is_running(thread))) {
648 CDEBUG(D_LFSCK, "%s: scan dir exit for engine stop, "
649 "parent "DFID", cookie "LPX64"\n",
650 lfsck_lfsck2name(lfsck),
651 PFID(lfsck_dto2fid(dir)),
652 lfsck->li_cookie_dir);
656 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_FATAL2)) {
657 spin_lock(&lfsck->li_lock);
658 thread_set_flags(thread, SVC_STOPPING);
659 spin_unlock(&lfsck->li_lock);
663 rc = iops->next(env, di);
666 if (rc > 0 && !lfsck->li_oit_over)
667 lfsck_close_dir(env, lfsck);
672 static int lfsck_master_oit_engine(const struct lu_env *env,
673 struct lfsck_instance *lfsck)
675 struct lfsck_thread_info *info = lfsck_env_info(env);
676 const struct dt_it_ops *iops =
677 &lfsck->li_obj_oit->do_index_ops->dio_it;
678 struct dt_it *di = lfsck->li_di_oit;
679 struct lu_fid *fid = &info->lti_fid;
680 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
681 struct ptlrpc_thread *thread = &lfsck->li_thread;
683 lfsck_dev_idx(lfsck->li_bottom);
688 struct dt_object *target;
689 bool update_lma = false;
691 if (lfsck->li_di_dir != NULL) {
692 rc = lfsck_master_dir_engine(env, lfsck);
697 if (unlikely(lfsck->li_oit_over))
700 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY1) &&
702 struct l_wait_info lwi;
704 lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val),
706 l_wait_event(thread->t_ctl_waitq,
707 !thread_is_running(thread),
711 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CRASH))
714 lfsck->li_current_oit_processed = 1;
715 lfsck->li_new_scanned++;
716 rc = iops->rec(env, di, (struct dt_rec *)fid, 0);
718 CDEBUG(D_LFSCK, "%s: OIT scan failed at rec(): "
719 "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
720 lfsck_fail(env, lfsck, true);
721 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
727 if (fid_is_idif(fid)) {
728 __u32 idx1 = fid_idif_ost_idx(fid);
730 LASSERT(!lfsck->li_master);
732 /* It is an old format device, update the LMA. */
734 struct ost_id *oi = &info->lti_oi;
736 fid_to_ostid(fid, oi);
737 ostid_to_fid(fid, oi, idx);
740 } else if (!fid_is_norm(fid) && !fid_is_igif(fid) &&
741 !fid_is_last_id(fid) && !fid_is_root(fid) &&
742 !fid_seq_is_dot(fid_seq(fid))) {
743 /* If the FID/object is only used locally and invisible
744 * to external nodes, then LFSCK will not handle it. */
748 target = lfsck_object_find(env, lfsck, fid);
749 if (IS_ERR(target)) {
750 CDEBUG(D_LFSCK, "%s: OIT scan failed at find target "
751 DFID", cookie "LPU64": rc = %d\n",
752 lfsck_lfsck2name(lfsck), PFID(fid),
753 iops->store(env, di), rc);
754 lfsck_fail(env, lfsck, true);
755 if (bk->lb_param & LPF_FAILOUT)
756 RETURN(PTR_ERR(target));
761 /* XXX: Currently, skip remote object, the consistency for
762 * remote object will be processed in LFSCK phase III. */
763 if (dt_object_exists(target) && !dt_object_remote(target)) {
765 rc = lfsck_update_lma(env, lfsck, target);
767 CDEBUG(D_LFSCK, "%s: fail to update "
768 "LMA for "DFID": rc = %d\n",
769 lfsck_lfsck2name(lfsck),
770 PFID(lfsck_dto2fid(target)), rc);
773 rc = lfsck_exec_oit(env, lfsck, target);
775 lfsck_object_put(env, target);
776 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
780 rc = lfsck_checkpoint(env, lfsck);
781 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
785 lfsck_control_speed(lfsck);
787 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_FATAL1)) {
788 spin_lock(&lfsck->li_lock);
789 thread_set_flags(thread, SVC_STOPPING);
790 spin_unlock(&lfsck->li_lock);
794 rc = iops->next(env, di);
795 if (unlikely(rc > 0))
796 lfsck->li_oit_over = 1;
797 else if (likely(rc == 0))
798 lfsck->li_current_oit_processed = 0;
800 if (unlikely(!thread_is_running(thread))) {
801 CDEBUG(D_LFSCK, "%s: OIT scan exit for engine stop, "
802 "cookie "LPU64"\n", lfsck_lfsck2name(lfsck),
803 iops->store(env, di));
806 } while (rc == 0 || lfsck->li_di_dir != NULL);
811 int lfsck_master_engine(void *args)
813 struct lfsck_thread_args *lta = args;
814 struct lu_env *env = <a->lta_env;
815 struct lfsck_instance *lfsck = lta->lta_lfsck;
816 struct ptlrpc_thread *thread = &lfsck->li_thread;
817 struct dt_object *oit_obj = lfsck->li_obj_oit;
818 const struct dt_it_ops *oit_iops = &oit_obj->do_index_ops->dio_it;
819 struct dt_it *oit_di;
820 struct l_wait_info lwi = { 0 };
824 oit_di = oit_iops->init(env, oit_obj, lfsck->li_args_oit, BYPASS_CAPA);
825 if (IS_ERR(oit_di)) {
826 rc = PTR_ERR(oit_di);
827 CDEBUG(D_LFSCK, "%s: master engine fail to init iteration: "
828 "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
833 spin_lock(&lfsck->li_lock);
834 lfsck->li_di_oit = oit_di;
835 spin_unlock(&lfsck->li_lock);
836 rc = lfsck_prep(env, lfsck, lta->lta_lsp);
840 CDEBUG(D_LFSCK, "LFSCK entry: oit_flags = %#x, dir_flags = %#x, "
841 "oit_cookie = "LPU64", dir_cookie = "LPX64", parent = "DFID
842 ", pid = %d\n", lfsck->li_args_oit, lfsck->li_args_dir,
843 lfsck->li_pos_current.lp_oit_cookie,
844 lfsck->li_pos_current.lp_dir_cookie,
845 PFID(&lfsck->li_pos_current.lp_dir_parent),
848 spin_lock(&lfsck->li_lock);
849 thread_set_flags(thread, SVC_RUNNING);
850 spin_unlock(&lfsck->li_lock);
851 wake_up_all(&thread->t_ctl_waitq);
853 l_wait_event(thread->t_ctl_waitq,
854 lfsck->li_start_unplug ||
855 !thread_is_running(thread),
857 if (!thread_is_running(thread))
858 GOTO(fini_oit, rc = 0);
860 if (!list_empty(&lfsck->li_list_scan) ||
861 list_empty(&lfsck->li_list_double_scan))
862 rc = lfsck_master_oit_engine(env, lfsck);
866 CDEBUG(D_LFSCK, "LFSCK exit: oit_flags = %#x, dir_flags = %#x, "
867 "oit_cookie = "LPU64", dir_cookie = "LPX64", parent = "DFID
868 ", pid = %d, rc = %d\n", lfsck->li_args_oit, lfsck->li_args_dir,
869 lfsck->li_pos_current.lp_oit_cookie,
870 lfsck->li_pos_current.lp_dir_cookie,
871 PFID(&lfsck->li_pos_current.lp_dir_parent),
874 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CRASH))
875 rc = lfsck_post(env, lfsck, rc);
877 if (lfsck->li_di_dir != NULL)
878 lfsck_close_dir(env, lfsck);
881 lfsck_di_oit_put(env, lfsck);
882 oit_iops->fini(env, oit_di);
884 if (!list_empty(&lfsck->li_list_double_scan))
885 rc = lfsck_double_scan(env, lfsck);
889 lfsck_quit(env, lfsck);
892 /* XXX: Purge the pinned objects in the future. */
895 spin_lock(&lfsck->li_lock);
896 thread_set_flags(thread, SVC_STOPPED);
897 spin_unlock(&lfsck->li_lock);
898 wake_up_all(&thread->t_ctl_waitq);
899 lfsck_thread_args_fini(lta);