Support users 2PC

Author: knizhnik

arbiter.c

@@ -997,15 +997,18 @@ static void MtmReceiver(Datum arg)
 									/* All nodes are finished their transactions */
 									if (ts->status == TRANSACTION_STATUS_ABORTED) { 
 										MtmWakeUpBackend(ts);								
-									} else if (MtmUseDtm) { 
-										Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-										ts->nVotes = 1; /* I voted myself */
-										MTM_TXTRACE(ts, "MtmTransReceiver send MSG_PREPARE");
-										MtmSend2PCMessage(ts, MSG_PREPARE);									  
 									} else { 
 										Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-										ts->status = TRANSACTION_STATUS_UNKNOWN;
-										MtmWakeUpBackend(ts);
+										if (ts->isTwoPhase) { 
+											MtmWakeUpBackend(ts);										
+										} else if (MtmUseDtm) { 
+											ts->nVotes = 1; /* I voted myself */
+											MTM_TXTRACE(ts, "MtmTransReceiver send MSG_PREPARE");
+											MtmSend2PCMessage(ts, MSG_PREPARE);									  
+										} else { 
+											ts->status = TRANSACTION_STATUS_UNKNOWN;
+											MtmWakeUpBackend(ts);
+										}
 									}
 								}
 							}

multimaster.c

@@ -129,6 +129,7 @@ static void MtmBeginTransaction(MtmCurrentTrans* x);
 static void MtmPrePrepareTransaction(MtmCurrentTrans* x);
 static void MtmPostPrepareTransaction(MtmCurrentTrans* x);
 static void MtmAbortPreparedTransaction(MtmCurrentTrans* x);
+static void MtmCommitPreparedTransaction(MtmCurrentTrans* x);
 static void MtmEndTransaction(MtmCurrentTrans* x, bool commit);
 static bool MtmTwoPhaseCommit(MtmCurrentTrans* x);
 static TransactionId MtmGetOldestXmin(Relation rel, bool ignoreVacuum);
@@ -684,6 +685,9 @@ MtmXactCallback(XactEvent event, void *arg)
 	  case XACT_EVENT_ABORT_PREPARED:
 		MtmAbortPreparedTransaction(&MtmTx);
 		break;
+	  case XACT_EVENT_COMMIT_PREPARED:
+		MtmCommitPreparedTransaction(&MtmTx);
+		break;
 	  case XACT_EVENT_COMMIT:
 		MtmEndTransaction(&MtmTx, true);
 		break;
@@ -792,6 +796,7 @@ MtmCreateTransState(MtmCurrentTrans* x)
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;
 	ts->snapshot = x->snapshot;
 	ts->isLocal = true;
+	ts->isTwoPhase = x->isTwoPhase;
 	if (!found) {
 		ts->isEnqueued = false;
 		ts->isActive = false;
@@ -969,6 +974,9 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		x->status = ts->status;
 		MTM_LOG3("%d: Result of vote: %d", MyProcPid, ts->status);
 		MtmUnlock();
+		if (x->isTwoPhase) { 
+			MtmResetTransaction();
+		}
 	}
 	//if (x->gid[0]) MTM_LOG1("Prepared transaction %d (%s) csn=%ld at %ld: %d", x->xid, x->gid, ts->csn, MtmGetCurrentTime(), ts->status);
 	if (Mtm->inject2PCError == 3) { 
@@ -979,6 +987,74 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 	MTM_TXTRACE(x, "PostPrepareTransaction Finish");
 }
 
