X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Ftarget%2Fupdate_trans.c;h=36ab33192e1ff58cde35f8430af24c33c6eed076;hb=03bbd4c27471751ada57282fad15e074ae01e9d7;hp=7a3243b902319cd0f60baa2ea9520c8ba90a413d;hpb=af13cfff297d4882de35fb7c11bf5261293b8287;p=fs%2Flustre-release.git

diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c
index 7a3243b..36ab331 100644
--- a/lustre/target/update_trans.c
+++ b/lustre/target/update_trans.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2014, Intel Corporation.
+ * Copyright (c) 2015, 2016, Intel Corporation.
  */
 /*
  * lustre/target/update_trans.c
@@ -72,8 +72,7 @@ static void top_multiple_thandle_dump(struct top_multiple_thandle *tmt,
 	struct sub_thandle	*st;
 
 	LASSERT(tmt->tmt_magic == TOP_THANDLE_MAGIC);
-	CDEBUG(mask, "%s tmt %p refcount %d committed %d result %d"
-	       "batchid "LPU64"\n",
+	CDEBUG(mask, "%s tmt %p refcount %d committed %d result %d batchid %llu\n",
 	       tmt->tmt_master_sub_dt ?
 	       tmt->tmt_master_sub_dt->dd_lu_dev.ld_obd->obd_name :
 	       "NULL",
@@ -83,13 +82,13 @@ 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 sub_th %p\n",
+		CDEBUG(mask, "st %p obd %s committed %d stopped %d sub_th %p\n",
 		       st, st->st_dt->dd_lu_dev.ld_obd->obd_name,
-		       st->st_committed, st->st_sub_th);
+		       st->st_committed, st->st_stopped, 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);
 		}
 	}
@@ -168,19 +167,20 @@ static int sub_updates_write(const struct lu_env *env,
 			     struct llog_update_record *record,
 			     struct sub_thandle *sub_th)
 {
-	struct dt_device	*dt = sub_th->st_dt;
-	struct llog_ctxt	*ctxt;
-	int			rc;
+	struct dt_device *dt = sub_th->st_dt;
+	struct llog_ctxt *ctxt;
 	struct llog_update_record *lur = NULL;
-	struct update_params	*params = NULL;
-	__u32			update_count = 0;
-	__u32			param_count = 0;
-	__u32			last_update_count = 0;
-	__u32			last_param_count = 0;
-	void			*src;
-	void			*start;
-	void			*next;
+	__u32 update_count = 0;
+	__u32 param_count = 0;
+	__u32 last_update_count = 0;
+	__u32 last_param_count = 0;
+	char *start;
+	char *cur;
+	char *next;
 	struct sub_thandle_cookie *stc;
+	size_t reclen;
+	bool eof = false;
+	int rc;
 	ENTRY;
 
 	ctxt = llog_get_context(dt->dd_lu_dev.ld_obd,
@@ -210,9 +210,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) {
@@ -233,87 +233,83 @@ static int sub_updates_write(const struct lu_env *env,
 	memcpy(lur, &record->lur_hdr, sizeof(record->lur_hdr));
 	lur->lur_update_rec.ur_update_count = 0;
 	lur->lur_update_rec.ur_param_count = 0;
-	src = &record->lur_update_rec.ur_ops;
-	start = next = src;
-	lur->lur_hdr.lrh_len = llog_update_record_size(lur);
-	params = update_records_get_params(&record->lur_update_rec);
+	start = (char *)&record->lur_update_rec.ur_ops;
+	cur = next = start;
 	do {
-		size_t rec_len;
-
-		if (update_count < record->lur_update_rec.ur_update_count) {
-			next = update_op_next_op((struct update_op *)src);
-		} else {
-			if (param_count == 0)
-				next = update_records_get_params(
-						&record->lur_update_rec);
-			else
-				next = (char *)src +
-					object_update_param_size(
-					(struct object_update_param *)src);
+		if (update_count < record->lur_update_rec.ur_update_count)
+			next = (char *)update_op_next_op(
+						(struct update_op *)cur);
+		else if (param_count < record->lur_update_rec.ur_param_count)
+			next = (char *)update_param_next_param(
+						(struct update_param *)cur);
+		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 ||
+		     reclen == ctxt->loc_chunk_size) &&
+		    !eof) {
+			cur = next;
+
+			if (update_count <
+			    record->lur_update_rec.ur_update_count)
+				update_count++;
+			else if (param_count <
+				 record->lur_update_rec.ur_param_count)
+				param_count++;
+			continue;
 		}
 
-		rec_len = cfs_size_round((unsigned long)(next - src));
-		/* If its size > llog chunk_size, then write current chunk to
-		 * the update llog. */
-		if (lur->lur_hdr.lrh_len + rec_len + LLOG_MIN_REC_SIZE >
-		    ctxt->loc_chunk_size ||
-		    param_count == record->lur_update_rec.ur_param_count) {
-			lur->lur_update_rec.ur_update_count =
-				update_count > last_update_count ?
-				update_count - last_update_count : 0;
-			lur->lur_update_rec.ur_param_count = param_count -
-							     last_param_count;
-
-			memcpy(&lur->lur_update_rec.ur_ops, start,
-			       (unsigned long)(src - start));
-			if (last_update_count != 0)
-				lur->lur_update_rec.ur_flags |=
-						UPDATE_RECORD_CONTINUE;
-
-			update_records_dump(&lur->lur_update_rec, D_INFO, true);
-			lur->lur_hdr.lrh_len = llog_update_record_size(lur);
-			LASSERT(lur->lur_hdr.lrh_len <= ctxt->loc_chunk_size);
-
-			OBD_ALLOC_PTR(stc);
-			if (stc == NULL)
-				GOTO(llog_put, rc = -ENOMEM);
-			INIT_LIST_HEAD(&stc->stc_list);
-
-			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", dt->dd_lu_dev.ld_obd->obd_name,
-			       POSTID(&stc->stc_cookie.lgc_lgl.lgl_oi),
-			       stc->stc_cookie.lgc_index, rc);
-
-			if (rc > 0) {
-				list_add(&stc->stc_list,
-					 &sub_th->st_cookie_list);
-				rc = 0;
-			} else {
-				OBD_FREE_PTR(stc);
-				GOTO(llog_put, rc);
-			}
+		lur->lur_update_rec.ur_update_count = update_count -
+						      last_update_count;
+		lur->lur_update_rec.ur_param_count = param_count -
+						     last_param_count;
+		memcpy(&lur->lur_update_rec.ur_ops, start, cur - start);
+		lur->lur_hdr.lrh_len = llog_update_record_size(lur);
 
