X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Flfsck%2Flfsck_lib.c;h=7e6755536214d5cfb9470adcbbc38dd64f4c6b77;hp=9f2a3210b747655dfdd7892d6ed36bd965d9b434;hb=3cce65712d94cffe8f1626545845b95b88aef672;hpb=91ef75669c7e41091378a8401ca0c093b7f17174

diff --git a/lustre/lfsck/lfsck_lib.c b/lustre/lfsck/lfsck_lib.c
index 9f2a321..7e67555 100644
--- a/lustre/lfsck/lfsck_lib.c
+++ b/lustre/lfsck/lfsck_lib.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2013, 2015, Intel Corporation.
+ * Copyright (c) 2013, 2017, Intel Corporation.
  */
 /*
  * lustre/lfsck/lfsck_lib.c
@@ -40,7 +40,6 @@
 #include <lustre_lib.h>
 #include <lustre_net.h>
 #include <lustre_lfsck.h>
-#include <lustre/lustre_lfsck_user.h>
 
 #include "lfsck_internal.h"
 
@@ -86,6 +85,8 @@ const char *lfsck_param_names[] = {
 	"orphan",
 	"create_ostobj",
 	"create_mdtobj",
+	NULL,
+	"delay_create_ostobj",
 	NULL
 };
 
@@ -378,8 +379,14 @@ static int __lfsck_ibits_lock(const struct lu_env *env,
 		einfo->ei_res_id = resid;
 
 		rc = dt_object_lock(env, obj, lh, einfo, policy);
+		/* for regular checks LFSCK doesn't use LDLM locking,
+		 * so the state isn't coherent. here we just took LDLM
+		 * lock for coherency and it's time to invalidate
+		 * previous state */
+		if (rc == ELDLM_OK)
+			dt_invalidate(env, obj);
 	} else {
-		rc = ldlm_cli_enqueue_local(lfsck->li_namespace, resid,
+		rc = ldlm_cli_enqueue_local(env, lfsck->li_namespace, resid,
 					    LDLM_IBITS, policy, mode,
 					    &flags, ldlm_blocking_ast,
 					    ldlm_completion_ast, NULL, NULL,
@@ -425,6 +432,36 @@ int lfsck_ibits_lock(const struct lu_env *env, struct lfsck_instance *lfsck,
 }
 
 /**
+ * Request the remote LOOKUP lock for the given object.
+ *
+ * If \a pobj is remote, the LOOKUP lock of \a obj is on the MDT where
+ * \a pobj is, acquire LOOKUP lock there.
+ *
+ * \param[in] env	pointer to the thread context
+ * \param[in] lfsck	pointer to the lfsck instance
+ * \param[in] pobj	pointer to parent dt_object
+ * \param[in] obj	pointer to the dt_object to be locked
+ * \param[out] lh	pointer to the lock handle
+ * \param[in] mode	the mode for the ldlm lock to be acquired
+ *
+ * \retval		0 for success
+ * \retval		negative error number on failure
+ */
+int lfsck_remote_lookup_lock(const struct lu_env *env,
+			     struct lfsck_instance *lfsck,
+			     struct dt_object *pobj, struct dt_object *obj,
+			     struct lustre_handle *lh, enum ldlm_mode mode)
+{
+	struct ldlm_res_id *resid = &lfsck_env_info(env)->lti_resid;
+
+	LASSERT(!lustre_handle_is_used(lh));
+
+	fid_build_reg_res_name(lfsck_dto2fid(obj), resid);
+	return __lfsck_ibits_lock(env, lfsck, pobj, resid, lh,
+				  MDS_INODELOCK_LOOKUP, mode);
+}
+
+/**
  * Release the the specified ibits lock.
  *
  * If the lock has been acquired before, release it
@@ -502,7 +539,8 @@ int lfsck_lock(const struct lu_env *env, struct lfsck_instance *lfsck,
 		return rc;
 
 	llh->llh_reg_mode = mode;
-	resid->name[LUSTRE_RES_ID_HSH_OFF] = full_name_hash(name, strlen(name));
+	resid->name[LUSTRE_RES_ID_HSH_OFF] = ll_full_name_hash(NULL, name,
+							       strlen(name));
 	LASSERT(resid->name[LUSTRE_RES_ID_HSH_OFF] != 0);
 	rc = __lfsck_ibits_lock(env, lfsck, obj, resid, &llh->llh_reg_lh,
 				bits, llh->llh_reg_mode);
@@ -531,6 +569,14 @@ int lfsck_find_mdt_idx_by_fid(const struct lu_env *env,
 	struct lu_seq_range	*range	= &lfsck_env_info(env)->lti_range;
 	int			 rc;
 
+	if (unlikely(fid_seq(fid) == FID_SEQ_LOCAL_FILE)) {
+		/* "ROOT" is always on the MDT0. */
+		if (lu_fid_eq(fid, &lfsck->li_global_root_fid))
+			return 0;
+
+		return lfsck_dev_idx(lfsck);
+	}
+
 	fld_range_set_mdt(range);
 	rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), range);
 	if (rc == 0)
