Fixes in 2pc

Author: knizhnik

arbiter.c

@@ -677,6 +677,7 @@ static void MtmTransReceiver(Datum arg)
 								ts->status = TRANSACTION_STATUS_UNKNOWN;
 								MtmWakeUpBackend(ts);
 							}
+							break;
 						  default:
 							Assert(false);
 						} 

multimaster.c

@@ -110,6 +110,7 @@ static void MtmXactCallback(XactEvent event, void *arg);
 static void MtmBeginTransaction(MtmCurrentTrans* x);
 static void MtmPrePrepareTransaction(MtmCurrentTrans* x);
 static void MtmPostPrepareTransaction(MtmCurrentTrans* x);
+static void MtmAbortPreparedTransaction(MtmCurrentTrans* x);
 static void MtmEndTransaction(MtmCurrentTrans* x, bool commit);
 static TransactionId MtmGetOldestXmin(Relation rel, bool ignoreVacuum);
 static bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
@@ -496,6 +497,9 @@ MtmXactCallback(XactEvent event, void *arg)
 	  case XACT_EVENT_POST_PREPARE:
 		MtmPostPrepareTransaction(&dtmTx);
 		break;
+	  case XACT_EVENT_ABORT_PREPARED:
+		MtmAbortPreparedTransaction(&dtmTx);
+		break;
 	  case XACT_EVENT_COMMIT:
 		MtmEndTransaction(&dtmTx, true);
 		break;
@@ -522,6 +526,7 @@ MtmResetTransaction(MtmCurrentTrans* x)
 	x->snapshot = INVALID_CSN;
 	x->xid = InvalidTransactionId;
 	x->gtid.xid = InvalidTransactionId;
+	x->isDistributed = false;
 }
 
 static void 
@@ -620,14 +625,16 @@ static void
 MtmPostPrepareTransaction(MtmCurrentTrans* x)
 { 
 	MtmTransState* ts;
+	MtmTransMap* tm;
 
 	MtmLock(LW_EXCLUSIVE);
 	ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
 	Assert(ts != NULL);
+	tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
+	Assert(x->gid[0]);
+	tm->state = ts;
+
 	if (!MtmIsCoordinator(ts)) {
-		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
-		Assert(x->gid[0]);
-		tm->state = ts;
 		MtmSendNotificationMessage(ts, MSG_READY); /* send notification to coordinator */
 		MtmUnlock();
 		MtmResetTransaction(x);
@@ -646,6 +653,20 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 }
 
 
+static void 
+MtmAbortPreparedTransaction(MtmCurrentTrans* x)
+{
+	MtmTransMap* tm;
+
+	MtmLock(LW_EXCLUSIVE);
+	tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
+	Assert(tm != NULL);
+	tm->state->status = TRANSACTION_STATUS_ABORTED;
+	MtmAdjustSubtransactions(tm->state);
+	MtmUnlock();
+	MtmResetTransaction(x);		
+}
+
 static void 
 MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 {
@@ -695,9 +716,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		}				
 		MtmUnlock();
 	}
-	x->snapshot = INVALID_CSN;
-	x->xid = InvalidTransactionId;
-	x->gtid.xid = InvalidTransactionId;
+	MtmResetTransaction(x);
 	MtmCheckSlots();
 }
 
@@ -1735,17 +1754,16 @@ MtmGenerateGid(char* gid)
 
 static void MtmTwoPhaseCommit(char *completionTag)
 {
-	char gid[MULTIMASTER_MAX_GID_SIZE];
-	MtmGenerateGid(gid);
+	MtmGenerateGid(dtmTx.gid);
 	if (!IsTransactionBlock()) { 
 		elog(WARNING, "Start transaction block for %d", dtmTx.xid);
 		BeginTransactionBlock();
 		CommitTransactionCommand();
 		StartTransactionCommand();
 	}
-	if (!PrepareTransactionBlock(gid))
+	if (!PrepareTransactionBlock(dtmTx.gid))
 	{
-		elog(WARNING, "Failed to prepare transaction %s", gid);
+		elog(WARNING, "Failed to prepare transaction %s", dtmTx.gid);
 		/* report unsuccessful commit in completionTag */
 		if (completionTag) { 
 			strcpy(completionTag, "ROLLBACK");
@@ -1755,10 +1773,10 @@ static void MtmTwoPhaseCommit(char *completionTag)
 		CommitTransactionCommand();
 		StartTransactionCommand();
 		if (MtmGetCurrentTransactionStatus() == TRANSACTION_STATUS_ABORTED) { 
-			FinishPreparedTransaction(gid, false);
-			elog(ERROR, "Transaction %s is aborted by DTM", gid);
+			FinishPreparedTransaction(dtmTx.gid, false);
+			elog(ERROR, "Transaction %s is aborted by DTM", dtmTx.gid);
 		} else {
-			FinishPreparedTransaction(gid, true);
+			FinishPreparedTransaction(dtmTx.gid, true);
 		}
 	}
 }

pglogical_apply.c

@@ -464,13 +464,21 @@ read_rel(StringInfo s, LOCKMODE mode)
 	return heap_open(relid, NoLock);
 }
 
+static void
+MtmSetCurrentSession(int nodeId)
+{
+	char slot_name[MULTIMASTER_MAX_SLOT_NAME_SIZE];
+	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, nodeId);
+	replorigin_session_origin = replorigin_by_name(slot_name, false); 
+	replorigin_session_setup(replorigin_session_origin);
+}
+
 static void
 process_remote_commit(StringInfo in)
 {
 	uint8 			flags;
 	uint8 			nodeId;
 	const char	   *gid = NULL;
-	char            slot_name[MULTIMASTER_MAX_SLOT_NAME_SIZE];
 
 	/* read flags */
 	flags = pq_getmsgbyte(in);
@@ -481,17 +489,14 @@ process_remote_commit(StringInfo in)
 	pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
 
-	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, nodeId);
-	replorigin_session_origin = replorigin_by_name(slot_name, false); 
-	replorigin_session_setup(replorigin_session_origin);
-
 	switch(PGLOGICAL_XACT_EVENT(flags))
 	{
 		case PGLOGICAL_COMMIT:
 		{
 			MTM_TRACE("%d: PGLOGICAL_COMMIT commit\n", MyProcPid);
 			if (IsTransactionState()) {
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
+				MtmSetCurrentSession(nodeId);
 				CommitTransactionCommand();
 			}
 			break;
@@ -505,6 +510,7 @@ process_remote_commit(StringInfo in)
 			BeginTransactionBlock();
 			CommitTransactionCommand();
 			StartTransactionCommand();
+			MtmSetCurrentSession(nodeId);
 			/* PREPARE itself */
 			MtmSetCurrentTransactionGID(gid);
 			PrepareTransactionBlock(gid);
@@ -517,6 +523,7 @@ process_remote_commit(StringInfo in)
 			gid = pq_getmsgstring(in);
 			MTM_TRACE("%d: PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s\n", MyProcPid, csn, gid);
 			StartTransactionCommand();
+			MtmSetCurrentSession(nodeId);
 			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, true);
 			CommitTransactionCommand();