-			last_update_count = update_count;
-			last_param_count = param_count;
-			start = src;
-			lur->lur_update_rec.ur_update_count = 0;
-			lur->lur_update_rec.ur_param_count = 0;
-			lur->lur_hdr.lrh_len = llog_update_record_size(lur);
+		LASSERT(lur->lur_hdr.lrh_len ==
+			 __llog_update_record_size(
+				__update_records_size(cur - start)));
+		LASSERT(lur->lur_hdr.lrh_len <= ctxt->loc_chunk_size);
+
+		update_records_dump(&lur->lur_update_rec, D_INFO, true);
+
+		OBD_ALLOC_PTR(stc);
+		if (stc == NULL)
+			GOTO(llog_put, rc = -ENOMEM);
+		INIT_LIST_HEAD(&stc->stc_list);
+
+		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 "DFID".%u: rc = %d\n",
+			dt->dd_lu_dev.ld_obd->obd_name,
+			PFID(&stc->stc_cookie.lgc_lgl.lgl_oi.oi_fid),
+			stc->stc_cookie.lgc_index, rc);
+
+		if (rc > 0) {
+			list_add(&stc->stc_list, &sub_th->st_cookie_list);
+			rc = 0;
+		} else {
+			OBD_FREE_PTR(stc);
+			GOTO(llog_put, rc);
 		}
 
-		src = next;
-		lur->lur_hdr.lrh_len += cfs_size_round(rec_len);
-		if (update_count < record->lur_update_rec.ur_update_count)
-			update_count++;
-		else if (param_count < record->lur_update_rec.ur_param_count)
-			param_count++;
-		else
-			break;
-	} while (1);
+		last_update_count = update_count;
+		last_param_count = param_count;
+		start = cur;
+		lur->lur_update_rec.ur_update_count = 0;
+		lur->lur_update_rec.ur_param_count = 0;
+		lur->lur_update_rec.ur_flags |= UPDATE_RECORD_CONTINUE;
+	} while (!eof);
 
 llog_put:
 	if (lur != NULL)