+static void 
+MtmCommitPreparedTransaction(MtmCurrentTrans* x)
+{
+    MtmTransMap* tm;
+	MtmTransState* ts;
+
+    if (Mtm->status == MTM_RECOVERY || x->isReplicated || x->isPrepared) { /* Ignore auto-2PC originated by multimaster */ 
+        return;
+    }
+	MtmLock(LW_EXCLUSIVE);
+	tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_FIND, NULL);
+	if (tm == NULL) {
+		elog(WARNING, "Global transaciton ID '%s' is not found", x->gid);
+	} else {
+        time_t transTimeout = MSEC_TO_USEC(Mtm2PCMinTimeout);
+        int nConfigChanges = Mtm->nConfigChanges;  
+		timestamp_t start = MtmGetSystemTime();
+		int result = 0;
+
+ 		Assert(tm->state != NULL);
+		MTM_LOG1("Commit prepared transaction %d with gid='%s'", x->xid, x->gid);
+		ts = tm->state;
+
+		Assert(MtmIsCoordinator(ts));
+
+		ts->votingCompleted = false;
+		ts->nVotes = 1; /* I voted myself */
+		ts->procno = MyProc->pgprocno;
+		MTM_TXTRACE(ts, "Coordinator sends MSG_PREPARE");
+		MtmSend2PCMessage(ts, MSG_PREPARE);
+		
+		/* Wait votes from all nodes until: */
+		while (!ts->votingCompleted                           /* all nodes voted */
+			   && nConfigChanges == Mtm->nConfigChanges       /* configarion is changed */
+			   && Mtm->status == MTM_ONLINE                   /* node is not online */
+			   && ts->status != TRANSACTION_STATUS_ABORTED    /* transaction is aborted */
+			   && start + transTimeout >= MtmGetSystemTime()) /* timeout is expired */
+		{
+			MtmUnlock();
+			MTM_TXTRACE(x, "CommitPreparedTransaction WaitLatch Start");
+			result = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, MtmHeartbeatRecvTimeout);
+			MTM_TXTRACE(x, "CommitPreparedTransaction WaitLatch Finish");
+			/* Emergency bailout if postmaster has died */
+			if (result & WL_POSTMASTER_DEATH) {
+				proc_exit(1);
+			}
+			if (result & WL_LATCH_SET) {
+				MTM_LOG3("Latch signaled at %ld", MtmGetSystemTime());
+				ResetLatch(&MyProc->procLatch);
+			}
+			MtmLock(LW_EXCLUSIVE);
+		}
+		if (ts->status != TRANSACTION_STATUS_ABORTED && (!ts->votingCompleted || nConfigChanges != Mtm->nConfigChanges)) {
+			if (nConfigChanges != Mtm->nConfigChanges) {
+				elog(WARNING, "Transaction %d (%s) is aborted because cluster configuration is changed during commit", x->xid, x->gid);
+			} else {
+				elog(WARNING, "Transaction %d (%s) is aborted because of %d msec timeout expiration, prepare time %d msec",
+					 x->xid, x->gid, (int)USEC_TO_MSEC(transTimeout), (int)USEC_TO_MSEC(ts->csn - x->snapshot));
+			}
+			MtmAbortTransaction(ts);
+		}
+		x->status = ts->status;
+		x->xid = ts->xid;
+		x->isPrepared = true;
+		MTM_LOG3("%d: Result of vote: %d", MyProcPid, ts->status);
+	}
+	MtmUnlock();
+}
 
 static void 
 MtmAbortPreparedTransaction(MtmCurrentTrans* x)
@@ -1008,9 +1084,9 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 static void 
 MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 {
-	MTM_LOG3("%d: End transaction %d, prepared=%d, replicated=%d, distributed=%d, gid=%s -> %s", 
-			 MyProcPid, x->xid, x->isPrepared, x->isReplicated, x->isDistributed, x->gid, commit ? "commit" : "abort");
-	if (x->status != TRANSACTION_STATUS_ABORTED && x->isDistributed && (x->isPrepared || x->isReplicated)) {
+	MTM_LOG1("%d: End transaction %d, prepared=%d, replicated=%d, distributed=%d, 2pc=%d, gid=%s -> %s", 
+			 MyProcPid, x->xid, x->isPrepared, x->isReplicated, x->isDistributed, x->isTwoPhase, x->gid, commit ? "commit" : "abort");
+	if (x->status != TRANSACTION_STATUS_ABORTED && x->isDistributed && (x->isPrepared || x->isReplicated) && !x->isTwoPhase) {
 		MtmTransState* ts = NULL;
 		MtmLock(LW_EXCLUSIVE);
 		if (x->isPrepared) { 
@@ -3819,9 +3895,13 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 					}
 					break;
 				case TRANS_STMT_PREPARE:
+  				    MtmTx.isTwoPhase = true;
+  				    strcpy(MtmTx.gid, stmt->gid);
+					break;
+					/* nobreak */
 				case TRANS_STMT_COMMIT_PREPARED:
 				case TRANS_STMT_ROLLBACK_PREPARED:
-  				    MtmTx.isTwoPhase = true;
+				    Assert(!MtmTx.isTwoPhase);
   				    strcpy(MtmTx.gid, stmt->gid);
 					break;
 				default:

multimaster.h

@@ -213,6 +213,7 @@ typedef struct MtmTransState
 	bool           isLocal;            /* Transaction is either replicated, either doesn't contain DML statements, so it shoudl be ignored by pglogical replication */
 	bool           isEnqueued;         /* Transaction is inserted in queue */
 	bool           isActive;           /* Transaction is active */
+	bool           isTwoPhase;         /* user level 2PC */
 	nodemask_t     participantsMask;   /* Mask of nodes involved in transaction */
 	nodemask_t     votedMask;          /* Mask of voted nodes */
 	TransactionId  xids[1];            /* [Mtm->nAllNodes]: transaction ID at replicas */

pglogical_proto.c

@@ -186,8 +186,6 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	else
     	Assert(false);
 
-	Assert(flags != PGLOGICAL_COMMIT_PREPARED || txn->xid < 1000 || MtmTransactionRecords != 1);
-
 	if (flags == PGLOGICAL_COMMIT || flags == PGLOGICAL_PREPARE) { 
 		// if (MtmIsFilteredTxn) { 
 			// Assert(MtmTransactionRecords == 0);