DAOS-16097 vos: assign persistent DTX entry in vos_dtx_prepared

Assign persistent DTX entry only via vos_dtx_prepared() that will
initialize such DTX entry immediately to avoid any potential race
between persistently allocating DTX entry and initializing it.

Add some check (for DTX flag) after DTX locally prepared.

Do not allow current transaction to deregister the record that is
referenced by another prepared (but non-committed) DTX.

Test-tag: test_ec_online_rebuild_fio
Skip-func-hw-test-large-md-on-ssd: false
Test-repeat-hw-large: 5

Signed-off-by: Fan Yong <fan.yong@intel.com>

src/dtx/dtx_rpc.c

@@ -1208,6 +1208,9 @@ dtx_refresh_internal(struct ds_cont_child *cont, int *check_count, d_list_t *che
 	/* Handle the entries whose leaders are on current server. */
 	d_list_for_each_entry_safe(dsp, tmp, &self, dsp_link) {
 		struct dtx_entry	dte;
+		struct dtx_entry	*pdte = &dte;
+		struct dtx_cos_key	 dck;
+
 
 		d_list_del(&dsp->dsp_link);
 
@@ -1216,13 +1219,32 @@ dtx_refresh_internal(struct ds_cont_child *cont, int *check_count, d_list_t *che
 		dte.dte_refs = 1;
 		dte.dte_mbs = dsp->dsp_mbs;
 
+		if (for_io) {
+			rc = vos_dtx_check(cont->sc_hdl, &dsp->dsp_xid, NULL, NULL, NULL, NULL,
+					   false);
+			switch(rc) {
+			case DTX_ST_COMMITTABLE:
+				dck.oid = dsp->dsp_oid;
+				dck.dkey_hash = dsp->dsp_dkey_hash;
+				rc = dtx_commit(cont, &pdte, &dck, 1);
+				if (rc < 0 && rc != -DER_NONEXIST && for_io)
+					d_list_add_tail(&dsp->dsp_link, cmt_list);
+				else
+					dtx_dsp_free(dsp);
+				continue;
+			case DTX_ST_COMMITTED:
+			case -DER_NONEXIST: /* Aborted */
+				dtx_dsp_free(dsp);
+				continue;
+			default:
+				break;
+			}
+		}
+
 		rc = dtx_status_handle_one(cont, &dte, dsp->dsp_oid, dsp->dsp_dkey_hash,
 					   dsp->dsp_epoch, NULL, NULL);
 		switch (rc) {
 		case DSHR_NEED_COMMIT: {
-			struct dtx_entry	*pdte = &dte;
-			struct dtx_cos_key	 dck;
-
 			dck.oid = dsp->dsp_oid;
 			dck.dkey_hash = dsp->dsp_dkey_hash;
 			rc = dtx_commit(cont, &pdte, &dck, 1);

src/vos/vos_common.c

@@ -269,13 +269,15 @@ vos_tx_end(struct vos_container *cont, struct dtx_handle *dth_in,
 	   struct umem_rsrvd_act **rsrvd_scmp, d_list_t *nvme_exts,
 	   bool started, struct bio_desc *biod, int err)
 {
-	struct vos_pool         *pool;
-	struct dtx_handle	*dth = dth_in;
-	struct vos_dtx_act_ent	*dae;
-	struct dtx_rsrvd_uint	*dru;
-	struct vos_dtx_cmt_ent	*dce = NULL;
-	struct dtx_handle	 tmp = {0};
-	int			 rc;
+	struct vos_pool         	*pool;
+	struct umem_instance		*umm;
+	struct dtx_handle		*dth = dth_in;
+	struct vos_dtx_act_ent		*dae;
+	struct vos_dtx_act_ent_df	*dae_df;
+	struct dtx_rsrvd_uint		*dru;
+	struct vos_dtx_cmt_ent		*dce = NULL;
+	struct dtx_handle		 tmp = {0};
+	int				 rc = 0;
 
 	if (!dtx_is_valid_handle(dth)) {
 		/** Created a dummy dth handle for publishing extents */
@@ -287,11 +289,11 @@ vos_tx_end(struct vos_container *cont, struct dtx_handle *dth_in,
 		D_INIT_LIST_HEAD(&tmp.dth_deferred_nvme);
 	}
 
-	if (dth->dth_local) {
+	if (dth->dth_local)
 		pool = vos_hdl2pool(dth_in->dth_poh);
-	} else {
+	else
 		pool = cont->vc_pool;
-	}
+	umm = vos_pool2umm(pool);
 
 	if (rsrvd_scmp != NULL) {
 		D_ASSERT(nvme_exts != NULL);
@@ -300,7 +302,7 @@ vos_tx_end(struct vos_container *cont, struct dtx_handle *dth_in,
 			 * Just do your best to release the SCM reservation. Can't handle another
 			 * error while handling one already anyway.
 			 */
-			(void)vos_publish_scm(vos_pool2umm(pool), *rsrvd_scmp, false /* publish */);
+			(void)vos_publish_scm(umm, *rsrvd_scmp, false /* publish */);
 			D_FREE(*rsrvd_scmp);
 			*rsrvd_scmp = NULL;
 			err         = -DER_NOMEM;
@@ -341,9 +343,9 @@ vos_tx_end(struct vos_container *cont, struct dtx_handle *dth_in,
 	vos_dth_set(NULL, pool->vp_sysdb);
 
 	if (bio_nvme_configured(SMD_DEV_TYPE_META) && biod != NULL)
-		err = umem_tx_end_ex(vos_pool2umm(pool), err, biod);
+		err = umem_tx_end_ex(umm, err, biod);
 	else
-		err = umem_tx_end(vos_pool2umm(pool), err);
+		err = umem_tx_end(umm, err);
 
 cancel:
 	if (dtx_is_valid_handle(dth_in)) {
@@ -409,8 +411,11 @@ vos_tx_end(struct vos_container *cont, struct dtx_handle *dth_in,
 				vos_dtx_post_handle(cont, &dae, &dce, 1, false, err != 0);
 			} else {
 				D_ASSERT(dce == NULL);
-				if (err == 0)
+				if (err == 0) {
 					dae->dae_prepared = 1;
+					dae_df = umem_off2ptr(umm, dae->dae_df_off);
+					D_ASSERT(!(dae_df->dae_flags & DTE_INVALID));
+				}
 			}
 		}
 	}