Handle distributed readonly transactions committed without setting status in CLOG

knizhnik · kelvich · commit fcd67714690e · 2015-12-14T15:13:12.000+03:00
diff --git a/contrib/multimaster/multimaster.c b/contrib/multimaster/multimaster.c
@@ -141,6 +141,7 @@ static int   MMWorkers;
 static char* DtmHost;
 static int   DtmPort;
 static int   DtmBufferSize;
+static bool  DtmVoted;
 
 static ExecutorFinish_hook_type PreviousExecutorFinishHook = NULL;
 static void MMExecutorFinish(QueryDesc *queryDesc);
@@ -645,6 +646,7 @@ static void DtmSetTransactionStatus(TransactionId xid, int nsubxids, Transaction
 	{
 		if (TransactionIdIsValid(DtmNextXid))
 		{
+            DtmVoted = true;
 			if (status == TRANSACTION_STATUS_ABORTED || !MMIsDistributedTrans)
 			{
 				PgTransactionIdSetTreeStatus(xid, nsubxids, subxids, status, lsn);
@@ -774,6 +776,9 @@ DtmXactCallback(XactEvent event, void *arg)
     case XACT_EVENT_ABORT:
 		if (TransactionIdIsValid(DtmNextXid))
 		{
+            if (!DtmVoted) {
+                DtmGlobalSetTransStatus(DtmNextXid, TRANSACTION_STATUS_ABORTED, false);
+            }
 			if (event == XACT_EVENT_COMMIT)
 			{
 				/*
@@ -784,6 +789,7 @@ DtmXactCallback(XactEvent event, void *arg)
 				hash_search(xid_in_doubt, &DtmNextXid, HASH_REMOVE, NULL);
 				LWLockRelease(dtm->hashLock);
 			}
+#if 0 /* should be handled now using DtmVoted flag */
 			else
 			{
 				/*
@@ -796,6 +802,7 @@ DtmXactCallback(XactEvent event, void *arg)
 					DtmGlobalSetTransStatus(DtmNextXid, TRANSACTION_STATUS_ABORTED, false);
                 }
 			}
+#endif
 			DtmNextXid = InvalidTransactionId;
 			DtmLastSnapshot = NULL;
         }
@@ -1008,6 +1015,7 @@ void MMBeginTransaction(void)
 		elog(ERROR, "Arbiter was not able to assign XID");
 	XTM_INFO("%d: Start global transaction %d, dtm->minXid=%d\n", getpid(), DtmNextXid, dtm->minXid);
 
+    DtmVoted = false;
 	DtmHasGlobalSnapshot = true;
 	DtmLastSnapshot = NULL;
     MMIsDistributedTrans = false;
@@ -1020,6 +1028,7 @@ void MMJoinTransaction(TransactionId xid)
 	DtmNextXid = xid;
 	if (!TransactionIdIsValid(DtmNextXid))
 		elog(ERROR, "Arbiter was not able to assign XID");
+    DtmVoted = false;
 
 	DtmGlobalGetSnapshot(DtmNextXid, &DtmSnapshot, &dtm->minXid);
 	XTM_INFO("%d: Join global transaction %d, dtm->minXid=%d\n", getpid(), DtmNextXid, dtm->minXid);
diff --git a/contrib/multimaster/tests/mm.pgb b/contrib/multimaster/tests/mm.pgb
@@ -0,0 +1,7 @@
+\setrandom id1 1 100000
+\setrandom id2 1 100000
+begin;
+--select v from t where u=:id;
+update t set v = v - 1 where u=:id1;
+update t set v = v + 1 where u=:id2;
+commit;
diff --git a/contrib/multimaster/tests/run-pgbench.sh b/contrib/multimaster/tests/run-pgbench.sh
@@ -0,0 +1 @@
+pgbench -M prepared -f mm.pgb -T 15 -P 1 -c 50 -j 16 -r postgres

Original file line number	Diff line number	Diff line change
`@@ -141,6 +141,7 @@ static int MMWorkers;`
`141`	`141`	`static char* DtmHost;`
`142`	`142`	`static int DtmPort;`
`143`	`143`	`static int DtmBufferSize;`
	`144`	`+static bool DtmVoted;`
`144`	`145`
`145`	`146`	`static ExecutorFinish_hook_type PreviousExecutorFinishHook = NULL;`
`146`	`147`	`static void MMExecutorFinish(QueryDesc *queryDesc);`
`@@ -645,6 +646,7 @@ static void DtmSetTransactionStatus(TransactionId xid, int nsubxids, Transaction`
`645`	`646`	`{`
`646`	`647`	`if (TransactionIdIsValid(DtmNextXid))`
`647`	`648`	`{`
	`649`	`+ DtmVoted = true;`
`648`	`650`	`if (status == TRANSACTION_STATUS_ABORTED \|\| !MMIsDistributedTrans)`
`649`	`651`	`{`
`650`	`652`	`PgTransactionIdSetTreeStatus(xid, nsubxids, subxids, status, lsn);`
`@@ -774,6 +776,9 @@ DtmXactCallback(XactEvent event, void *arg)`
`774`	`776`	`case XACT_EVENT_ABORT:`
`775`	`777`	`if (TransactionIdIsValid(DtmNextXid))`
`776`	`778`	`{`
	`779`	`+ if (!DtmVoted) {`
	`780`	`+ DtmGlobalSetTransStatus(DtmNextXid, TRANSACTION_STATUS_ABORTED, false);`
	`781`	`+ }`
`777`	`782`	`if (event == XACT_EVENT_COMMIT)`
`778`	`783`	`{`
`779`	`784`	`/*`
`@@ -784,6 +789,7 @@ DtmXactCallback(XactEvent event, void *arg)`
`784`	`789`	`hash_search(xid_in_doubt, &DtmNextXid, HASH_REMOVE, NULL);`
`785`	`790`	`LWLockRelease(dtm->hashLock);`
`786`	`791`	`}`
	`792`	`+#if 0 /* should be handled now using DtmVoted flag */`
`787`	`793`	`else`
`788`	`794`	`{`
`789`	`795`	`/*`
`@@ -796,6 +802,7 @@ DtmXactCallback(XactEvent event, void *arg)`
`796`	`802`	`DtmGlobalSetTransStatus(DtmNextXid, TRANSACTION_STATUS_ABORTED, false);`
`797`	`803`	`}`
`798`	`804`	`}`
	`805`	`+#endif`
`799`	`806`	`DtmNextXid = InvalidTransactionId;`
`800`	`807`	`DtmLastSnapshot = NULL;`
`801`	`808`	`}`
`@@ -1008,6 +1015,7 @@ void MMBeginTransaction(void)`
`1008`	`1015`	`elog(ERROR, "Arbiter was not able to assign XID");`
`1009`	`1016`	`XTM_INFO("%d: Start global transaction %d, dtm->minXid=%d\n", getpid(), DtmNextXid, dtm->minXid);`
`1010`	`1017`
	`1018`	`+ DtmVoted = false;`
`1011`	`1019`	`DtmHasGlobalSnapshot = true;`
`1012`	`1020`	`DtmLastSnapshot = NULL;`
`1013`	`1021`	`MMIsDistributedTrans = false;`
`@@ -1020,6 +1028,7 @@ void MMJoinTransaction(TransactionId xid)`
`1020`	`1028`	`DtmNextXid = xid;`
`1021`	`1029`	`if (!TransactionIdIsValid(DtmNextXid))`
`1022`	`1030`	`elog(ERROR, "Arbiter was not able to assign XID");`
	`1031`	`+ DtmVoted = false;`
`1023`	`1032`
`1024`	`1033`	`DtmGlobalGetSnapshot(DtmNextXid, &DtmSnapshot, &dtm->minXid);`
`1025`	`1034`	`XTM_INFO("%d: Join global transaction %d, dtm->minXid=%d\n", getpid(), DtmNextXid, dtm->minXid);`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+pgbench -M prepared -f mm.pgb -T 15 -P 1 -c 50 -j 16 -r postgres`