get rid of gtid2xid hash

Author: kelvich

contrib/postgres_fdw/connection.c

@@ -832,7 +832,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
 		}
 		else /* Otherwise, just finalize dtm and proceed below to COMMIT */
 		{
-			finalize_dtm();
+			// finalize_dtm();
 		}
 	}
 

src/backend/access/transam/global_snapshot.c

@@ -85,7 +85,6 @@ typedef struct
 
 // static shmem_startup_hook_type prev_shmem_startup_hook;
 static HTAB *xid2status;
-static HTAB *gtid2xid;
 static DtmNodeState *local;
 static uint64 totalSleepInterrupts;
 static int	DtmVacuumDelay = 15; /* sec */
@@ -432,13 +431,6 @@ DtmInitialize()
 							   &info,
 							   HASH_ELEM | HASH_BLOBS);
 
-	info.keysize = MAX_GTID_SIZE;
-	info.entrysize = sizeof(DtmTransId);
-	gtid2xid = ShmemInitHash("gtid2xid",
-							 DTM_HASH_INIT_SIZE, DTM_HASH_INIT_SIZE,
-							 &info,
-							 HASH_ELEM);
-
 	TM = &DtmTM;
 
 	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
@@ -478,16 +470,7 @@ DtmLocalExtend(GlobalTransactionId gtid)
 
 	if (gtid != NULL)
 	{
-		SpinLockAcquire(&local->lock);
-		{
-			DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_ENTER, NULL);
-
-			id->xid = GetCurrentTransactionId();
-			id->nSubxids = 0;
-			id->subxids = 0;
-		}
 		strncpy(x->gtid, gtid, MAX_GTID_SIZE);
-		SpinLockRelease(&local->lock);
 	}
 	DtmInitGlobalXmin(TransactionXmin);
 	return x->snapshot;
@@ -506,14 +489,6 @@ DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t global_cid)
 
 	SpinLockAcquire(&local->lock);
 	{
-		if (gtid != NULL)
-		{
-			DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_ENTER, NULL);
-
-			id->xid = GetCurrentTransactionId();
-			id->nSubxids = 0;
-			id->subxids = 0;
-		}
 		local_cid = dtm_sync(global_cid);
 		x->snapshot = global_cid;
 	}
@@ -536,18 +511,20 @@ DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t global_cid)
 void
 DtmLocalBeginPrepare(GlobalTransactionId gtid)
 {
-	// TransactionId xid = TwoPhaseGetTransactionId(gtid);
+	TransactionId xid = TwoPhaseGetTransactionId(gtid);
+
+	if (!TransactionIdIsValid(xid))
+	{
+		// XXX: check that it is global tx with the same xid, XactTopTransactionId?
+		xid = GetCurrentTransactionId();
+	}
 
 	SpinLockAcquire(&local->lock);
 	{
 		DtmTransStatus *ts;
-		DtmTransId *id;
 		bool found;
 
-		id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_FIND, NULL);
-		Assert(id != NULL);
-		Assert(TransactionIdIsValid(id->xid));
-		ts = (DtmTransStatus *) hash_search(xid2status, &id->xid, HASH_ENTER, &found);
+		ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, &found);
 		ts->status = TRANSACTION_STATUS_UNKNOWN;
 		ts->cid = dtm_get_cid();
 		if (!found)
@@ -582,15 +559,23 @@ DtmLocalPrepare(GlobalTransactionId gtid, cid_t global_cid)
 void
 DtmLocalEndPrepare(GlobalTransactionId gtid, cid_t cid)
 {
+	TransactionId xid = TwoPhaseGetTransactionId(gtid);
+
+	if (!TransactionIdIsValid(xid))
+	{
+		// XXX: check that it is global tx with the same xid, XactTopTransactionId?
+		xid = GetCurrentTransactionId();
+	}
+
+	dtm_tx.xid = xid;
+
 	SpinLockAcquire(&local->lock);
 	{
 		DtmTransStatus *ts;
 		DtmTransId *id;
 		int			i;
 
-		id = (DtmTransId *) hash_search(gtid2xid, gtid, HASH_FIND, NULL);
-
-		ts = (DtmTransStatus *) hash_search(xid2status, &id->xid, HASH_FIND, NULL);
+		ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_FIND, NULL);
 		Assert(ts != NULL);
 		ts->cid = cid;
 		DtmAdjustSubtransactions(ts);
@@ -613,19 +598,7 @@ DtmLocalFinish(bool is_commit)
 
 	if (x->gtid[0] && finishing_prepared)
 	{
-		// Assert(!TransactionIdIsValid(xid));
-
-		SpinLockAcquire(&local->lock);
-		{
-			DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, x->gtid, HASH_REMOVE, NULL);
-
-			Assert(id != NULL);
-			Assert(TransactionIdIsValid(id->xid));
-
-			xid = id->xid;
-			free(id->subxids);
-		}
-		SpinLockRelease(&local->lock);
+		xid = x->xid;
 	}
 	else if (!TransactionIdIsValid(xid))
 	{
@@ -716,20 +689,6 @@ DtmLocalSavePreparedState(DtmCurrentTrans * x)
 		int			  nSubxids = xactGetCommittedChildren(&subxids);
 
 		SpinLockAcquire(&local->lock);
-		{
-			DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, x->gtid, HASH_FIND, NULL);
-
-			if (id != NULL)
-			{
-				id->xid = GetCurrentTransactionId();
-
-			}
-		}
-		// SpinLockRelease(&local->lock);
-
-
-
-		// SpinLockAcquire(&local->lock);
 		{
 			DtmTransStatus *ts;
 

src/backend/access/transam/twophase.c

@@ -872,6 +872,31 @@ TwoPhaseGetDummyProc(TransactionId xid)
 	return &ProcGlobal->allProcs[gxact->pgprocno];
 }
 
+TransactionId
+TwoPhaseGetTransactionId(const char *gid)
+{
+	int i;
+
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+
+		/* Ignore not-yet-valid GIDs */
+		if (!gxact->valid)
+			continue;
+
+		if (strcmp(gxact->gid, gid) == 0)
+		{
+			LWLockRelease(TwoPhaseStateLock);
+			return gxact->xid;
+		}
+	}
+	LWLockRelease(TwoPhaseStateLock);
+
+	return InvalidTransactionId;
+}
+
 /************************************************************************/
 /* State file support													*/
 /************************************************************************/

src/include/access/global_snapshot.h

@@ -10,6 +10,7 @@ typedef struct
 {
 	cid_t		snapshot;
 	char		gtid[MAX_GTID_SIZE];
+	TransactionId xid;
 }	DtmCurrentTrans;
 
 typedef char const *GlobalTransactionId;

src/include/access/twophase.h

@@ -35,6 +35,7 @@ extern void PostPrepare_Twophase(void);
 
 extern PGPROC *TwoPhaseGetDummyProc(TransactionId xid);
 extern BackendId TwoPhaseGetDummyBackendId(TransactionId xid);
+extern TransactionId TwoPhaseGetTransactionId(const char *gid);
 
 extern GlobalTransaction MarkAsPreparing(TransactionId xid, const char *gid,
 				TimestampTz prepared_at,