always add local node as participant

Author: kelvich

contrib/postgres_fdw/connection.c

@@ -485,35 +485,37 @@ begin_remote_xact(ConnCacheEntry *entry)
 				MemoryContextSwitchTo(oldcxt);
 
 
-				res = PQexec(entry->conn, psprintf("SELECT pg_global_snaphot_create('%s')",
-															two_phase_xact_gid));
+				// res = PQexec(entry->conn, psprintf("SELECT pg_global_snaphot_create('%s')",
+				// 											two_phase_xact_gid));
+
+				// if (PQresultStatus(res) != PGRES_TUPLES_OK)
+				// {
+				// 	pgfdw_report_error(ERROR, res, entry->conn, true, sql);
+				// }
+				// resp = PQgetvalue(res, 0, 0);
+				// if (resp == NULL || (*resp) == '\0' || sscanf(resp, "%ld", &current_global_cid) != 1)
+				// {
+				// 	pgfdw_report_error(ERROR, res, entry->conn, true, sql);
+				// }
+				// PQclear(res);
+
+
+				current_global_cid = DtmLocalExtend(two_phase_xact_gid);
+			}
+			// else
+			// {
+				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));
 
 				if (PQresultStatus(res) != PGRES_TUPLES_OK)
 				{
 					pgfdw_report_error(ERROR, res, entry->conn, true, sql);
 				}
-				resp = PQgetvalue(res, 0, 0);
-				if (resp == NULL || (*resp) == '\0' || sscanf(resp, "%ld", &current_global_cid) != 1)
-				{
-					pgfdw_report_error(ERROR, res, entry->conn, true, sql);
-				}
 				PQclear(res);
-
-
-				// current_global_cid = DtmLocalExtend(two_phase_xact_gid);
-			}
-
-			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));
-
-			if (PQresultStatus(res) != PGRES_TUPLES_OK)
-			{
-				pgfdw_report_error(ERROR, res, entry->conn, true, sql);
-			}
-			PQclear(res);
+			// }
 		}
 
 		/* A new potential participant for 2PC */
@@ -1045,22 +1047,22 @@ finalize_dtm(void)
 	{
 		char *gid = two_phase_xact_gid; // != NULL? two_phase_xact_gid : "";
 		cid_t maxCSN = 0;
-		// cid_t localCSN = 0;
+		cid_t localCSN = 0;
 
 		Assert(gid);
 
-		// DtmLocalBeginPrepare(gid);
+		DtmLocalBeginPrepare(gid);
 		BroadcastFunc(psprintf("SELECT pg_global_snaphot_begin_prepare('%s')",
 								gid));
 
 		/* Collect CSNs and choose max */
-		// localCSN = DtmLocalPrepare(gid, 0);
+		localCSN = DtmLocalPrepare(gid, 0);
 		BroadcastStmt(psprintf("SELECT pg_global_snaphot_prepare('%s', 0)",
 								 gid), PGRES_TUPLES_OK, DtmMaxCSN, &maxCSN);
-		// if (localCSN > maxCSN)
-		// 	maxCSN = localCSN;
+		if (localCSN > maxCSN)
+			maxCSN = localCSN;
 
-		// DtmLocalEndPrepare(gid, maxCSN);
+		DtmLocalEndPrepare(gid, maxCSN);
 		BroadcastFunc(psprintf("SELECT pg_global_snaphot_end_prepare('%s',"UINT64_FORMAT")",
 								gid, maxCSN));
 

src/backend/access/transam/global_snapshot.c

@@ -88,7 +88,7 @@ static HTAB *xid2status;
 static HTAB *gtid2xid;
 static DtmNodeState *local;
 static uint64 totalSleepInterrupts;
-static int	DtmVacuumDelay = 2; /* sec */
+static int	DtmVacuumDelay = 10; /* sec */
 static bool DtmRecordCommits = 0;
 
 DtmCurrentTrans dtm_tx; // XXXX: make static
@@ -489,19 +489,20 @@ DtmLocalBegin(DtmCurrentTrans * x)
  * Returns snapshot of current transaction.
  */
 cid_t
-DtmLocalExtend(DtmCurrentTrans * x, GlobalTransactionId gtid)
+DtmLocalExtend(GlobalTransactionId gtid)
 {
-	// DtmCurrentTrans *x = &dtm_tx;
+	DtmCurrentTrans *x = &dtm_tx;
 
 	if (gtid != NULL)
 	{
 		SpinLockAcquire(&local->lock);
 		{
 			DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_ENTER, NULL);
 
-			id->xid = x->xid;
+			id->xid = GetCurrentTransactionId();
 			id->nSubxids = 0;
 			id->subxids = 0;
+			x->xid = id->xid;
 		}
 		strncpy(x->gtid, gtid, MAX_GTID_SIZE);
 		SpinLockRelease(&local->lock);
@@ -681,7 +682,7 @@ DtmLocalCommit(DtmCurrentTrans * x)
 
 		ts = (DtmTransStatus *) hash_search(xid2status, &x->xid, HASH_ENTER, &found);
 		ts->status = TRANSACTION_STATUS_COMMITTED;
-		if (x->is_prepared)
+		if (found)
 		{
 			int			i;
 			DtmTransStatus *sts = ts;
@@ -755,7 +756,7 @@ DtmLocalAbort(DtmCurrentTrans * x)
 
 		Assert(TransactionIdIsValid(x->xid));
 		ts = (DtmTransStatus *) hash_search(xid2status, &x->xid, HASH_ENTER, &found);
-		if (x->is_prepared)
+		if (found)
 		{
 			Assert(found);
 			Assert(x->is_global);
@@ -902,7 +903,7 @@ Datum
 pg_global_snaphot_create(PG_FUNCTION_ARGS)
 {
 	GlobalTransactionId gtid = text_to_cstring(PG_GETARG_TEXT_PP(0));
-	cid_t		cid = DtmLocalExtend(&dtm_tx, gtid);
+	cid_t		cid = DtmLocalExtend(gtid);
 
 	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);

src/include/access/global_snapshot.h

@@ -30,7 +30,7 @@ void		DtmInitialize(void);
 void		DtmLocalBegin(DtmCurrentTrans * x);
 
 /* Extend local transaction to global by assigning upper bound CSN which is returned to coordinator */
-cid_t		DtmLocalExtend(DtmCurrentTrans * x, GlobalTransactionId gtid);
+extern cid_t		DtmLocalExtend(GlobalTransactionId gtid);
 
 /* 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);