DtmAdjustSubtransactions

Author: kelvich

contrib/postgres_fdw/t/001_bank_coordinator.pl

@@ -217,13 +217,14 @@ sub count_and_delete_rows
 		($hash1, $_, $hash2) = split "\n", $node->safe_psql('postgres', qq[
 			begin isolation level repeatable read;
 			select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;
-			select pg_sleep(1);
+			select pg_sleep(3);
 			select md5(array_agg((t.*)::text)::text) from (select * from accounts order by id) as t;
 			commit;
 		]);
 
 		if ($hash1 ne $hash2)
 		{
+			diag("oops");
 			$stability_errors++;
 		}
 		elsif ($hash1 eq '' or $hash2 eq '')

…ntrib/postgres_fdw/t/0003_bank_shared.pl contrib/postgres_fdw/t/003_bank_shared.plcontrib/postgres_fdw/t/0003_bank_shared.pl renamed to contrib/postgres_fdw/t/003_bank_shared.pl contrib/postgres_fdw/t/0003_bank_shared.pl renamed to contrib/postgres_fdw/t/003_bank_shared.pl

@@ -88,7 +88,7 @@ static HTAB *xid2status;
 static HTAB *gtid2xid;
 static DtmNodeState *local;
 static uint64 totalSleepInterrupts;
-static int	DtmVacuumDelay = 10; /* sec */
+static int	DtmVacuumDelay = 15; /* sec */
 static bool finishing_prepared;
 
 
@@ -99,6 +99,7 @@ static void DtmAdjustOldestXid(void);
 static void DtmInitGlobalXmin(TransactionId xid);
 static bool DtmDetectGlobalDeadLock(PGPROC *proc);
 static void DtmAddSubtransactions(DtmTransStatus * ts, TransactionId *subxids, int nSubxids);
+static void DtmAdjustSubtransactions(DtmTransStatus *ts);
 static char const *DtmGetName(void);
 static size_t DtmGetTransactionStateSize(void);
 static void DtmSerializeTransactionState(void* ctx);
@@ -283,6 +284,20 @@ DtmTransactionListInsertAfter(DtmTransStatus * after, DtmTransStatus * ts)
 	}
 }
 
+static void 
+DtmAdjustSubtransactions(DtmTransStatus *ts)
+{
+	int i;
+	int nSubxids = ts->nSubxids;
+	DtmTransStatus* sts = ts;
+
+	for (i = 0; i < nSubxids; i++) {
+		sts = sts->next;
+		sts->status = ts->status;
+		Assert(sts->cid == ts->cid);
+	}
+}
+
 /*
  * There can be different oldest XIDs at different cluster node.
  * Seince we do not have centralized aribiter, we have to rely in DtmVacuumDelay.
@@ -521,20 +536,23 @@ DtmLocalAccess(DtmCurrentTrans * x, GlobalTransactionId gtid, cid_t global_cid)
 void
 DtmLocalBeginPrepare(GlobalTransactionId gtid)
 {
+	// TransactionId xid = TwoPhaseGetTransactionId(gtid);
+
 	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, NULL);
+		ts = (DtmTransStatus *) hash_search(xid2status, &id->xid, HASH_ENTER, &found);
 		ts->status = TRANSACTION_STATUS_UNKNOWN;
 		ts->cid = dtm_get_cid();
-		ts->nSubxids = id->nSubxids;
-		DtmTransactionListAppend(ts);
-		DtmAddSubtransactions(ts, id->subxids, id->nSubxids);
+		if (!found)
+			ts->nSubxids = 0;
+		DtmAdjustSubtransactions(ts);
 	}
 	SpinLockRelease(&local->lock);
 }
@@ -575,11 +593,7 @@ DtmLocalEndPrepare(GlobalTransactionId gtid, cid_t cid)
 		ts = (DtmTransStatus *) hash_search(xid2status, &id->xid, HASH_FIND, NULL);
 		Assert(ts != NULL);
 		ts->cid = cid;
-		for (i = 0; i < ts->nSubxids; i++)
-		{
-			ts = ts->next;
-			ts->cid = cid;
-		}
+		DtmAdjustSubtransactions(ts);
 		dtm_sync(cid);
 
 		DTM_TRACE((stderr, "Prepare transaction %u(%s) with CSN %lu\n", id->xid, gtid, cid));
@@ -625,39 +639,14 @@ DtmLocalFinish(bool is_commit)
 
 		ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, &found);
 		ts->status = is_commit ? TRANSACTION_STATUS_COMMITTED : TRANSACTION_STATUS_ABORTED;
-		if (found)
-		{
 
-			if (is_commit) // XXX: why only for commit?
-			{
-				int			i;
-				DtmTransStatus *sts = ts;
-
-				for (i = 0; i < ts->nSubxids; i++)
-				{
-					sts = sts->next;
-					Assert(sts->cid == ts->cid);
-					sts->status = TRANSACTION_STATUS_COMMITTED;
-				}
-			}
-		}
-		else
+		if (!found)
 		{
-			TransactionId *subxids;
-
-			Assert(!found);
 			ts->cid = dtm_get_cid();
+			ts->nSubxids = 0;
 			DtmTransactionListAppend(ts);
-			if (is_commit) // XXX: why?
-			{
-				ts->nSubxids = xactGetCommittedChildren(&subxids);
-				DtmAddSubtransactions(ts, subxids, ts->nSubxids);
-			}
-			else
-			{
-				ts->nSubxids = 0;
-			}
 		}
+		DtmAdjustSubtransactions(ts);
 	}
 	SpinLockRelease(&local->lock);
 
@@ -722,24 +711,35 @@ DtmLocalSavePreparedState(DtmCurrentTrans * x)
 
 	if (x->gtid[0])
 	{
+		TransactionId *subxids;
+		TransactionId xid = GetCurrentTransactionId();
+		int			  nSubxids = xactGetCommittedChildren(&subxids);
+
 		SpinLockAcquire(&local->lock);
 		{
 			DtmTransId *id = (DtmTransId *) hash_search(gtid2xid, x->gtid, HASH_FIND, NULL);
 
 			if (id != NULL)
 			{
-				TransactionId *subxids;
-				int			nSubxids = xactGetCommittedChildren(&subxids);
-
 				id->xid = GetCurrentTransactionId();
-				if (nSubxids != 0)
-				{
-					id->subxids = (TransactionId *) malloc(nSubxids * sizeof(TransactionId));
-					id->nSubxids = nSubxids;
-					memcpy(id->subxids, subxids, nSubxids * sizeof(TransactionId));
-				}
+
 			}
 		}
+		// SpinLockRelease(&local->lock);
+
+
+
+		// SpinLockAcquire(&local->lock);
+		{
+			DtmTransStatus *ts;
+
+			ts = (DtmTransStatus *) hash_search(xid2status, &xid, HASH_ENTER, NULL);
+			ts->status = TRANSACTION_STATUS_IN_PROGRESS;
+			ts->cid = dtm_get_cid();
+			ts->nSubxids = nSubxids;
+			DtmTransactionListAppend(ts);
+			DtmAddSubtransactions(ts, subxids, nSubxids);
+		}
 		SpinLockRelease(&local->lock);
 	}
 }