Rewrite 2pc support

Author: knizhnik

contrib/mmts/multimaster.c

@@ -126,7 +126,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError);
 static void MtmVoteForTransaction(MtmTransState* ts);
 
 static HTAB* xid2state;
-static HTAB* git2xid;
+static HTAB* gid2xid;
 static MtmCurrentTrans dtmTx;
 static MtmState* dtm;
 
@@ -529,7 +529,6 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 			elog(ERROR, "Multimaster node is not online: current status %s", MtmNodeStatusMnem[dtm->status]);
 		}
 		x->containsDML = false;
-		x->isPrepared = false;
         x->snapshot = MtmAssignCSN();	
 		x->gtid.xid = InvalidTransactionId;
 		x->gid[0] = '\0';
@@ -541,7 +540,8 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 }
 
 /* 
- * Prepare transaction for two-phase commit
+ * Prepare transaction for two-phase commit.
+ * This code is executed by PRE_PREPARE hook before PREPARE message is sent to replicas by logical replication
  */
 MtmPrePrepareTransaction(MtmCurrentTrans* x)
 { 
@@ -575,7 +575,6 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	ts->gtid = x->gtid;
 	ts->procno = MyProc->pgprocno;
 	ts->nVotes = 0; 
-	ts->voteCompleted = false;
 
 	x->isPrepared = true;
 	x->csn = csn;

contrib/mmts/multimaster.h

@@ -84,11 +84,9 @@ typedef struct MtmTransState
 							              finally should be nNodes-1 */
 	int            procno;             /* pgprocno of transaction coordinator waiting for responses from replicas, 
 							              used to notify coordinator by arbiter */
-	bool           voteCompleted;      /* Responses necessary to make a decision are received by coordinator of transaction */
 	int            nSubxids;           /* Number of subtransanctions */
 	struct MtmTransState* nextVoting;  /* Next element in L1-list of voting transactions. */
     struct MtmTransState* next;        /* Next element in L1 list of all finished transaction present in xid2state hash */
-	TransactionId xids[1];             /* transaction ID at replicas: varying size MtmNodes */
 } MtmTransState;
 
 typedef struct

contrib/mmts/pglogical_apply.c

@@ -499,7 +499,7 @@ process_remote_commit(StringInfo in)
 			CommitTransactionCommand();
 			StartTransactionCommand();
 			/* PREPARE itself */
-			MtmSetCurrentTransactionGid(gid);
+			MtmSetCurrentTransactionGID(gid);
 			PrepareTransactionBlock(gid);
 			CommitTransactionCommand();
 			break;
@@ -519,7 +519,7 @@ process_remote_commit(StringInfo in)
 		{
 			gid = pq_getmsgstring(in);
 			StartTransactionCommand();
-			MtmSetCurrentTransactionGid(gid);
+			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, false);
 			CommitTransactionCommand();
 			break;

contrib/mmts/pglogical_proto.c

@@ -122,12 +122,12 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 
 
 static csn_t
-MtmGetCSN(TransactionId xid)
+MtmGetCSN(PGLogicalProtoMM* mm, TransactionId xid)
 {
 	MtmTransState* ts;
 	csn_t csn;
 	MtmLock(LW_SHARED);
-	ts = (MtmTransState*)hash_search(xid2state, &xid, HASH_FIND, NULL);
+	ts = (MtmTransState*)hash_search(mm->xid2state, &xid, HASH_FIND, NULL);
 	Assert(ts != NULL);
 	csn = ts->csn;
 	MtmUnlock();
@@ -171,7 +171,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     pq_sendint64(out, txn->commit_time);
 
 	if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED) { 
-		pq_sendint64(out, MtmGetCSN(txn->xid));
+		pq_sendint64(out, MtmGetCSN(mm, txn->xid));
 	}
     if (txn->xact_action == XLOG_XACT_PREPARE ||
 		txn->xact_action == XLOG_XACT_COMMIT_PREPARED ||