@@ -450,25 +446,11 @@ struct sub_thandle *create_sub_thandle(struct top_multiple_thandle *tmt,
 	return st;
 }
 
-/**
- * sub thandle commit callback
- *
- * Mark the sub thandle to be committed and if all sub thandle are committed
- * notify the top thandle.
- *
- * \param[in] env	execution environment
- * \param[in] sub_th	sub thandle being committed
- * \param[in] cb	commit callback
- * \param[in] err	trans result
- */
-static void sub_trans_commit_cb(struct lu_env *env,
-				struct thandle *sub_th,
-				struct dt_txn_commit_cb *cb, int err)
+static void sub_trans_commit_cb_internal(struct top_multiple_thandle *tmt,
+					 struct thandle *sub_th, int err)
 {
 	struct sub_thandle	*st;
-	struct top_multiple_thandle *tmt = cb->dcb_data;
 	bool			all_committed = true;
-	ENTRY;
 
 	/* Check if all sub thandles are committed */
 	spin_lock(&tmt->tmt_sub_lock);
@@ -493,6 +475,26 @@ static void sub_trans_commit_cb(struct lu_env *env,
 	RETURN_EXIT;
 }
 
+/**
+ * sub thandle commit callback
+ *
+ * Mark the sub thandle to be committed and if all sub thandle are committed
+ * notify the top thandle.
+ *
+ * \param[in] env	execution environment
+ * \param[in] sub_th	sub thandle being committed
+ * \param[in] cb	commit callback
+ * \param[in] err	trans result
+ */
+static void sub_trans_commit_cb(struct lu_env *env,
+				struct thandle *sub_th,
+				struct dt_txn_commit_cb *cb, int err)
+{
+	struct top_multiple_thandle *tmt = cb->dcb_data;
+
+	sub_trans_commit_cb_internal(tmt, sub_th, err);
+}
+
 static void sub_thandle_register_commit_cb(struct sub_thandle *st,
 				    struct top_multiple_thandle *tmt)
 {
@@ -572,6 +574,7 @@ int sub_thandle_trans_create(const struct lu_env *env,
 	st->st_sub_th = sub_th;
 
 	sub_th->th_wait_submit = 1;
+	sub_thandle_register_stop_cb(st, top_th->tt_multiple_thandle);
 	return 0;
 }
 
@@ -661,6 +664,7 @@ static void distribute_txn_assign_batchid(struct top_multiple_thandle *new)
 {
 	struct target_distribute_txn_data *tdtd;
 	struct dt_device *dt = new->tmt_master_sub_dt;
+	struct sub_thandle *st;
 
 	LASSERT(dt != NULL);
 	tdtd = dt2lu_dev(dt)->ld_site->ls_tgt->lut_tdtd;
@@ -668,6 +672,10 @@ static void distribute_txn_assign_batchid(struct top_multiple_thandle *new)
 	new->tmt_batchid = tdtd->tdtd_batchid++;
 	list_add_tail(&new->tmt_commit_list, &tdtd->tdtd_list);
 	spin_unlock(&tdtd->tdtd_batchid_lock);
+	list_for_each_entry(st, &new->tmt_sub_thandle_list, st_sub_list) {
+		if (st->st_sub_th != NULL)
+			sub_thandle_register_commit_cb(st, new);
+	}
 	top_multiple_thandle_get(new);
 	top_multiple_thandle_dump(new, D_INFO);
 }
@@ -684,6 +692,7 @@ void distribute_txn_insert_by_batchid(struct top_multiple_thandle *new)
 	struct dt_device *dt = new->tmt_master_sub_dt;
 	struct top_multiple_thandle *tmt;
 	struct target_distribute_txn_data *tdtd;
+	struct sub_thandle *st;
 	bool	at_head = false;
 
 	LASSERT(dt != NULL);
@@ -701,6 +710,12 @@ void distribute_txn_insert_by_batchid(struct top_multiple_thandle *new)
 		list_add(&new->tmt_commit_list, &tdtd->tdtd_list);
 	}
 	spin_unlock(&tdtd->tdtd_batchid_lock);
+
+	list_for_each_entry(st, &new->tmt_sub_thandle_list, st_sub_list) {
+		if (st->st_sub_th != NULL)
+			sub_thandle_register_commit_cb(st, new);
+	}
+
 	top_multiple_thandle_get(new);
 	top_multiple_thandle_dump(new, D_INFO);
 	if (new->tmt_committed && at_head)
@@ -773,6 +788,11 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
 	ENTRY;
 
 	if (tmt == NULL) {
+		if (th->th_sync)
+			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);
@@ -788,15 +808,16 @@ int top_trans_start(const struct lu_env *env, struct dt_device *master_dev,
 			continue;
 		if (th->th_sync)
 			st->st_sub_th->th_sync = th->th_sync;
-		st->st_sub_th->th_local = th->th_local;
+		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)
 			GOTO(out, rc);
 
-		sub_thandle_register_stop_cb(st, tmt);
-		sub_thandle_register_commit_cb(st, tmt);
+		LASSERT(st->st_started == 0);
+		st->st_started = 1;
 	}
 out:
 	th->th_result = rc;
