cleanup DtmLocalFinish and get rid of x->gtid

kelvich · kelvich · commit 57002d158426 · 2018-03-02T15:16:38.000+03:00
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
@@ -483,14 +483,14 @@ begin_remote_xact(ConnCacheEntry *entry)
 								++two_phase_xact_count);
 				MemoryContextSwitchTo(oldcxt);
 
-				current_global_cid = DtmLocalExtend(two_phase_xact_gid);
+				current_global_cid = DtmLocalExtend();
 			}
 
 			Assert(two_phase_xact_gid);
 			/* join the new participant */
 			res = PQexec(entry->conn,
-						psprintf("SELECT pg_global_snaphot_join("UINT64_FORMAT", '%s')",
-								current_global_cid, two_phase_xact_gid));
+						psprintf("SELECT pg_global_snaphot_join("UINT64_FORMAT")",
+								current_global_cid));
 
 			if (PQresultStatus(res) != PGRES_TUPLES_OK)
 			{
diff --git a/src/backend/access/transam/global_snapshot.c b/src/backend/access/transam/global_snapshot.c
@@ -295,6 +295,29 @@ DtmAdjustSubtransactions(DtmTransStatus *ts)
 	}
 }
 
+/*
+ * Add subtransactions to finished transactions list.
+ * Copy CSN and status of parent transaction.
+ */
+static void
+DtmAddSubtransactions(DtmTransStatus * ts, TransactionId *subxids, int nSubxids)
+{
+	int			i;
+
+	for (i = 0; i < nSubxids; i++)
+	{
+		bool		found;
+		DtmTransStatus *sts;
+
+		Assert(TransactionIdIsValid(subxids[i]));
+		sts = (DtmTransStatus *) hash_search(xid2status, &subxids[i], HASH_ENTER, &found);
+		Assert(!found);
+		sts->cid = ts->cid;
+		sts->nSubxids = 0;
+		DtmTransactionListInsertAfter(ts, sts);
+	}
+}
+
 /*
  * There can be different oldest XIDs at different cluster node.
  * Seince we do not have centralized aribiter, we have to rely in DtmVacuumDelay.
@@ -465,27 +488,19 @@ DtmLocalBegin(DtmCurrentTrans * x)
  * Returns snapshot of current transaction.
  */
 cid_t
-DtmLocalExtend(GlobalTransactionId gtid)
+DtmLocalExtend()
 {
-	DtmCurrentTrans *x = &dtm_tx;
-
-	if (gtid != NULL)
-	{
-		strncpy(x->gtid, gtid, MAX_GTID_SIZE);
-	}
 	DtmInitGlobalXmin(TransactionXmin);
-
 	dtm_tx.is_global = true;
-
-	return x->snapshot;
+	return dtm_tx.snapshot;
 }
 
 /*
  * This function is executed on all nodes joining distributed transaction.
  * global_cid is snapshot taken from node initiated this transaction
  */
 cid_t
-DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t global_cid)
+DtmLocalAccess(cid_t global_cid)
 {
 	cid_t		local_cid;
 
@@ -494,9 +509,8 @@ DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t global_cid)
 	SpinLockAcquire(&local->lock);
 	{
 		local_cid = dtm_sync(global_cid);
-		x->snapshot = global_cid;
+		dtm_tx.snapshot = global_cid;
 	}
-	strncpy(x->gtid, gtid, MAX_GTID_SIZE);
 	SpinLockRelease(&local->lock);
 
 	dtm_tx.is_global = true;
@@ -510,6 +524,37 @@ DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t global_cid)
 	return global_cid;
 }
 
