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

diff --git a/lustre/target/update_trans.c b/lustre/target/update_trans.c
index 95d1fb6..4d9a4c0 100644
--- a/lustre/target/update_trans.c
+++ b/lustre/target/update_trans.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2014, 2015, Intel Corporation.
+ * Copyright (c) 2015, 2017, 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,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);
 		}
 	}
@@ -168,19 +169,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,
@@ -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) {
@@ -233,87 +245,76 @@ 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;
+
+		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)
@@ -527,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;
@@ -537,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;
@@ -616,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;
 }
@@ -796,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);
@@ -814,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)
@@ -911,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);
 }
@@ -952,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);
@@ -1012,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
@@ -1023,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) {
@@ -1060,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;
 			}
@@ -1076,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);
@@ -1301,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);
 		}
 
@@ -1424,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));
 	}
@@ -1449,13 +1453,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);
 }
 
@@ -1523,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;
@@ -1546,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;
@@ -1564,9 +1569,8 @@ 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 "LPU64"\n",
+	CDEBUG(D_HA, "%s: start commit thread committed batchid %llu\n",
 	       tdtd->tdtd_lut->lut_obd->obd_name,
 	       tdtd->tdtd_committed_batchid);
 
@@ -1586,8 +1590,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
@@ -1615,8 +1619,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) {
@@ -1637,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,
@@ -1684,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;
 
@@ -1700,21 +1703,26 @@ int distribute_txn_init(const struct lu_env *env,
 
 	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));
 
-	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);
@@ -1745,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;
 	}
 }