@@ -922,6 +943,12 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
 
 	if (likely(top_th->tt_multiple_thandle == NULL)) {
 		LASSERT(master_dev != NULL);
+
+		if (th->th_sync)
+			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);
@@ -949,6 +976,7 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
 			CERROR("%s: cannot prepare updates: rc = %d\n",
 			       master_dev->dd_lu_dev.ld_obd->obd_name, rc);
 			th->th_result = rc;
+			write_updates = false;
 			GOTO(stop_master_trans, rc);
 		}
 
@@ -967,6 +995,7 @@ int top_trans_stop(const struct lu_env *env, struct dt_device *master_dev,
 			CERROR("%s: write updates failed: rc = %d\n",
 			       master_dev->dd_lu_dev.ld_obd->obd_name, rc);
 			th->th_result = rc;
+			write_updates = false;
 			GOTO(stop_master_trans, rc);
 		}
 	}
@@ -975,12 +1004,20 @@ stop_master_trans:
 	/* Step 2: Stop the transaction on the master MDT, and fill the
 	 * master transno in the update logs to other MDT. */
 	if (master_st != NULL && master_st->st_sub_th != NULL) {
-		master_st->st_sub_th->th_local = th->th_local;
+		if (th->th_local)
+			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
+		 * here, otherwise callback is done through
+		 * osd(osp)_trans_commit_cb() */
+		if (!master_st->st_started &&
+		    !list_empty(&tmt->tmt_commit_list))
+			sub_trans_commit_cb_internal(tmt,
+						master_st->st_sub_th, rc);
 		if (rc < 0) {
 			th->th_result = rc;
 			GOTO(stop_other_trans, rc);
@@ -1033,7 +1070,8 @@ stop_other_trans:
 
 		if (th->th_sync)
 			st->st_sub_th->th_sync = th->th_sync;
-		st->st_sub_th->th_local = th->th_local;
+		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,
@@ -1103,6 +1141,7 @@ create_sub_thandle_with_thandle(struct top_thandle *top_th,
 	st->st_sub_th = sub_th;
 
 	sub_th->th_top = &top_th->tt_super;
+	sub_thandle_register_stop_cb(st, top_th->tt_multiple_thandle);
 	return st;
 }
 
@@ -1123,6 +1162,7 @@ struct thandle *thandle_get_sub_by_dt(const struct lu_env *env,
 				      struct dt_device *sub_dt)
 {
 	struct sub_thandle	*st = NULL;
+	struct sub_thandle	*master_st = NULL;
 	struct top_thandle	*top_th;
 	struct thandle		*sub_th = NULL;
 	int			rc = 0;
@@ -1156,19 +1196,29 @@ struct thandle *thandle_get_sub_by_dt(const struct lu_env *env,
 		/* Add master sub th to the top trans list */
 		tmt->tmt_master_sub_dt =
 			top_th->tt_master_sub_thandle->th_dev;
-		st = create_sub_thandle_with_thandle(top_th,
-				top_th->tt_master_sub_thandle);
-		if (IS_ERR(st))
-			GOTO(stop_trans, rc = PTR_ERR(st));
+		master_st = create_sub_thandle_with_thandle(top_th,
+					top_th->tt_master_sub_thandle);
+		if (IS_ERR(master_st)) {
+			rc = PTR_ERR(master_st);
+			master_st = NULL;
+			GOTO(stop_trans, rc);
+		}
 	}
 
 	/* create and init sub th to the top trans list */
 	st = create_sub_thandle_with_thandle(top_th, sub_th);
+	if (IS_ERR(st)) {
+		rc = PTR_ERR(st);
+		st = NULL;
+		GOTO(stop_trans, rc);
+	}
 	st->st_sub_th->th_wait_submit = 1;
 stop_trans:
 	if (rc < 0) {
-		if (st != NULL)
-			OBD_FREE_PTR(st);
+		if (master_st != NULL) {
+			list_del(&master_st->st_sub_list);
+			OBD_FREE_PTR(master_st);
+		}
 		sub_th->th_result = rc;
 		dt_trans_stop(env, sub_dt, sub_th);
 		sub_th = ERR_PTR(rc);
@@ -1247,9 +1297,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);
 		}
 
@@ -1370,6 +1420,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));
 	}