+
+/*
+ * Save state of parepared transaction
+ */
+void
+DtmLocalSavePreparedState(DtmCurrentTrans * x)
+{
+
+	if (dtm_tx.is_global)
+	{
+		TransactionId *subxids;
+		TransactionId xid = GetCurrentTransactionId();
+		int			  nSubxids = xactGetCommittedChildren(&subxids);
+
+		SpinLockAcquire(&local->lock);
+		{
+			DtmTransStatus *ts;
+			bool found;
+
+			ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, &found);
+			Assert(!found);
+			ts->cid = InDoubtGlobalCSN;
+			ts->nSubxids = nSubxids;
+			DtmTransactionListAppend(ts);
+			DtmAddSubtransactions(ts, subxids, nSubxids);
+		}
+		SpinLockRelease(&local->lock);
+	}
+}
+
+
 /*
  * Set transaction status to in-doubt. Now all transactions accessing tuples updated by this transaction have to
  * wait until it is either committed either aborted
@@ -587,10 +632,8 @@ DtmLocalEndPrepare(GlobalTransactionId gtid, cid_t cid)
 			return; // global ro tx
 	}
 
-
 	dtm_tx.xid = xid;
 	dtm_tx.csn = cid;
-
 }
 
 /*
@@ -599,43 +642,57 @@ DtmLocalEndPrepare(GlobalTransactionId gtid, cid_t cid)
 void
 DtmLocalFinish(bool is_commit)
 {
-	DtmCurrentTrans * x = &dtm_tx;
 	TransactionId xid = GetCurrentTransactionIdIfAny();
+	bool		found;
+	DtmTransStatus *ts;
 
-	if (x->gtid[0] && finishing_prepared)
+	// We can't check just TransactionIdIsValid(dtm_tx.xid) because
+	// then we catch commit of `select pg_global_snaphot_end_prepare(...)`
+	if (TransactionIdIsValid(dtm_tx.xid) &&
+			(finishing_prepared ||				// commit prepared of global
+			 TransactionIdIsValid(xid)))		// ordinary commit of global
 	{
+		// Commit of global prepared tx
+
 		xid = dtm_tx.xid;
+		Assert(GlobalCSNIsNormal(dtm_tx.csn));
+
+		SpinLockAcquire(&local->lock);
+		ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_FIND, &found);
+		Assert(found);
+		ts->cid = is_commit ? dtm_tx.csn : AbortedGlobalCSN;
+		DtmAdjustSubtransactions(ts); // !
+		SpinLockRelease(&local->lock);
+
+		dtm_tx.xid = InvalidTransactionId;
+		dtm_tx.csn = InvalidGlobalCSN;
+		dtm_tx.is_global = false;
 	}
-	else if (!TransactionIdIsValid(xid))
+	else if (TransactionIdIsValid(xid))
 	{
-		return;
-	}
+		// Commit of local tx
 
-	SpinLockAcquire(&local->lock);
-	{
-		bool		found;
-		DtmTransStatus *ts;
+		TransactionId *subxids;
+		int nSubxids = xactGetCommittedChildren(&subxids);
 
-		ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, &found);
+		Assert(!GlobalCSNIsNormal(dtm_tx.csn));
+		Assert(!TransactionIdIsValid(dtm_tx.xid));
 
-		if (found)
+		if (dtm_tx.is_global)
 		{
-			Assert(GlobalCSNIsNormal(dtm_tx.csn));
-			ts->cid = is_commit ? dtm_tx.csn : AbortedGlobalCSN;
-
-			dtm_tx.xid = InvalidTransactionId;
-			dtm_tx.csn = InvalidGlobalCSN;
-		}
-		else
-		{
-			Assert(!GlobalCSNIsNormal(dtm_tx.csn));
-			ts->cid = is_commit ? dtm_get_cid() : AbortedGlobalCSN;
-			DtmTransactionListAppend(ts);
+			Assert(!is_commit);
+			dtm_tx.is_global = false;
 		}
-		DtmAdjustSubtransactions(ts);
-	}
-	SpinLockRelease(&local->lock);
 
+		SpinLockAcquire(&local->lock);
+		ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, &found);
+		Assert(!found);
+		ts->cid = is_commit ? dtm_get_cid() : AbortedGlobalCSN;
+		ts->nSubxids = nSubxids;
+		DtmTransactionListAppend(ts);
+		DtmAddSubtransactions(ts, subxids, nSubxids);
+		SpinLockRelease(&local->lock);
+	}
 }
 
 /*
@@ -669,59 +726,6 @@ DtmDeserializeTransactionState(void* ctx)
 }
 
 
-/*
- * Save state of parepared transaction
- */
-void
-DtmLocalSavePreparedState(DtmCurrentTrans * x)
-{
-
-	if (dtm_tx.is_global)
-	{
-		TransactionId *subxids;
-		TransactionId xid = GetCurrentTransactionId();
-		int			  nSubxids = xactGetCommittedChildren(&subxids);
-
-		SpinLockAcquire(&local->lock);
-		{
-			DtmTransStatus *ts;
-			bool found;
-
-			ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, &found);
-			Assert(!found);
-			ts->cid = InDoubtGlobalCSN;
-			ts->nSubxids = nSubxids;
-			DtmTransactionListAppend(ts);
-			DtmAddSubtransactions(ts, subxids, nSubxids);
-		}
-		SpinLockRelease(&local->lock);
-	}
-}
-
-/*
- * Add subtransactions to finished transactions list.
- * Copy CSN and status of parent transaction.
- */
-static void
-DtmAddSubtransactions(DtmTransStatus * ts, TransactionId *subxids, int nSubxids)
-{
-	int			i;
-
-	for (i = 0; i < nSubxids; i++)
-	{
-		bool		found;
-		DtmTransStatus *sts;
-
-		Assert(TransactionIdIsValid(subxids[i]));
-		sts = (DtmTransStatus *) hash_search(xid2status, &subxids[i], HASH_ENTER, &found);
-		Assert(!found);
-		sts->cid = ts->cid;
-		sts->nSubxids = 0;
-		DtmTransactionListInsertAfter(ts, sts);
-	}
-}
-
-
 /*
  *
  * SQL functions for global snapshot mamagement.
@@ -731,8 +735,7 @@ DtmAddSubtransactions(DtmTransStatus * ts, TransactionId *subxids, int nSubxids)
 Datum
 pg_global_snaphot_create(PG_FUNCTION_ARGS)
 {
-	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(0));
-	cid_t		cid = DtmLocalExtend(gtid);
+	cid_t		cid = DtmLocalExtend();
 
 	DTM_TRACE((stderr, "Backend %d extends transaction %u(%s) to global with cid=%lu\n", getpid(), dtm_tx.xid, gtid, cid));
 	PG_RETURN_INT64(cid);
@@ -742,10 +745,9 @@ Datum
 pg_global_snaphot_join(PG_FUNCTION_ARGS)
 {
 	cid_t		cid = PG_GETARG_INT64(0);
-	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(1));
 
 	DTM_TRACE((stderr, "Backend %d joins transaction %u(%s) with cid=%lu\n", getpid(), dtm_tx.xid, gtid, cid));
-	cid = DtmLocalAccess(&dtm_tx, gtid, cid);
+	cid = DtmLocalAccess(cid);
 	PG_RETURN_INT64(cid);
 }
 
diff --git a/src/include/access/global_snapshot.h b/src/include/access/global_snapshot.h
@@ -11,7 +11,6 @@ typedef struct
 {
 	cid_t		snapshot;
 	GlobalCSN	csn;
-	char		gtid[MAX_GTID_SIZE];
 	TransactionId xid;
 	bool		is_global;
 }	DtmCurrentTrans;
@@ -36,10 +35,10 @@ void		DtmInitialize(void);
 void		DtmLocalBegin(DtmCurrentTrans * x);
 
 /* Extend local transaction to global by assigning upper bound CSN which is returned to coordinator */
