Optimize TwoPhaseGetGXact

Author: knizhnik

contrib/mmts/arbiter.c

@@ -1051,6 +1051,7 @@ static void MtmReceiver(Datum arg)
 									MtmWakeUpBackend(ts);
 								}
 							} else { 
+								elog(WARNING, "Receive PRECOMMITTED response for aborted transaction"); // How it can happen? SHould we use assert here?
 								if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 									MtmWakeUpBackend(ts);
 								}

contrib/mmts/multimaster.c

@@ -942,6 +942,7 @@ void MtmPrecommitTransaction(char const* gid)
 			ts->csn = MtmAssignCSN();
 			MtmAdjustSubtransactions(ts);
 			MtmSend2PCMessage(ts, MSG_PRECOMMITTED);
+			SetPrepareTransactionState(ts->gid, "precommitted");
 		}
 	}
 	MtmUnlock();
@@ -1561,7 +1562,10 @@ void MtmHandleApplyError(void)
 }
 
 /**
- * Check status of all prepared transactions with coordinator at disabled node
+ * Check status of all prepared transactions with coordinator at disabled node.
+ * Actually, if node is precommitted (state == UNKNOWN) at any of nodes, then is is prepared at all nodes and so can be committed. 
+ * But if coordinator of transaction is crashed, we made a decision about transaction commit only if transaction is precommitted at ALL live nodes.
+ * The reason is that we want to avoid extra polling to obtain maximum CSN from all nodes to assign it to committed transaction.
  * Called only from MtmDisableNode in critical section.
  */
 static void MtmPollStatusOfPreparedTransactions(int disabledNodeId)
@@ -1602,9 +1606,12 @@ static void MtmDisableNode(int nodeId)
 	Mtm->nodes[nodeId-1].lastStatusChangeTime = now;
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	if (nodeId != MtmNodeId) { 
-		Mtm->nLiveNodes -= 1;
+		Mtm->nLiveNodes -= 1;		
+	}
+	if (Mtm->nLiveNodes >= Mtm->nAllNodes/2+1) {
+		/* Make decision about prepared transaction status only in quorum */
+		MtmPollStatusOfPreparedTransactions(nodeId);
 	}
-	MtmPollStatusOfPreparedTransactions(nodeId);
 } 
 
 static void MtmEnableNode(int nodeId)

contrib/mmts/multimaster.h

@@ -154,6 +154,12 @@ typedef struct
 	pgid_t         gid;    /* Global transaction identifier */
 } MtmArbiterMessage;
 
+/*
+ * Abort logical message is send by replica when error is happen while applying prepared transaction.
+ * In this case we do not have prepared transaction and can not do abort-prepared.
+ * But we have to record the fact of abort to be able to replay it in case of crash of coordinator of this transaction.
+ * We are using logical abort message with code 'A' for it
+ */
 typedef struct MtmAbortLogicalMessage
 {
 	pgid_t         gid;

contrib/mmts/pglogical_apply.c

@@ -628,7 +628,9 @@ process_remote_commit(StringInfo in)
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
 			MTM_LOG2("%d: PGLOGICAL_PRECOMMIT_PREPARED %s", MyProcPid, gid);
+			MtmBeginSession(origin_node);
 			MtmPrecommitTransaction(gid);
+			MtmEndSession(origin_node, true);
 			return;
 		}
 		case PGLOGICAL_COMMIT:
@@ -693,6 +695,7 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
+			/* MtmRollbackPreparedTransaction will set origin session itself */
 			MtmRollbackPreparedTransaction(origin_node, gid);
 			break;
 		}

src/backend/access/transam/twophase.c

@@ -183,6 +183,9 @@ static TwoPhaseStateData *TwoPhaseState;
 static GlobalTransaction MyLockedGxact = NULL;
 
 static bool twophaseExitRegistered = false;
+static TransactionId cached_xid = InvalidTransactionId;
+static GlobalTransaction cached_gxact = NULL;
+
 
 static void RecordTransactionCommitPrepared(TransactionId xid,
 								int nchildren,
@@ -438,6 +441,10 @@ MarkAsPreparing(TransactionId xid, const char *gid,
 	proc->lwWaitMode = 0;
 	proc->waitLock = NULL;
 	proc->waitProcLock = NULL;
+	
+	cached_xid = xid;
+	cached_gxact = gxact;
+		
 	for (i = 0; i < NUM_LOCK_PARTITIONS; i++)
 		SHMQueueInit(&(proc->myProcLocks[i]));
 	/* subxid data must be filled later by GXactLoadSubxactData */
@@ -825,9 +832,6 @@ TwoPhaseGetGXact(TransactionId xid)
 	GlobalTransaction result = NULL;
 	int			i;
 
-	static TransactionId cached_xid = InvalidTransactionId;
-	static GlobalTransaction cached_gxact = NULL;
-
 	/*
 	 * During a recovery, COMMIT PREPARED, or ABORT PREPARED, we'll be called
 	 * repeatedly for the same XID.  We can save work with a simple cache.