@@ -1395,13 +1446,15 @@ distribute_txn_commit_batchid_update(const struct lu_env *env,
 	rc = dt_record_write(env, tdtd->tdtd_batchid_obj, &buf,
 			     &off, th);
 
-	CDEBUG(D_INFO, "%s: update batchid "LPU64": rc = %d\n",
+	CDEBUG(D_INFO, "%s: update batchid %llu: rc = %d\n",
 	       tdtd->tdtd_lut->lut_obd->obd_name, batchid, rc);
 
 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);
 }
 
@@ -1469,7 +1522,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;
@@ -1512,7 +1565,7 @@ static int distribute_txn_commit_thread(void *_arg)
 	wake_up(&thread->t_ctl_waitq);
 	INIT_LIST_HEAD(&list);
 
-	CDEBUG(D_HA, "%s: start commit thread committed batchid "LPU64"\n",
+	CDEBUG(D_HA, "%s: start commit thread committed batchid %llu\n",
 	       tdtd->tdtd_lut->lut_obd->obd_name,
 	       tdtd->tdtd_committed_batchid);
 
@@ -1532,8 +1585,8 @@ static int distribute_txn_commit_thread(void *_arg)
 				list_move_tail(&tmt->tmt_commit_list, &list);
 			} else if (!tdtd->tdtd_lut->lut_obd->obd_recovering) {
 				LASSERTF(tmt->tmt_batchid >= batchid,
-					 "tmt %p tmt_batchid: "LPU64", batchid "
-					  LPU64"\n", tmt, tmt->tmt_batchid,
+					 "tmt %p tmt_batchid: %llu, batchid "
+					  "%llu\n", tmt, tmt->tmt_batchid,
 					 batchid);
 				/* There are three types of distribution
 				 * transaction result
@@ -1561,8 +1614,8 @@ static int distribute_txn_commit_thread(void *_arg)
 		}
 		spin_unlock(&tdtd->tdtd_batchid_lock);
 
-		CDEBUG(D_HA, "%s: batchid: "LPU64" committed batchid "
-		       LPU64"\n", tdtd->tdtd_lut->lut_obd->obd_name, batchid,
+		CDEBUG(D_HA, "%s: batchid: %llu committed batchid "
+		       "%llu\n", tdtd->tdtd_lut->lut_obd->obd_name, batchid,
 		       tdtd->tdtd_committed_batchid);
 		/* update globally committed on a storage */
 		if (batchid > tdtd->tdtd_committed_batchid) {
@@ -1634,21 +1687,31 @@ int distribute_txn_init(const struct lu_env *env,
 	int			rc;
 	ENTRY;
 
-	spin_lock_init(&tdtd->tdtd_batchid_lock);
 	INIT_LIST_HEAD(&tdtd->tdtd_list);
+	INIT_LIST_HEAD(&tdtd->tdtd_replay_finish_list);
+	INIT_LIST_HEAD(&tdtd->tdtd_replay_list);
+	spin_lock_init(&tdtd->tdtd_batchid_lock);
+	spin_lock_init(&tdtd->tdtd_replay_list_lock);
+	tdtd->tdtd_replay_handler = distribute_txn_replay_handle;
+	tdtd->tdtd_replay_ready = 0;
 
 	tdtd->tdtd_batchid = lut->lut_last_transno + 1;
 
 	init_waitqueue_head(&lut->lut_tdtd_commit_thread.t_ctl_waitq);
 	init_waitqueue_head(&tdtd->tdtd_commit_thread_waitq);
+	init_waitqueue_head(&tdtd->tdtd_recovery_threads_waitq);
 	atomic_set(&tdtd->tdtd_refcount, 0);
+	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));
@@ -1686,7 +1749,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;
 	}
 }