@@ -634,13 +680,9 @@ static int lfsck_create_lpf_local(const struct lu_env *env,
 	int			 rc;
 	ENTRY;
 
-	rc = linkea_data_new(&ldata,
-			     &lfsck_env_info(env)->lti_linkea_buf2);
-	if (rc != 0)
-		RETURN(rc);
-
 	cname = lfsck_name_get_const(env, name, strlen(name));
-	rc = linkea_add_buf(&ldata, cname, lfsck_dto2fid(parent));
+	rc = linkea_links_new(&ldata, &lfsck_env_info(env)->lti_linkea_buf2,
+			      cname, lfsck_dto2fid(parent));
 	if (rc != 0)
 		RETURN(rc);
 
@@ -721,14 +763,14 @@ static int lfsck_create_lpf_local(const struct lu_env *env,
 	/* 3b. insert dot into child dir */
 	rec->rec_fid = cfid;
 	rc = dt_insert(env, child, (const struct dt_rec *)rec,
-		       (const struct dt_key *)dot, th, 1);
+		       (const struct dt_key *)dot, th);
 	if (rc != 0)
 		GOTO(unlock, rc);
 
 	/* 4b. insert dotdot into child dir */
 	rec->rec_fid = &LU_LPF_FID;
 	rc = dt_insert(env, child, (const struct dt_rec *)rec,
-		       (const struct dt_key *)dotdot, th, 1);
+		       (const struct dt_key *)dotdot, th);
 	if (rc != 0)
 		GOTO(unlock, rc);
 
@@ -742,7 +784,7 @@ static int lfsck_create_lpf_local(const struct lu_env *env,
 	/* 6b. insert name into parent dir */
 	rec->rec_fid = cfid;
 	rc = dt_insert(env, parent, (const struct dt_rec *)rec,
-		       (const struct dt_key *)name, th, 1);
+		       (const struct dt_key *)name, th);
 	if (rc != 0)
 		GOTO(stop, rc);
 
@@ -793,13 +835,9 @@ static int lfsck_create_lpf_remote(const struct lu_env *env,
 	int			 rc;
 	ENTRY;
 
-	rc = linkea_data_new(&ldata,
-			     &lfsck_env_info(env)->lti_linkea_buf2);
-	if (rc != 0)
-		RETURN(rc);
-
 	cname = lfsck_name_get_const(env, name, strlen(name));
-	rc = linkea_add_buf(&ldata, cname, lfsck_dto2fid(parent));
+	rc = linkea_links_new(&ldata, &lfsck_env_info(env)->lti_linkea_buf2,
+			      cname, lfsck_dto2fid(parent));
 	if (rc != 0)
 		RETURN(rc);
 
@@ -891,14 +929,14 @@ static int lfsck_create_lpf_remote(const struct lu_env *env,
 	rec->rec_type = S_IFDIR;
 	rec->rec_fid = cfid;
 	rc = dt_insert(env, child, (const struct dt_rec *)rec,
-		       (const struct dt_key *)dot, th, 1);
+		       (const struct dt_key *)dot, th);
 	if (rc != 0)
 		GOTO(unlock, rc);
 
 	/* 4b. insert dotdot into child dir */
 	rec->rec_fid = &LU_LPF_FID;
 	rc = dt_insert(env, child, (const struct dt_rec *)rec,
-		       (const struct dt_key *)dotdot, th, 1);
+		       (const struct dt_key *)dotdot, th);
 	if (rc != 0)
 		GOTO(unlock, rc);
 
@@ -946,7 +984,7 @@ static int lfsck_create_lpf_remote(const struct lu_env *env,
 
 	/* 5b. insert name into parent dir */
 	rc = dt_insert(env, parent, (const struct dt_rec *)rec,
-		       (const struct dt_key *)name, th, 1);
+		       (const struct dt_key *)name, th);
 	if (rc != 0)
 		GOTO(stop, rc);
 
@@ -1034,7 +1072,7 @@ static int lfsck_create_lpf(const struct lu_env *env,
 		*cfid = bk->lb_lpf_fid;
 	}
 
