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_master_dir_engine(const struct lu_env *env,
155 struct lfsck_instance *lfsck)
157 struct lfsck_thread_info *info = lfsck_env_info(env);
158 const struct dt_it_ops *iops =
159 &lfsck->li_obj_dir->do_index_ops->dio_it;
160 struct dt_it *di = lfsck->li_di_dir;
161 struct lu_dirent *ent = &info->lti_ent;
162 struct lu_fid *fid = &info->lti_fid;
163 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
164 struct ptlrpc_thread *thread = &lfsck->li_thread;
169 struct dt_object *child;
171 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY2) &&
173 struct l_wait_info lwi;
175 lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val),
177 l_wait_event(thread->t_ctl_waitq,
178 !thread_is_running(thread),
182 lfsck->li_new_scanned++;
183 rc = iops->rec(env, di, (struct dt_rec *)ent,
185 lfsck_unpack_ent(ent, &lfsck->li_cookie_dir);
187 lfsck_fail(env, lfsck, true);
188 if (bk->lb_param & LPF_FAILOUT)
194 if (ent->lde_attrs & LUDA_IGNORE)
198 child = lfsck_object_find(env, lfsck, fid);
201 } else if (IS_ERR(child)) {
202 lfsck_fail(env, lfsck, true);
203 if (bk->lb_param & LPF_FAILOUT)
204 RETURN(PTR_ERR(child));
209 /* XXX: Currently, skip remote object, the consistency for
210 * remote object will be processed in LFSCK phase III. */
211 if (dt_object_exists(child) && !dt_object_remote(child))
212 rc = lfsck_exec_dir(env, lfsck, child, ent);
213 lfsck_object_put(env, child);
214 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
218 rc = lfsck_checkpoint(env, lfsck);
219 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
223 lfsck_control_speed(lfsck);
224 if (unlikely(!thread_is_running(thread)))
227 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_FATAL2)) {
228 spin_lock(&lfsck->li_lock);
229 thread_set_flags(thread, SVC_STOPPING);
230 spin_unlock(&lfsck->li_lock);
234 rc = iops->next(env, di);
237 if (rc > 0 && !lfsck->li_oit_over)
238 lfsck_close_dir(env, lfsck);
243 static int lfsck_master_oit_engine(const struct lu_env *env,
244 struct lfsck_instance *lfsck)
246 struct lfsck_thread_info *info = lfsck_env_info(env);
247 const struct dt_it_ops *iops =
248 &lfsck->li_obj_oit->do_index_ops->dio_it;
249 struct dt_it *di = lfsck->li_di_oit;
250 struct lu_fid *fid = &info->lti_fid;
251 struct lfsck_bookmark *bk = &lfsck->li_bookmark_ram;
252 struct ptlrpc_thread *thread = &lfsck->li_thread;
254 lfsck_dev_idx(lfsck->li_bottom);
259 struct dt_object *target;
260 bool update_lma = false;
262 if (lfsck->li_di_dir != NULL) {
263 rc = lfsck_master_dir_engine(env, lfsck);
268 if (unlikely(lfsck->li_oit_over))
271 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY1) &&
273 struct l_wait_info lwi;
275 lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val),
277 l_wait_event(thread->t_ctl_waitq,
278 !thread_is_running(thread),
282 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CRASH))
285 lfsck->li_current_oit_processed = 1;
286 lfsck->li_new_scanned++;
287 rc = iops->rec(env, di, (struct dt_rec *)fid, 0);
289 lfsck_fail(env, lfsck, true);
290 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
296 if (fid_is_idif(fid)) {
297 __u32 idx1 = fid_idif_ost_idx(fid);
299 LASSERT(!lfsck->li_master);
301 /* It is an old format device, update the LMA. */
303 struct ost_id *oi = &info->lti_oi;
305 fid_to_ostid(fid, oi);
306 ostid_to_fid(fid, oi, idx);
309 } else if (!fid_is_norm(fid) && !fid_is_igif(fid) &&
310 !fid_is_last_id(fid) && !fid_is_root(fid) &&
311 !fid_seq_is_dot(fid_seq(fid))) {
312 /* If the FID/object is only used locally and invisible
313 * to external nodes, then LFSCK will not handle it. */
317 target = lfsck_object_find(env, lfsck, fid);
318 if (target == NULL) {
320 } else if (IS_ERR(target)) {
321 lfsck_fail(env, lfsck, true);
322 if (bk->lb_param & LPF_FAILOUT)
323 RETURN(PTR_ERR(target));
328 /* XXX: Currently, skip remote object, the consistency for
329 * remote object will be processed in LFSCK phase III. */
330 if (dt_object_exists(target) && !dt_object_remote(target)) {
332 rc = lfsck_update_lma(env, lfsck, target);
334 rc = lfsck_exec_oit(env, lfsck, target);
336 lfsck_object_put(env, target);
337 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
341 rc = lfsck_checkpoint(env, lfsck);
342 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
346 lfsck_control_speed(lfsck);
348 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_FATAL1)) {
349 spin_lock(&lfsck->li_lock);
350 thread_set_flags(thread, SVC_STOPPING);
351 spin_unlock(&lfsck->li_lock);
355 rc = iops->next(env, di);
356 if (unlikely(rc > 0))
357 lfsck->li_oit_over = 1;
358 else if (likely(rc == 0))
359 lfsck->li_current_oit_processed = 0;
361 if (unlikely(!thread_is_running(thread)))
363 } while (rc == 0 || lfsck->li_di_dir != NULL);
368 int lfsck_master_engine(void *args)
370 struct lfsck_thread_args *lta = args;
371 struct lu_env *env = <a->lta_env;
372 struct lfsck_instance *lfsck = lta->lta_lfsck;
373 struct ptlrpc_thread *thread = &lfsck->li_thread;
374 struct dt_object *oit_obj = lfsck->li_obj_oit;
375 const struct dt_it_ops *oit_iops = &oit_obj->do_index_ops->dio_it;
376 struct dt_it *oit_di;
377 struct l_wait_info lwi = { 0 };
381 oit_di = oit_iops->init(env, oit_obj, lfsck->li_args_oit, BYPASS_CAPA);
382 if (IS_ERR(oit_di)) {
383 rc = PTR_ERR(oit_di);
384 CERROR("%s: LFSCK, fail to init iteration: rc = %d\n",
385 lfsck_lfsck2name(lfsck), rc);
390 spin_lock(&lfsck->li_lock);
391 lfsck->li_di_oit = oit_di;
392 spin_unlock(&lfsck->li_lock);
393 rc = lfsck_prep(env, lfsck, lta->lta_lsp);
397 CDEBUG(D_LFSCK, "LFSCK entry: oit_flags = %#x, dir_flags = %#x, "
398 "oit_cookie = "LPU64", dir_cookie = "LPU64", parent = "DFID
399 ", pid = %d\n", lfsck->li_args_oit, lfsck->li_args_dir,
400 lfsck->li_pos_current.lp_oit_cookie,
401 lfsck->li_pos_current.lp_dir_cookie,
402 PFID(&lfsck->li_pos_current.lp_dir_parent),
405 spin_lock(&lfsck->li_lock);
406 thread_set_flags(thread, SVC_RUNNING);
407 spin_unlock(&lfsck->li_lock);
408 wake_up_all(&thread->t_ctl_waitq);
410 l_wait_event(thread->t_ctl_waitq,
411 lfsck->li_start_unplug ||
412 !thread_is_running(thread),
414 if (!thread_is_running(thread))
415 GOTO(fini_oit, rc = 0);
417 if (!cfs_list_empty(&lfsck->li_list_scan) ||
418 cfs_list_empty(&lfsck->li_list_double_scan))
419 rc = lfsck_master_oit_engine(env, lfsck);
423 CDEBUG(D_LFSCK, "LFSCK exit: oit_flags = %#x, dir_flags = %#x, "
424 "oit_cookie = "LPU64", dir_cookie = "LPU64", parent = "DFID
425 ", pid = %d, rc = %d\n", lfsck->li_args_oit, lfsck->li_args_dir,
426 lfsck->li_pos_current.lp_oit_cookie,
427 lfsck->li_pos_current.lp_dir_cookie,
428 PFID(&lfsck->li_pos_current.lp_dir_parent),
431 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CRASH))
432 rc = lfsck_post(env, lfsck, rc);
434 if (lfsck->li_di_dir != NULL)
435 lfsck_close_dir(env, lfsck);
438 lfsck_di_oit_put(env, lfsck);
439 oit_iops->fini(env, oit_di);
441 if (!cfs_list_empty(&lfsck->li_list_double_scan))
442 rc = lfsck_double_scan(env, lfsck);
446 lfsck_quit(env, lfsck);
449 /* XXX: Purge the pinned objects in the future. */
452 spin_lock(&lfsck->li_lock);
453 thread_set_flags(thread, SVC_STOPPED);
454 spin_unlock(&lfsck->li_lock);
455 wake_up_all(&thread->t_ctl_waitq);
456 lfsck_thread_args_fini(lta);