Skip to content

Commit 258b5f8

Browse files
knizhnikknizhnik
committed
Do not set XMAX_INVALID for transaction which are in progress
1 parent 2151bff commit 258b5f8

File tree

3 files changed

+23
-14
lines changed

3 files changed

+23
-14
lines changed

‎contrib/mmts/multimaster.c

Lines changed: 16 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@
7272
typedef struct {
7373
TransactionId xid; /* local transaction ID */
7474
GlobalTransactionId gtid; /* global transaction ID assigned by coordinator of transaction */
75+
bool isTwoPhase; /* user level 2PC */
7576
bool isReplicated; /* transaction on replica */
7677
bool isDistributed; /* transaction performed INSERT/UPDATE/DELETE and has to be replicated to other nodes */
7778
bool isPrepared; /* transaction is perpared at first stage of 2PC */
@@ -719,6 +720,7 @@ MtmResetTransaction()
719720
x->gtid.xid = InvalidTransactionId;
720721
x->isDistributed = false;
721722
x->isPrepared = false;
723+
x->isTwoPhase = false;
722724
x->status = TRANSACTION_STATUS_UNKNOWN;
723725
}
724726

@@ -746,6 +748,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
746748
x->isReplicated = MtmIsLogicalReceiver;
747749
x->isDistributed = MtmIsUserTransaction();
748750
x->isPrepared = false;
751+
x->isTwoPhase = false;
749752
x->isTransactionBlock = IsTransactionBlock();
750753
/* Application name can be changed usnig PGAPPNAME environment variable */
751754
if (x->isDistributed && Mtm->status != MTM_ONLINE && strcmp(application_name, MULTIMASTER_ADMIN) != 0) {
@@ -906,8 +909,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
906909
Assert(ts != NULL);
907910
//if (x->gid[0]) MTM_LOG1("Preparing transaction %d (%s) at %ld", x->xid, x->gid, MtmGetCurrentTime());
908911
if (!MtmIsCoordinator(ts) || Mtm->status == MTM_RECOVERY) {
909-
bool found;
910-
MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, &found);
912+
MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
911913
Assert(x->gid[0]);
912914
tm->state = ts;
913915
ts->votingCompleted = true;
@@ -925,8 +927,13 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
925927
time_t transTimeout = Max(MSEC_TO_USEC(Mtm2PCMinTimeout), (ts->csn - ts->snapshot)*Mtm2PCPrepareRatio/100);
926928
int result = 0;
927929
int nConfigChanges = Mtm->nConfigChanges;
928-
929930
timestamp_t start = MtmGetSystemTime();
931+
932+
if (x->isTwoPhase) {
933+
MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
934+
tm->state = ts;
935+
}
936+
930937
/* Wait votes from all nodes until: */
931938
while (!ts->votingCompleted /* all nodes voted */
932939
&& nConfigChanges == Mtm->nConfigChanges /* configarion is changed */
@@ -982,7 +989,7 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
982989
MtmLock(LW_EXCLUSIVE);
983990
tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
984991
if (tm == NULL) {
985-
elog(WARNING, "Global transaciton ID %s is not found", x->gid);
992+
elog(WARNING, "Global transaciton ID '%s' is not found", x->gid);
986993
} else {
987994
Assert(tm->state != NULL);
988995
MTM_LOG1("Abort prepared transaction %d with gid='%s'", x->xid, x->gid);
@@ -1265,7 +1272,7 @@ void MtmAbortTransaction(MtmTransState* ts)
12651272
Assert(MtmLockCount != 0); /* should be invoked with exclsuive lock */
12661273
if (ts->status != TRANSACTION_STATUS_ABORTED) {
12671274
if (ts->status == TRANSACTION_STATUS_COMMITTED) {
1268-
elog(WARNING, "Attempt to rollback already committed transaction %d (%s)", ts->xid, ts->gid);
1275+
elog(LOG, "Attempt to rollback already committed transaction %d (%s)", ts->xid, ts->gid);
12691276
} else {
12701277
MTM_LOG1("Rollback active transaction %d:%d (local xid %d) status %d", ts->gtid.node, ts->gtid.xid, ts->xid, ts->status);
12711278
ts->status = TRANSACTION_STATUS_ABORTED;
@@ -3803,11 +3810,10 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
38033810
}
38043811
break;
38053812
case TRANS_STMT_PREPARE:
3806-
//elog(ERROR, "Two phase commit is not supported by multimaster");
3807-
break;
38083813
case TRANS_STMT_COMMIT_PREPARED:
38093814
case TRANS_STMT_ROLLBACK_PREPARED:
3810-
skipCommand = true;
3815+
MtmTx.isTwoPhase = true;
3816+
strcpy(MtmTx.gid, stmt->gid);
38113817
break;
38123818
default:
38133819
break;
@@ -3940,7 +3946,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
39403946
params, dest, completionTag);
39413947
}
39423948

3943-
if (MtmTx.isDistributed && XactIsoLevel != XACT_REPEATABLE_READ) {
3949+
if (!MtmVolksWagenMode && MtmTx.isDistributed && XactIsoLevel != XACT_REPEATABLE_READ) {
39443950
elog(ERROR, "Isolation level %s is not supported by multimaster", isoLevelStr[XactIsoLevel]);
39453951
}
39463952

@@ -4126,7 +4132,7 @@ MtmDetectGlobalDeadLockFortXid(TransactionId xid)
41264132
}
41274133
MtmGetGtid(xid, &gtid);
41284134
hasDeadlock = MtmGraphFindLoop(&graph, &gtid);
4129-
elog(WARNING, "Distributed deadlock check by backend %d for %u:%u = %d", MyProcPid, gtid.node, gtid.xid, hasDeadlock);
4135+
elog(LOG, "Distributed deadlock check by backend %d for %u:%u = %d", MyProcPid, gtid.node, gtid.xid, hasDeadlock);
41304136
if (!hasDeadlock) {
41314137
/* There is no deadlock loop in graph, but deadlock can be caused by lack of apply workers: if all of them are busy, then some transactions
41324138
* can not be appied just because there are no vacant workers and it cause additional dependency between transactions which is not

‎src/backend/storage/lmgr/predicate.c

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3901,6 +3901,7 @@ CheckForSerializableConflictOut(bool visible, Relation relation,
39013901
* is going on with it.
39023902
*/
39033903
htsvResult = HeapTupleSatisfiesVacuum(tuple, TransactionXmin, buffer);
3904+
elog(LOG, "HeapTupleSatisfiesVacuum(%d) = %d, TransactionXmin=%d", HeapTupleHeaderGetXmin(tuple->t_data), htsvResult, TransactionXmin);
39043905
switch (htsvResult)
39053906
{
39063907
case HEAPTUPLE_LIVE:
@@ -3939,7 +3940,7 @@ CheckForSerializableConflictOut(bool visible, Relation relation,
39393940
xid = InvalidTransactionId;
39403941
}
39413942
Assert(TransactionIdIsValid(xid));
3942-
Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
3943+
//Assert(TransactionIdFollowsOrEquals(xid, TransactionXmin));
39433944

39443945
/*
39453946
* Find top level xid. Bail out if xid is too early to be a conflict, or

‎src/backend/utils/time/tqual.c

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1131,8 +1131,9 @@ HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot,
11311131
if (!TransactionIdDidCommit(HeapTupleHeaderGetRawXmax(tuple)))
11321132
{
11331133
/* it must have aborted or crashed */
1134-
SetHintBits(tuple, buffer, HEAP_XMAX_INVALID,
1135-
InvalidTransactionId);
1134+
if (!TransactionIdIsInProgress(HeapTupleHeaderGetRawXmax(tuple)))
1135+
SetHintBits(tuple, buffer, HEAP_XMAX_INVALID,
1136+
InvalidTransactionId);
11361137
return true;
11371138
}
11381139

@@ -1271,8 +1272,9 @@ HeapTupleSatisfiesVacuum(HeapTuple htup, TransactionId OldestXmin,
12711272
* Okay, the inserter committed, so it was good at some point. Now what
12721273
* about the deleting transaction?
12731274
*/
1274-
if (tuple->t_infomask & HEAP_XMAX_INVALID)
1275+
if (tuple->t_infomask & HEAP_XMAX_INVALID) {
12751276
return HEAPTUPLE_LIVE;
1277+
}
12761278

12771279
if (HEAP_XMAX_IS_LOCKED_ONLY(tuple->t_infomask))
12781280
{

0 commit comments

Comments
 (0)