-	child = lfsck_object_find_bottom(env, lfsck, cfid);
+	child = lfsck_object_find_bottom_new(env, lfsck, cfid);
 	if (IS_ERR(child))
 		GOTO(unlock, rc = PTR_ERR(child));
 
@@ -1048,7 +1086,7 @@ static int lfsck_create_lpf(const struct lu_env *env,
 	}
 
 	memset(la, 0, sizeof(*la));
-	la->la_atime = la->la_mtime = la->la_ctime = cfs_time_current_sec();
+	la->la_atime = la->la_mtime = la->la_ctime = ktime_get_real_seconds();
 	la->la_mode = S_IFDIR | S_IRWXU;
 	la->la_valid = LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
 		       LA_UID | LA_GID;
@@ -1764,15 +1802,15 @@ void lfsck_bits_dump(struct seq_file *m, int bits, const char *names[],
 		seq_putc(m, '\n');
 }
 
-void lfsck_time_dump(struct seq_file *m, __u64 time, const char *name)
+void lfsck_time_dump(struct seq_file *m, time64_t time, const char *name)
 {
 	if (time == 0) {
 		seq_printf(m, "%s_time: N/A\n", name);
 		seq_printf(m, "time_since_%s: N/A\n", name);
 	} else {
-		seq_printf(m, "%s_time: "LPU64"\n", name, time);
-		seq_printf(m, "time_since_%s: "LPU64" seconds\n",
-			   name, cfs_time_current_sec() - time);
+		seq_printf(m, "%s_time: %lld\n", name, time);
+		seq_printf(m, "time_since_%s: %lld seconds\n",
+			   name, ktime_get_real_seconds() - time);
 	}
 }
 
@@ -1784,10 +1822,10 @@ void lfsck_pos_dump(struct seq_file *m, struct lfsck_position *pos,
 			seq_printf(m, "%s: N/A, N/A, N/A\n", prefix);
 			return;
 		}
