X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Ftarget%2Fupdate_trans.c;h=4d9a4c041cff8d57b28ee45a94a106fbb6c8ce84;hp=9fccd3babacab71481583b41a169c82f3571f368;hb=0098396983e1075668414aa5298a4990e61ffbda;hpb=d10200a80770f0029d1d665af954187b9ad883df

diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c
index 9fccd3b..4d9a4c0 100644
--- a/lustre/target/update_trans.c
+++ b/lustre/target/update_trans.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2015, 2016, Intel Corporation.
+ * Copyright (c) 2015, 2017, Intel Corporation.
  */
 /*
  * lustre/target/update_trans.c
@@ -82,13 +82,15 @@ static void top_multiple_thandle_dump(struct top_multiple_thandle *tmt,
 	list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) {
 		struct sub_thandle_cookie *stc;
 
-		CDEBUG(mask, "st %p obd %s committed %d stopped %d sub_th %p\n",
+		CDEBUG(mask, "st %p obd %s committed %d started %d stopped %d "
+		       "result %d sub_th %p\n",
 		       st, st->st_dt->dd_lu_dev.ld_obd->obd_name,
-		       st->st_committed, st->st_stopped, st->st_sub_th);
+		       st->st_committed, st->st_started, st->st_stopped,
+		       st->st_result, st->st_sub_th);
 
 		list_for_each_entry(stc, &st->st_cookie_list, stc_list) {
-			CDEBUG(mask, " cookie "DOSTID": %u\n",
-			       POSTID(&stc->stc_cookie.lgc_lgl.lgl_oi),
+			CDEBUG(mask, " cookie "DFID".%u\n",
+			       PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
 			       stc->stc_cookie.lgc_index);
 		}
 	}
@@ -201,7 +203,17 @@ static int sub_updates_write(const struct lu_env *env,
 		 "lrh_len %u record_size %zu\n", record->lur_hdr.lrh_len,
 		 llog_update_record_size(record));
 
-	if (likely(record->lur_hdr.lrh_len <= ctxt->loc_chunk_size)) {
+	/*
+	 * If its size > llog chunk_size, then write current chunk to the update
+	 * llog, NB the padding should >= LLOG_MIN_REC_SIZE.
+	 *
+	 * So check padding length is either >= LLOG_MIN_REC_SIZE or is 0
+	 * (record length just matches the chunk size).
+	 */
+
+	reclen = record->lur_hdr.lrh_len;
+	if (reclen + LLOG_MIN_REC_SIZE <= ctxt->loc_chunk_size ||
+	    reclen == ctxt->loc_chunk_size) {
 		OBD_ALLOC_PTR(stc);
 		if (stc == NULL)
 			GOTO(llog_put, rc = -ENOMEM);
@@ -210,9 +222,9 @@ static int sub_updates_write(const struct lu_env *env,
 		rc = llog_add(env, ctxt->loc_handle, &record->lur_hdr,
 			      &stc->stc_cookie, sub_th->st_sub_th);
 
-		CDEBUG(D_INFO, "%s: Add update log "DOSTID":%u: rc = %d\n",
+		CDEBUG(D_INFO, "%s: Add update log "DFID".%u: rc = %d\n",
 		       dt->dd_lu_dev.ld_obd->obd_name,
-		       POSTID(&stc->stc_cookie.lgc_lgl.lgl_oi),
+		       PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
 		       stc->stc_cookie.lgc_index, rc);
 
 		if (rc > 0) {
@@ -245,13 +257,6 @@ static int sub_updates_write(const struct lu_env *env,
 		else
 			eof = true;
 
-		/*
-		 * If its size > llog chunk_size, then write current chunk to
-		 * the update llog, NB the padding should >= LLOG_MIN_REC_SIZE.
-		 *
-		 * So check padding length is either >= LLOG_MIN_REC_SIZE or is
-		 * 0 (record length just matches the chunk size).
-		 */
 		reclen = __llog_update_record_size(
 				__update_records_size(next - start));
 		if ((reclen + LLOG_MIN_REC_SIZE <= ctxt->loc_chunk_size ||
@@ -290,9 +295,9 @@ static int sub_updates_write(const struct lu_env *env,
 		rc = llog_add(env, ctxt->loc_handle, &lur->lur_hdr,
 			      &stc->stc_cookie, sub_th->st_sub_th);
 
-		CDEBUG(D_INFO, "%s: Add update log "DOSTID":%u rc = %d\n",
+		CDEBUG(D_INFO, "%s: Add update log "DFID".%u: rc = %d\n",
 			dt->dd_lu_dev.ld_obd->obd_name,
-			POSTID(&stc->stc_cookie.lgc_lgl.lgl_oi),
+			PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
 			stc->stc_cookie.lgc_index, rc);
 
 		if (rc > 0) {
@@ -523,6 +528,7 @@ static void sub_trans_stop_cb(struct lu_env *env,
 	struct top_multiple_thandle	*tmt = cb->dcb_data;
 	ENTRY;
 
+	spin_lock(&tmt->tmt_sub_lock);
 	list_for_each_entry(st, &tmt->tmt_sub_thandle_list, st_sub_list) {
 		if (st->st_stopped)
 			continue;
@@ -533,6 +539,7 @@ static void sub_trans_stop_cb(struct lu_env *env,
 			break;
 		}
 	}
+	spin_unlock(&tmt->tmt_sub_lock);
 
 	wake_up(&tmt->tmt_stop_waitq);
 	RETURN_EXIT;
@@ -612,8 +619,6 @@ top_trans_create(const struct lu_env *env, struct dt_device *master_dev)
 		child_th->th_top = &top_th->tt_super;
 		child_th->th_wait_submit = 1;
 		top_th->tt_master_sub_thandle = child_th;
-
-		top_th->tt_super.th_tags |= child_th->th_tags;
 	}
 	return &top_th->tt_super;
 }
@@ -792,7 +797,6 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
 			top_th->tt_master_sub_thandle->th_sync = th->th_sync;
 		if (th->th_local)
 			top_th->tt_master_sub_thandle->th_local = th->th_local;
-		top_th->tt_master_sub_thandle->th_tags = th->th_tags;
 		rc = dt_trans_start(env, top_th->tt_master_sub_thandle->th_dev,
 				    top_th->tt_master_sub_thandle);
 		RETURN(rc);
@@ -810,7 +814,6 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
 			st->st_sub_th->th_sync = th->th_sync;
 		if (th->th_local)
 			st->st_sub_th->th_local = th->th_local;
-		st->st_sub_th->th_tags = th->th_tags;
 		rc = dt_trans_start(env, st->st_sub_th->th_dev,
 				    st->st_sub_th);
 		if (rc != 0)
@@ -907,10 +910,8 @@ static bool top_trans_is_stopped(struct top_thandle *top_th)
  */
 static int top_trans_wait_result(struct top_thandle *top_th)
 {
-	struct l_wait_info	lwi = {0};
-
-	l_wait_event(top_th->tt_multiple_thandle->tmt_stop_waitq,
-		     top_trans_is_stopped(top_th), &lwi);
+	wait_event_idle(top_th->tt_multiple_thandle->tmt_stop_waitq,
+			top_trans_is_stopped(top_th));
 
 	RETURN(top_th->tt_super.th_result);
 }
@@ -948,7 +949,6 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
 			top_th->tt_master_sub_thandle->th_sync = th->th_sync;
 		if (th->th_local)
 			top_th->tt_master_sub_thandle->th_local = th->th_local;
-		top_th->tt_master_sub_thandle->th_tags = th->th_tags;
 		rc = dt_trans_stop(env, master_dev,
 				   top_th->tt_master_sub_thandle);
 		OBD_FREE_PTR(top_th);
@@ -1008,7 +1008,6 @@ stop_master_trans:
 			master_st->st_sub_th->th_local = th->th_local;
 		if (th->th_sync)
 			master_st->st_sub_th->th_sync = th->th_sync;
-		master_st->st_sub_th->th_tags = th->th_tags;
 		master_st->st_sub_th->th_result = th->th_result;
 		rc = dt_trans_stop(env, master_st->st_dt, master_st->st_sub_th);
 		/* If it does not write_updates, then we call submit callback
@@ -1019,6 +1018,8 @@ stop_master_trans:
 			sub_trans_commit_cb_internal(tmt,
 						master_st->st_sub_th, rc);
 		if (rc < 0) {
+			CERROR("%s: stop trans failed: rc = %d\n",
+			       master_dev->dd_lu_dev.ld_obd->obd_name, rc);
 			th->th_result = rc;
 			GOTO(stop_other_trans, rc);
 		} else if (tur != NULL && tur->tur_update_records != NULL) {
@@ -1056,6 +1057,9 @@ stop_master_trans:
 
 			rc = sub_updates_write(env, lur, st);
 			if (rc < 0) {
+				CERROR("%s: write updates failed: rc = %d\n",
+				       st->st_dt->dd_lu_dev.ld_obd->obd_name,
+				       rc);
 				th->th_result = rc;
 				break;
 			}
@@ -1072,12 +1076,15 @@ stop_other_trans:
 			st->st_sub_th->th_sync = th->th_sync;
 		if (th->th_local)
 			st->st_sub_th->th_local = th->th_local;
-		st->st_sub_th->th_tags = th->th_tags;
 		st->st_sub_th->th_result = th->th_result;
 		rc = dt_trans_stop(env, st->st_sub_th->th_dev,
 				   st->st_sub_th);
-		if (unlikely(rc < 0 && th->th_result == 0))
-			th->th_result = rc;
+		if (rc < 0) {
+			CERROR("%s: stop trans failed: rc = %d\n",
+			       st->st_dt->dd_lu_dev.ld_obd->obd_name, rc);
+			if (th->th_result == 0)
+				th->th_result = rc;
+		}
 	}
 
 	rc = top_trans_wait_result(top_th);
@@ -1297,9 +1304,9 @@ static int distribute_txn_cancel_records(const struct lu_env *env,
 			rc = llog_cat_cancel_records(env, ctxt->loc_handle, 1,
 						     cookie);
 			CDEBUG(D_HA, "%s: batchid %llu cancel update log "
-			       DOSTID ".%u : rc = %d\n", obd->obd_name,
+			       DFID".%u: rc = %d\n", obd->obd_name,
 			       tmt->tmt_batchid,
-			       POSTID(&cookie->lgc_lgl.lgl_oi),
+			       PFID(&cookie->lgc_lgl.lgl_oi.oi_fid),
 			       cookie->lgc_index, rc);
 		}
 
@@ -1420,6 +1427,7 @@ distribute_txn_commit_batchid_update(const struct lu_env *env,
 
 	th = dt_trans_create(env, tdtd->tdtd_lut->lut_bottom);
 	if (IS_ERR(th)) {
+		atomic_dec(&tdtd->tdtd_refcount);
 		OBD_FREE_PTR(dtbd);
 		RETURN(PTR_ERR(th));
 	}
@@ -1450,8 +1458,10 @@ distribute_txn_commit_batchid_update(const struct lu_env *env,
 
 stop:
 	dt_trans_stop(env, tdtd->tdtd_lut->lut_bottom, th);
-	if (rc < 0)
+	if (rc < 0) {
+		atomic_dec(&tdtd->tdtd_refcount);
 		OBD_FREE_PTR(dtbd);
+	}
 	RETURN(rc);
 }
 
@@ -1519,7 +1529,7 @@ distribute_txn_commit_batchid_init(const struct lu_env *env,
 
 out_put:
 	if (rc < 0 && dt_obj != NULL) {
-		lu_object_put(env, &dt_obj->do_lu);
+		dt_object_put(env, dt_obj);
 		tdtd->tdtd_batchid_obj = NULL;
 	}
 	return rc;
@@ -1542,9 +1552,8 @@ static int distribute_txn_commit_thread(void *_arg)
 	struct target_distribute_txn_data *tdtd = _arg;
 	struct lu_target	*lut = tdtd->tdtd_lut;
 	struct ptlrpc_thread	*thread = &lut->lut_tdtd_commit_thread;
-	struct l_wait_info	 lwi = { 0 };
 	struct lu_env		 env;
-	struct list_head	 list;
+	LIST_HEAD(list);
 	int			 rc;
 	struct top_multiple_thandle *tmt;
 	struct top_multiple_thandle *tmp;
@@ -1560,7 +1569,6 @@ static int distribute_txn_commit_thread(void *_arg)
 	thread->t_flags = SVC_RUNNING;
 	spin_unlock(&tdtd->tdtd_batchid_lock);
 	wake_up(&thread->t_ctl_waitq);
-	INIT_LIST_HEAD(&list);
 
 	CDEBUG(D_HA, "%s: start commit thread committed batchid %llu\n",
 	       tdtd->tdtd_lut->lut_obd->obd_name,
@@ -1633,14 +1641,14 @@ static int distribute_txn_commit_thread(void *_arg)
 			top_multiple_thandle_put(tmt);
 		}
 
-		l_wait_event(tdtd->tdtd_commit_thread_waitq,
-			     !distribute_txn_commit_thread_running(lut) ||
-			     committed < tdtd->tdtd_committed_batchid ||
-			     tdtd_ready_for_cancel_log(tdtd), &lwi);
+		wait_event_idle(tdtd->tdtd_commit_thread_waitq,
+				!distribute_txn_commit_thread_running(lut) ||
+				committed < tdtd->tdtd_committed_batchid ||
+				tdtd_ready_for_cancel_log(tdtd));
 	};
 
-	l_wait_event(tdtd->tdtd_commit_thread_waitq,
-		     atomic_read(&tdtd->tdtd_refcount) == 0, &lwi);
+	wait_event_idle(tdtd->tdtd_commit_thread_waitq,
+			atomic_read(&tdtd->tdtd_refcount) == 0);
 
 	spin_lock(&tdtd->tdtd_batchid_lock);
 	list_for_each_entry_safe(tmt, tmp, &tdtd->tdtd_list,
@@ -1680,7 +1688,6 @@ int distribute_txn_init(const struct lu_env *env,
 			__u32 index)
 {
 	struct task_struct	*task;
-	struct l_wait_info	 lwi = { 0 };
 	int			rc;
 	ENTRY;
 
@@ -1701,18 +1708,21 @@ int distribute_txn_init(const struct lu_env *env,
 	atomic_set(&tdtd->tdtd_recovery_threads_count, 0);
 
 	tdtd->tdtd_lut = lut;
+	if (lut->lut_bottom->dd_rdonly)
+		RETURN(0);
+
 	rc = distribute_txn_commit_batchid_init(env, tdtd);
 	if (rc != 0)
 		RETURN(rc);
 
-	task = kthread_run(distribute_txn_commit_thread, tdtd, "tdtd-%u",
+	task = kthread_run(distribute_txn_commit_thread, tdtd, "dist_txn-%u",
 			   index);
 	if (IS_ERR(task))
 		RETURN(PTR_ERR(task));
 
-	l_wait_event(lut->lut_tdtd_commit_thread.t_ctl_waitq,
-		     distribute_txn_commit_thread_running(lut) ||
-		     distribute_txn_commit_thread_stopped(lut), &lwi);
+	wait_event_idle(lut->lut_tdtd_commit_thread.t_ctl_waitq,
+			distribute_txn_commit_thread_running(lut) ||
+			distribute_txn_commit_thread_stopped(lut));
 	RETURN(0);
 }
 EXPORT_SYMBOL(distribute_txn_init);
@@ -1743,7 +1753,7 @@ void distribute_txn_fini(const struct lu_env *env,
 
 	dtrq_list_destroy(tdtd);
 	if (tdtd->tdtd_batchid_obj != NULL) {
-		lu_object_put(env, &tdtd->tdtd_batchid_obj->do_lu);
+		dt_object_put(env, tdtd->tdtd_batchid_obj);
 		tdtd->tdtd_batchid_obj = NULL;
 	}
 }