-extern cid_t		DtmLocalExtend(GlobalTransactionId gtid);
+extern cid_t		DtmLocalExtend(void);
 
 /* Function called at first access to any datanode except first one involved in distributed transaction */
-cid_t		DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t snapshot);
+cid_t		DtmLocalAccess(cid_t snapshot);
 
 /* Mark transaction as in-doubt */
 void		DtmLocalBeginPrepare(GlobalTransactionId gtid);
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
@@ -5534,9 +5534,9 @@ DATA(insert OID = 5028 ( satisfies_hash_partition PGNSP PGUID 12 1 0 2276 0 f f
 DESCR("hash partition CHECK constraint");
 
 /* global snapshot management functions */
-DATA(insert OID = 3434 ( pg_global_snaphot_create			 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 20 "25" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_create _null_ _null_ _null_ ));
+DATA(insert OID = 3434 ( pg_global_snaphot_create			 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 0 0 20 "" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_create _null_ _null_ _null_ ));
 DESCR("create global transaction snapshot");
-DATA(insert OID = 3435 ( pg_global_snaphot_join				 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 2 0 20 "20 25" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_join _null_ _null_ _null_ ));
+DATA(insert OID = 3435 ( pg_global_snaphot_join				 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 20 "20" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_join _null_ _null_ _null_ ));
 DESCR("set given global snapshot for current transaction");
 DATA(insert OID = 3436 ( pg_global_snaphot_begin_prepare	 PGNSP PGUID 12 1 0 0 0 f f f f t f v u 1 0 2278 "25" _null_ _null_ _null_ _null_ _null_ pg_global_snaphot_begin_prepare _null_ _null_ _null_ ));
 DESCR("start prepare of global transaction");