-		seq_printf(m, "%s: "LPU64", N/A, N/A\n",
+		seq_printf(m, "%s: %llu, N/A, N/A\n",
 			   prefix, pos->lp_oit_cookie);
 	} else {
-		seq_printf(m, "%s: "LPU64", "DFID", "LPX64"\n",
+		seq_printf(m, "%s: %llu, "DFID", %#llx\n",
 			   prefix, pos->lp_oit_cookie,
 			   PFID(&pos->lp_dir_parent), pos->lp_dir_cookie);
 	}
@@ -1807,7 +1845,8 @@ void lfsck_pos_fill(const struct lu_env *env, struct lfsck_instance *lfsck,
 	if (!lfsck->li_current_oit_processed && !init)
 		pos->lp_oit_cookie--;
 
-	LASSERT(pos->lp_oit_cookie > 0);
+	if (unlikely(pos->lp_oit_cookie == 0))
+		pos->lp_oit_cookie = 1;
 
 	if (lfsck->li_di_dir != NULL) {
 		struct dt_object *dto = lfsck->li_obj_dir;
@@ -1832,14 +1871,12 @@ bool __lfsck_set_speed(struct lfsck_instance *lfsck, __u32 limit)
 	bool dirty = false;
 
 	if (limit != LFSCK_SPEED_NO_LIMIT) {
-		if (limit > msecs_to_jiffies(MSEC_PER_SEC)) {
-			lfsck->li_sleep_rate = limit /
-					       msecs_to_jiffies(MSEC_PER_SEC);
+		if (limit > cfs_time_seconds(1)) {
+			lfsck->li_sleep_rate = limit / cfs_time_seconds(1);
 			lfsck->li_sleep_jif = 1;
 		} else {
 			lfsck->li_sleep_rate = 1;
-			lfsck->li_sleep_jif = msecs_to_jiffies(MSEC_PER_SEC) /
-					      limit;
+			lfsck->li_sleep_jif = cfs_time_seconds(1) / limit;
 		}
 	} else {
 		lfsck->li_sleep_jif = 0;
@@ -2033,6 +2070,11 @@ int lfsck_async_interpret_common(const struct lu_env *env,
 
 	switch (lr->lr_event) {
 	case LE_START:
+		if (unlikely(rc == -EINPROGRESS)) {
+			ltd->ltd_retry_start = 1;
+			break;
+		}
+
 		if (rc != 0) {
 			CDEBUG(D_LFSCK, "%s: fail to notify %s %x for %s "
 			       "start: rc = %d\n",
@@ -2339,7 +2381,7 @@ static int lfsck_stop_notify(const struct lu_env *env,
 			       ltd->ltd_index, lad->lad_name, rc);
 			lfsck_tgt_put(ltd);
 		} else {
-			rc = ptlrpc_set_wait(set);
+			rc = ptlrpc_set_wait(env, set);
 		}
 
 		ptlrpc_set_destroy(set);
@@ -2412,6 +2454,7 @@ int lfsck_async_request(const struct lu_env *env, struct obd_export *exp,
 		lfsck_component_get(laia->laia_com);
 	req->rq_interpret_reply = interpreter;
 	req->rq_allow_intr = 1;
+	req->rq_no_delay = 1;
 	ptlrpc_set_add_req(set, req);
 
 	return 0;
@@ -2475,7 +2518,7 @@ again:
 		goto again;
 	}
 
-	rc = ptlrpc_set_wait(set);
+	rc = ptlrpc_set_wait(env, set);
 	ptlrpc_set_destroy(set);
 
 	RETURN(rc);
@@ -2495,10 +2538,7 @@ int lfsck_start_assistant(const struct lu_env *env, struct lfsck_component *com,
 
 	lad->lad_assistant_status = 0;
 	lad->lad_post_result = 0;
-	lad->lad_to_post = 0;
-	lad->lad_to_double_scan = 0;
-	lad->lad_in_double_scan = 0;
-	lad->lad_exit = 0;
+	lad->lad_flags = 0;
 	lad->lad_advance_lock = false;
 	thread_set_flags(athread, 0);
 
@@ -2517,9 +2557,13 @@ int lfsck_start_assistant(const struct lu_env *env, struct lfsck_component *com,
 
 		l_wait_event(mthread->t_ctl_waitq,
 			     thread_is_running(athread) ||
-			     thread_is_stopped(athread),
+			     thread_is_stopped(athread) ||
+			     !thread_is_starting(mthread),
 			     &lwi);
-		if (unlikely(!thread_is_running(athread)))
+		if (unlikely(!thread_is_starting(mthread)))
+			/* stopped by race */
+			rc = -ESRCH;
+		else if (unlikely(!thread_is_running(athread)))
 			rc = lad->lad_assistant_status;
 		else
 			rc = 0;
@@ -2558,8 +2602,8 @@ void lfsck_post_generic(const struct lu_env *env,
 
 	lad->lad_post_result = *result;
 	if (*result <= 0)
-		lad->lad_exit = 1;
-	lad->lad_to_post = 1;
+		set_bit(LAD_EXIT, &lad->lad_flags);
+	set_bit(LAD_TO_POST, &lad->lad_flags);
 
 	CDEBUG(D_LFSCK, "%s: waiting for assistant to do %s post, rc = %d\n",
 	       lfsck_lfsck2name(com->lc_lfsck), lad->lad_name, *result);
@@ -2586,9 +2630,9 @@ int lfsck_double_scan_generic(const struct lu_env *env,
 	struct l_wait_info		 lwi	 = { 0 };
 
 	if (status != LS_SCANNING_PHASE2)
-		lad->lad_exit = 1;
+		set_bit(LAD_EXIT, &lad->lad_flags);
 	else
-		lad->lad_to_double_scan = 1;
+		set_bit(LAD_TO_DOUBLE_SCAN, &lad->lad_flags);
 
 	CDEBUG(D_LFSCK, "%s: waiting for assistant to do %s double_scan, "
 	       "status %d\n",
@@ -2596,7 +2640,7 @@ int lfsck_double_scan_generic(const struct lu_env *env,
 
 	wake_up_all(&athread->t_ctl_waitq);
 	l_wait_event(mthread->t_ctl_waitq,
-		     lad->lad_in_double_scan ||
+		     test_bit(LAD_IN_DOUBLE_SCAN, &lad->lad_flags) ||
 		     thread_is_stopped(athread),
 		     &lwi);
 
@@ -2618,7 +2662,7 @@ void lfsck_quit_generic(const struct lu_env *env,
 	struct ptlrpc_thread		*athread = &lad->lad_thread;
 	struct l_wait_info		 lwi     = { 0 };
 
-	lad->lad_exit = 1;
+	set_bit(LAD_EXIT, &lad->lad_flags);
 	wake_up_all(&athread->t_ctl_waitq);
 	l_wait_event(mthread->t_ctl_waitq,
 		     thread_is_init(athread) ||
@@ -2626,9 +2670,107 @@ void lfsck_quit_generic(const struct lu_env *env,
 		     &lwi);
 }
 
-/* external interfaces */
+int lfsck_load_one_trace_file(const struct lu_env *env,
+			      struct lfsck_component *com,
+			      struct dt_object *parent,
+			      struct dt_object **child,
+			      const struct dt_index_features *ft,
+			      const char *name, bool reset)
+{
+	struct lfsck_instance *lfsck = com->lc_lfsck;
+	struct dt_object *obj;
+	int rc;
+	ENTRY;
+
+	if (*child != NULL) {
+		struct dt_it *it;
+		const struct dt_it_ops *iops;
+		struct lu_fid *fid = &lfsck_env_info(env)->lti_fid3;
+
+		if (!reset)
+			RETURN(0);
+
+		obj = *child;
+		rc = obj->do_ops->do_index_try(env, obj, ft);
+		if (rc)
+			/* unlink by force */
+			goto unlink;
+
+		iops = &obj->do_index_ops->dio_it;
+		it = iops->init(env, obj, 0);
+		if (IS_ERR(it))
+			/* unlink by force */
+			goto unlink;
+
+		fid_zero(fid);
+		rc = iops->get(env, it, (const struct dt_key *)fid);
+		if (rc >= 0) {
+			rc = iops->next(env, it);
+			iops->put(env, it);
+		}
+		iops->fini(env, it);
+		if (rc > 0)
+			/* "rc > 0" means the index file is empty. */
+			RETURN(0);
+
+unlink:
+		/* The old index is not empty, remove it firstly. */
+		rc = local_object_unlink(env, lfsck->li_bottom, parent, name);
+		CDEBUG_LIMIT(rc ? D_ERROR : D_LFSCK,
+			     "%s: unlink lfsck sub trace file %s: rc = %d\n",
+			     lfsck_lfsck2name(com->lc_lfsck), name, rc);
+		if (rc)
+			RETURN(rc);
+
+		if (*child) {
+			lfsck_object_put(env, *child);
+			*child = NULL;
+		}
+	} else if (reset) {
+		goto unlink;
+	}
+
+	obj = local_index_find_or_create(env, lfsck->li_los, parent, name,
+					 S_IFREG | S_IRUGO | S_IWUSR, ft);
+	if (IS_ERR(obj))
+		RETURN(PTR_ERR(obj));
+
+	rc = obj->do_ops->do_index_try(env, obj, ft);
+	if (rc) {
+		lfsck_object_put(env, obj);
+		CDEBUG(D_LFSCK, "%s: LFSCK fail to load "
+		       "sub trace file %s: rc = %d\n",
+		       lfsck_lfsck2name(com->lc_lfsck), name, rc);
+	} else {
+		*child = obj;
+	}
+
+	RETURN(rc);
+}
 
-int lfsck_get_speed(struct seq_file *m, struct dt_device *key)
+int lfsck_load_sub_trace_files(const struct lu_env *env,
+			       struct lfsck_component *com,
+			       const struct dt_index_features *ft,
+			       const char *prefix, bool reset)
+{
+	char *name = lfsck_env_info(env)->lti_key;
+	struct lfsck_sub_trace_obj *lsto;
+	int rc;
+	int i;
+
+	for (i = 0, rc = 0, lsto = &com->lc_sub_trace_objs[0];
+	     i < LFSCK_STF_COUNT && rc == 0; i++, lsto++) {
+		snprintf(name, NAME_MAX, "%s_%02d", prefix, i);
+		rc = lfsck_load_one_trace_file(env, com,
+				com->lc_lfsck->li_lfsck_dir,
+				&lsto->lsto_obj, ft, name, reset);
+	}
+
+	return rc;
+}
+
+/* external interfaces */
+int lfsck_get_speed(char *buf, struct dt_device *key)
 {
 	struct lu_env		env;
 	struct lfsck_instance  *lfsck;
@@ -2640,8 +2782,9 @@ int lfsck_get_speed(struct seq_file *m, struct dt_device *key)
 		RETURN(rc);
 
 	lfsck = lfsck_instance_find(key, true, false);
-	if (likely(lfsck != NULL)) {
-		seq_printf(m, "%u\n", lfsck->li_bookmark_ram.lb_speed_limit);
+	if (lfsck && buf) {
+		rc = sprintf(buf, "%u\n",
+			     lfsck->li_bookmark_ram.lb_speed_limit);
 		lfsck_instance_put(&env, lfsck);
 	} else {
 		rc = -ENXIO;
@@ -2681,7 +2824,7 @@ int lfsck_set_speed(struct dt_device *key, __u32 val)
 }
 EXPORT_SYMBOL(lfsck_set_speed);
 
-int lfsck_get_windows(struct seq_file *m, struct dt_device *key)
+int lfsck_get_windows(char *buf, struct dt_device *key)
 {
 	struct lu_env		env;
 	struct lfsck_instance  *lfsck;
@@ -2694,7 +2837,8 @@ int lfsck_get_windows(struct seq_file *m, struct dt_device *key)
 
 	lfsck = lfsck_instance_find(key, true, false);
 	if (likely(lfsck != NULL)) {
-		seq_printf(m, "%u\n", lfsck->li_bookmark_ram.lb_async_windows);
+		rc = sprintf(buf, "%u\n",
+			     lfsck->li_bookmark_ram.lb_async_windows);
 		lfsck_instance_put(&env, lfsck);
 	} else {
 		rc = -ENXIO;
@@ -2706,7 +2850,7 @@ int lfsck_get_windows(struct seq_file *m, struct dt_device *key)
 }
 EXPORT_SYMBOL(lfsck_get_windows);
 
-int lfsck_set_windows(struct dt_device *key, int val)
+int lfsck_set_windows(struct dt_device *key, unsigned int val)
 {
 	struct lu_env		env;
 	struct lfsck_instance  *lfsck;
@@ -2829,7 +2973,7 @@ static int lfsck_stop_all(const struct lu_env *env,
 	}
 	up_read(&ltds->ltd_rw_sem);
 
-	rc = ptlrpc_set_wait(set);
+	rc = ptlrpc_set_wait(env, set);
 	ptlrpc_set_destroy(set);
 
 	if (rc == 0)
@@ -2858,14 +3002,11 @@ static int lfsck_start_all(const struct lu_env *env,
 	struct lfsck_bookmark		  *bk	  = &lfsck->li_bookmark_ram;
 	__u32				   idx;
 	int				   rc	  = 0;
+	bool retry = false;
 	ENTRY;
 
 	LASSERT(start->ls_flags & LPF_BROADCAST);
 
-	set = ptlrpc_prep_set();
-	if (unlikely(set == NULL))
-		RETURN(-ENOMEM);
-
 	memset(lr, 0, sizeof(*lr));
 	lr->lr_event = LE_START;
 	lr->lr_index = lfsck_dev_idx(lfsck);
@@ -2883,12 +3024,23 @@ static int lfsck_start_all(const struct lu_env *env,
 	laia->laia_lr = lr;
 	laia->laia_shared = 1;
 
+again:
+	set = ptlrpc_prep_set();
+	if (unlikely(!set))
+		RETURN(-ENOMEM);
+
 	down_read(&ltds->ltd_rw_sem);
 	cfs_foreach_bit(ltds->ltd_tgts_bitmap, idx) {
 		ltd = lfsck_tgt_get(ltds, idx);
 		LASSERT(ltd != NULL);
 
+		if (retry && !ltd->ltd_retry_start) {
+			lfsck_tgt_put(ltd);
+			continue;
+		}
+
 		laia->laia_ltd = ltd;
+		ltd->ltd_retry_start = 0;
 		ltd->ltd_layout_done = 0;
 		ltd->ltd_namespace_done = 0;
 		ltd->ltd_synced_failures = 0;
@@ -2912,12 +3064,24 @@ static int lfsck_start_all(const struct lu_env *env,
 		RETURN(rc);
 	}
 
-	rc = ptlrpc_set_wait(set);
+	rc = ptlrpc_set_wait(env, set);
 	ptlrpc_set_destroy(set);
 
 	if (rc == 0)
 		rc = laia->laia_result;
 
+	if (unlikely(rc == -EINPROGRESS)) {
+		retry = true;
+		set_current_state(TASK_INTERRUPTIBLE);
+		schedule_timeout(cfs_time_seconds(1));
+		set_current_state(TASK_RUNNING);
+		if (!signal_pending(current) &&
+		    thread_is_running(&lfsck->li_thread))
+			goto again;
+
+		rc = -EINTR;
+	}
+
 	if (rc != 0) {
 		struct lfsck_stop *stop = &info->lti_stop;
 
@@ -2954,24 +3118,43 @@ int lfsck_start(const struct lu_env *env, struct dt_device *key,
 	__u16				 type   = 1;
 	ENTRY;
 
+	if (key->dd_rdonly)
+		RETURN(-EROFS);
+
 	lfsck = lfsck_instance_find(key, true, false);
 	if (unlikely(lfsck == NULL))
 		RETURN(-ENXIO);
 
+	if (unlikely(lfsck->li_stopping))
+		GOTO(put, rc = -ENXIO);
+
 	/* System is not ready, try again later. */
-	if (unlikely(lfsck->li_namespace == NULL))
-		GOTO(put, rc = -EAGAIN);
+	if (unlikely(lfsck->li_namespace == NULL ||
+		     lfsck_dev_site(lfsck)->ss_server_fld == NULL))
+		GOTO(put, rc = -EINPROGRESS);
 
 	/* start == NULL means auto trigger paused LFSCK. */
-	if ((start == NULL) &&
-	    (list_empty(&lfsck->li_list_scan) ||
-	     OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_AUTO)))
-		GOTO(put, rc = 0);
+	if (!start) {
+		if (list_empty(&lfsck->li_list_scan) ||
+		    OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_AUTO))
+			GOTO(put, rc = 0);
+	} else if (start->ls_flags & LPF_BROADCAST && !lfsck->li_master) {
+		CERROR("%s: only allow to specify '-A | -o' via MDS\n",
+		       lfsck_lfsck2name(lfsck));
+
+		GOTO(put, rc = -EPERM);
+	}
 
 	bk = &lfsck->li_bookmark_ram;
 	thread = &lfsck->li_thread;
 	mutex_lock(&lfsck->li_mutex);
 	spin_lock(&lfsck->li_lock);
+	if (unlikely(thread_is_stopping(thread))) {
+		/* Someone is stopping the LFSCK. */
+		spin_unlock(&lfsck->li_lock);
+		GOTO(out, rc = -EBUSY);
+	}
+
 	if (!thread_is_init(thread) && !thread_is_stopped(thread)) {
 		rc = -EALREADY;
 		if (unlikely(start == NULL)) {
@@ -3018,13 +3201,6 @@ int lfsck_start(const struct lu_env *env, struct dt_device *key,
 	if (start == NULL)
 		goto trigger;
 
-	if (start->ls_flags & LPF_BROADCAST && !lfsck->li_master) {
-		CERROR("%s: only allow to specify '-A | -o' via MDS\n",
-		       lfsck_lfsck2name(lfsck));
-
-		GOTO(out, rc = -EPERM);
-	}
-
 	start->ls_version = bk->lb_version;
 
 	if (start->ls_active != 0) {
@@ -3147,12 +3323,14 @@ trigger:
 		flags |= DOIF_OUTUSED;
 
 	lfsck->li_args_oit = (flags << DT_OTABLE_IT_FLAGS_SHIFT) | valid;
-	thread_set_flags(thread, 0);
 	lta = lfsck_thread_args_init(lfsck, NULL, lsp);
 	if (IS_ERR(lta))
 		GOTO(out, rc = PTR_ERR(lta));
 
 	__lfsck_set_speed(lfsck, bk->lb_speed_limit);
+	spin_lock(&lfsck->li_lock);
+	thread_set_flags(thread, SVC_STARTING);
+	spin_unlock(&lfsck->li_lock);
 	task = kthread_run(lfsck_master_engine, lta, "lfsck");
 	if (IS_ERR(task)) {
 		rc = PTR_ERR(task);
@@ -3225,27 +3403,26 @@ int lfsck_stop(const struct lu_env *env, struct dt_device *key,
 		RETURN(-ENXIO);
 
 	thread = &lfsck->li_thread;
-	/* release lfsck::li_mutex to avoid deadlock. */
-	if (stop != NULL && stop->ls_flags & LPF_BROADCAST) {
-		if (!lfsck->li_master) {
-			CERROR("%s: only allow to specify '-A' via MDS\n",
-			       lfsck_lfsck2name(lfsck));
-
-			GOTO(out, rc = -EPERM);
-		}
-
-		rc1 = lfsck_stop_all(env, lfsck, stop);
+	if (stop && stop->ls_flags & LPF_BROADCAST && !lfsck->li_master) {
+		CERROR("%s: only allow to specify '-A' via MDS\n",
+		       lfsck_lfsck2name(lfsck));
+		GOTO(put, rc = -EPERM);
 	}
 
-	mutex_lock(&lfsck->li_mutex);
 	spin_lock(&lfsck->li_lock);
-	/* no error if LFSCK is already stopped, or was never started */
-	if (thread_is_init(thread) || thread_is_stopped(thread)) {
-		spin_unlock(&lfsck->li_lock);
-		GOTO(out, rc = 0);
-	}
+	/* The target is umounted */
+	if (stop && stop->ls_status == LS_PAUSED)
+		lfsck->li_stopping = 1;
+
+	if (thread_is_init(thread) || thread_is_stopped(thread))
+		/* no error if LFSCK stopped already, or not started */
+		GOTO(unlock, rc = 0);
 
-	if (stop != NULL) {
+	if (thread_is_stopping(thread))
+		/* Someone is stopping LFSCK. */
+		GOTO(unlock, rc = -EINPROGRESS);
+
+	if (stop) {
 		lfsck->li_status = stop->ls_status;
 		lfsck->li_flags = stop->ls_flags;
 	} else {
@@ -3255,6 +3432,9 @@ int lfsck_stop(const struct lu_env *env, struct dt_device *key,
 
 	thread_set_flags(thread, SVC_STOPPING);
 
+	LASSERT(lfsck->li_task != NULL);
+	force_sig(SIGINT, lfsck->li_task);
+
 	if (lfsck->li_master) {
 		struct lfsck_component *com;
 		struct lfsck_assistant_data *lad;
@@ -3276,25 +3456,54 @@ int lfsck_stop(const struct lu_env *env, struct dt_device *key,
 		}
 	}
 
+	wake_up_all(&thread->t_ctl_waitq);
 	spin_unlock(&lfsck->li_lock);
+	if (stop && stop->ls_flags & LPF_BROADCAST)
+		rc1 = lfsck_stop_all(env, lfsck, stop);
 
-	wake_up_all(&thread->t_ctl_waitq);
+	/* It was me set the status as 'stopping' just now, if it is not
+	 * 'stopping' now, then either stopped, or re-started by race. */
 	l_wait_event(thread->t_ctl_waitq,
-		     thread_is_stopped(thread),
+		     !thread_is_stopping(thread),
 		     &lwi);
 
-	GOTO(out, rc = 0);
+	GOTO(put, rc = 0);
 
-out:
-	mutex_unlock(&lfsck->li_mutex);
+unlock:
+	spin_unlock(&lfsck->li_lock);
+put:
 	lfsck_instance_put(env, lfsck);
 
 	return rc != 0 ? rc : rc1;
 }
 EXPORT_SYMBOL(lfsck_stop);
 
+int lfsck_in_notify_local(const struct lu_env *env, struct dt_device *key,
+			  struct lfsck_req_local *lrl, struct thandle *th)
+{
+	struct lfsck_instance *lfsck;
+	struct lfsck_component *com;
+	int rc = -EOPNOTSUPP;
+	ENTRY;
+
+	lfsck = lfsck_instance_find(key, true, false);
+	if (unlikely(!lfsck))
+		RETURN(-ENXIO);
+
+	com = lfsck_component_find(lfsck, lrl->lrl_active);
+	if (likely(com && com->lc_ops->lfsck_in_notify_local)) {
+		rc = com->lc_ops->lfsck_in_notify_local(env, com, lrl, th);
+		lfsck_component_put(env, com);
+	}
+
+	lfsck_instance_put(env, lfsck);
+
+	RETURN(rc);
+}
+EXPORT_SYMBOL(lfsck_in_notify_local);
+
 int lfsck_in_notify(const struct lu_env *env, struct dt_device *key,
-		    struct lfsck_request *lr, struct thandle *th)
+		    struct lfsck_request *lr)
 {
 	int rc = -EOPNOTSUPP;
 	ENTRY;
@@ -3329,11 +3538,8 @@ int lfsck_in_notify(const struct lu_env *env, struct dt_device *key,
 	}
 	case LE_PHASE1_DONE:
 	case LE_PHASE2_DONE:
-	case LE_FID_ACCESSED:
 	case LE_PEER_EXIT:
 	case LE_CONDITIONAL_DESTROY:
-	case LE_SKIP_NLINK_DECLARE:
-	case LE_SKIP_NLINK:
 	case LE_SET_LMV_MASTER:
 	case LE_SET_LMV_SLAVE:
 	case LE_PAIRS_VERIFY: {
@@ -3345,8 +3551,8 @@ int lfsck_in_notify(const struct lu_env *env, struct dt_device *key,
 			RETURN(-ENXIO);
 
 		com = lfsck_component_find(lfsck, lr->lr_active);
-		if (likely(com != NULL)) {
-			rc = com->lc_ops->lfsck_in_notify(env, com, lr, th);
+		if (likely(com)) {
+			rc = com->lc_ops->lfsck_in_notify(env, com, lr);
 			lfsck_component_put(env, com);
 		}
 
@@ -3796,7 +4002,8 @@ static int __init lfsck_init(void)
 	INIT_LIST_HEAD(&lfsck_mdt_orphan_list);
 	lfsck_key_init_generic(&lfsck_thread_key, NULL);
 	rc = lu_context_key_register(&lfsck_thread_key);
-	if (rc == 0) {
+	if (!rc) {
+		tgt_register_lfsck_in_notify_local(lfsck_in_notify_local);
 		tgt_register_lfsck_in_notify(lfsck_in_notify);
 		tgt_register_lfsck_query(lfsck_query);
 	}