Add 3PC support

Author: knizhnik

arbiter.c

@@ -536,11 +536,13 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 	nodemask_t save_mask = busy_mask;
 	BIT_SET(busy_mask, node);
 	while (true) {
+#if 0
 		if (sockets[node] >= 0 && BIT_CHECK(Mtm->reconnectMask, node)) {
 			elog(WARNING, "Arbiter is forced to reconnect to node %d", node+1); 
 			close(sockets[node]);
 			sockets[node] = -1;
 		}
+#endif
 		if (BIT_CHECK(Mtm->reconnectMask, node)) {
 			MtmLock(LW_EXCLUSIVE);		
 			BIT_CLEAR(Mtm->reconnectMask, node);
@@ -872,7 +874,8 @@ static void MtmReceiver(Datum arg)
 				}  
 
 				rc = MtmReadFromNode(i, (char*)rxBuffer[i].data + rxBuffer[i].used, rxBuffer[i].size-rxBuffer[i].used);
-				if (rc <= 0) { 
+				if (rc <= 0) {
+					MTM_LOG1("Failed to read response from node %d", i+1);
 					continue;
 				}
 
@@ -940,6 +943,8 @@ static void MtmReceiver(Datum arg)
 									if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 										elog(LOG, "Commit transaction %s because it is prepared at all live nodes", msg->gid);		
 										MtmFinishPreparedTransaction(ts, true);
+									} else { 
+										MTM_LOG1("Receive response for transaction %s -> %d, participants=%llx, voted=%llx", msg->gid, msg->status, (long long)ts->participantsMask, (long long)ts->votedMask);		
 									}
 								} else {
 									elog(LOG, "Receive response %s for transaction %s for node %d, votedMask %llx, participantsMask %llx",
@@ -1009,7 +1014,8 @@ static void MtmReceiver(Datum arg)
 									} else if (MtmUseDtm) { 
 										ts->votedMask = 0;
 										MTM_TXTRACE(ts, "MtmTransReceiver send MSG_PRECOMMIT");
-										MtmSend2PCMessage(ts, MSG_PRECOMMIT);									  
+										//MtmSend2PCMessage(ts, MSG_PRECOMMIT);	
+										SetPrepareTransactionState(ts->gid, "precommitted");								  
 									} else { 
 										ts->status = TRANSACTION_STATUS_UNKNOWN;
 										MtmWakeUpBackend(ts);
@@ -1056,6 +1062,7 @@ static void MtmReceiver(Datum arg)
 					} else { 
 						switch (msg->code) { 
 						  case MSG_PRECOMMIT:
+							Assert(false); // Now send through pglogical 
 							if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
 								ts->status = TRANSACTION_STATUS_UNKNOWN;
 								ts->csn = MtmAssignCSN();

bgwpool.c

@@ -17,11 +17,13 @@
 bool MtmIsLogicalReceiver;
 int  MtmMaxWorkers;
 
-static BgwPool* pool;
+static BgwPool* MtmPool;
 
 static void BgwShutdownWorker(int sig)
 {
-	BgwPoolStop(pool);
+	if (MtmPool) { 
+		BgwPoolStop(MtmPool);
+	}
 }
 
 static void BgwPoolMainLoop(BgwPool* pool)
@@ -32,6 +34,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
 	sigset_t sset;
 
 	MtmIsLogicalReceiver = true;
+	MtmPool = pool;
 
 	signal(SIGINT, BgwShutdownWorker);
 	signal(SIGQUIT, BgwShutdownWorker);
@@ -88,6 +91,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
 
 void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  char const* dbuser, size_t queueSize, size_t nWorkers)
 {
+	MtmPool = pool;
     pool->queue = (char*)ShmemAlloc(queueSize);
     pool->executor = executor;
     PGSemaphoreCreate(&pool->available);

multimaster.c

@@ -416,7 +416,7 @@ csn_t MtmTransactionSnapshot(TransactionId xid)
 
 	MtmLock(LW_SHARED);
 	if (Mtm->status == MTM_ONLINE) {
-		MtmTransState* ts = hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+		MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
 		if (ts != NULL && !ts->isLocal) { 
 			snapshot = ts->snapshot;
 			Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession); 		
@@ -806,7 +806,7 @@ static MtmTransState*
 MtmCreateTransState(MtmCurrentTrans* x)
 {
 	bool found;
-	MtmTransState* ts = hash_search(MtmXid2State, &x->xid, HASH_ENTER, &found);
+	MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &x->xid, HASH_ENTER, &found);
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;
 	ts->snapshot = x->snapshot;
 	ts->isLocal = true;
@@ -859,6 +859,10 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 		/* Do not take in account bg-workers which are performing recovery */
 		elog(ERROR, "Abort current transaction because this cluster node is in %s status", MtmNodeStatusMnem[Mtm->status]);			
 	}
+	if (TransactionIdIsValid(x->gtid.xid) && BIT_CHECK(Mtm->disabledNodeMask, x->gtid.node-1)) {
+		/* Coordinator of transaction is disabled: just abort transaction without any further steps */
+		elog(ERROR, "Abort transaction %d because it's coordinator %d was disabled", x->xid, x->gtid.node);
+	}
 
 	MtmLock(LW_EXCLUSIVE);
 
@@ -920,6 +924,32 @@ bool MtmWatchdog(timestamp_t now)
 	return allAlive;
 }
 
+/* 
+ * Mark transaction as precommitted
+ */
+void MtmPrecommitTransaction(char const* gid)
+{
+	MtmLock(LW_EXCLUSIVE);
+	{
+		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, gid, HASH_FIND, NULL);
+		if (tm == NULL) {
+			elog(WARNING, "MtmPrecommitTransaction: transaciton '%s' is not found", gid);
+		} else { 
+			MtmTransState* ts = tm->state;
+			Assert(ts != NULL);
+			Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
+			ts->status = TRANSACTION_STATUS_UNKNOWN;
+			ts->csn = MtmAssignCSN();
+			MtmAdjustSubtransactions(ts);
+			MtmSend2PCMessage(ts, MSG_PRECOMMITTED);
+		}
+	}
+	MtmUnlock();
+}
+				
+				
+	
+
 
 static bool 
 MtmVotingCompleted(MtmTransState* ts)
@@ -944,7 +974,8 @@ MtmVotingCompleted(MtmTransState* ts)
 				return true;
 			} else if (MtmUseDtm) {
 				ts->votedMask = 0;
-				MtmSend2PCMessage(ts, MSG_PRECOMMIT);
+				SetPrepareTransactionState(ts->gid, "precommitted");
+				//MtmSend2PCMessage(ts, MSG_PRECOMMIT);
 				return false;
 			} else {
 				ts->status = TRANSACTION_STATUS_UNKNOWN;
@@ -964,7 +995,8 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 	int nConfigChanges = Mtm->nConfigChanges;
 	timestamp_t prepareTime = ts->csn - ts->snapshot;
 	timestamp_t timeout = Max(prepareTime + MSEC_TO_USEC(MtmMin2PCTimeout), prepareTime*MtmMax2PCRatio/100);
-	timestamp_t deadline = MtmGetSystemTime() + timeout;
+	timestamp_t start = MtmGetSystemTime();
+	timestamp_t deadline = start + timeout;
 	timestamp_t now;
 
 	Assert(ts->csn > ts->snapshot);
@@ -989,7 +1021,8 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 		if (now > deadline) { 
 			if (ts->isPrepared) { 
 				/* resend precommit message */
-				MtmSend2PCMessage(ts, MSG_PRECOMMIT);
+				// MtmSend2PCMessage(ts, MSG_PRECOMMIT);
+				elog(LOG, "Distributes transaction is not committed in %ld msec", USEC_TO_MSEC(now - start));
 			} else {
 				elog(WARNING, "Commit of distributed transaction is canceled because of %ld msec timeout expiration", USEC_TO_MSEC(timeout));
 				MtmAbortTransaction(ts);
@@ -1032,15 +1065,15 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		elog(ERROR, "ERROR INJECTION for transaction %d (%s)", x->xid, x->gid);
 	}
 	MtmLock(LW_EXCLUSIVE);
-	ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
+	ts = (MtmTransState*)hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
 	Assert(ts != NULL);
 	//if (x->gid[0]) MTM_LOG1("Preparing transaction %d (%s) at %ld", x->xid, x->gid, MtmGetCurrentTime());
 	if (!MtmIsCoordinator(ts) || Mtm->status == MTM_RECOVERY) {
 		MTM_TXTRACE(x, "recovery?");
 		Assert(x->gid[0]);
 		ts->votingCompleted = true;
 		MTM_TXTRACE(x, "recovery? 1");
-		if (Mtm->status != MTM_RECOVERY || Mtm->recoverySlot != MtmReplicationNodeId) {
+		if (Mtm->status != MTM_RECOVERY/* || Mtm->recoverySlot != MtmReplicationNodeId*/) {
 			MTM_TXTRACE(x, "recovery? 2"); 
 			MtmSend2PCMessage(ts, MSG_PREPARED); /* send notification to coordinator */
 			if (!MtmUseDtm) { 
@@ -1097,7 +1130,8 @@ MtmCommitPreparedTransaction(MtmCurrentTrans* x)
 		ts->votedMask = 0;
 		ts->procno = MyProc->pgprocno;
 		MTM_TXTRACE(ts, "Coordinator sends MSG_PRECOMMIT");
-		MtmSend2PCMessage(ts, MSG_PRECOMMIT);
+		SetPrepareTransactionState(ts->gid, "precommitted");
+		//MtmSend2PCMessage(ts, MSG_PRECOMMIT);
 
 		Mtm2PCVoting(x, ts);
 
@@ -1154,7 +1188,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		MtmTransState* ts = NULL;
 		MtmLock(LW_EXCLUSIVE);
 		if (x->isPrepared) { 
-			ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
+			ts = (MtmTransState*)hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
 			Assert(ts != NULL);
 			Assert(strcmp(x->gid, ts->gid) == 0);
 		} else if (x->gid[0]) { 
@@ -1206,7 +1240,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 			if (ts == NULL) { 
 				bool found;
 				Assert(TransactionIdIsValid(x->xid));
-				ts = hash_search(MtmXid2State, &x->xid, HASH_ENTER, &found);
+				ts = (MtmTransState*)hash_search(MtmXid2State, &x->xid, HASH_ENTER, &found);
 				if (!found) { 
 					ts->isEnqueued = false;
 					ts->isActive = false;
@@ -1316,6 +1350,53 @@ static void MtmBroadcastPollMessage(MtmTransState* ts)
 	}
 }
 
+/*
+ * Restore state of recovered prepared transaction in memory
+ */
+static void	MtmLoadPreparedTransactions(void)
+{
+	PreparedTransaction pxacts;
+	int n = GetPreparedTransactions(&pxacts);
+	int i;
+
+	for (i = 0; i < n; i++) { 
+		bool found;
+		char const* gid = pxacts[i].gid;
+		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, gid, HASH_ENTER, &found);
+		if (!found) {
+			TransactionId xid = GetNewTransactionId(false);
+			MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_ENTER, &found);
+			MTM_LOG1("Recover prepared transaction %s xid %d", gid, xid);
+			MyPgXact->xid = InvalidTransactionId; /* dirty hack:((( */
+			Assert(!found);
+			Mtm->nActiveTransactions += 1;
+			ts->isEnqueued = false;
+			ts->isActive = true;
+			ts->status = strcmp(pxacts[i].state_3pc, "precommitted") == 0 ? TRANSACTION_STATUS_UNKNOWN : TRANSACTION_STATUS_IN_PROGRESS;
+			ts->isLocal = true;
+			ts->isPrepared = false;
+			ts->isPinned = false;
+			ts->snapshot = INVALID_CSN;
+			ts->isTwoPhase = false;
+			ts->csn = 0; /* should be replaced with real CSN by poll result */
+			ts->gtid.node = MtmNodeId;
+			ts->gtid.xid = xid;
+			ts->nSubxids = 0;
+			ts->votingCompleted = true;
+			ts->participantsMask = (((nodemask_t)1 << Mtm->nAllNodes) - 1) & ~Mtm->disabledNodeMask & ~((nodemask_t)1 << (MtmNodeId-1));
+			ts->votedMask = 0;
+			strcpy(ts->gid, gid);
+			MtmTransactionListAppend(ts);			
+			tm->status = ts->status;
+			tm->state = ts;
+			MtmBroadcastPollMessage(ts);
+		}
+	}
+	MTM_LOG1("Recover %d prepared transactions", n);
+	if (pxacts) { 
+		pfree(pxacts);
+	}
+}
 
 static void MtmStartRecovery()
 {
@@ -2051,6 +2132,7 @@ static void MtmCheckControlFile(void)
 	}
 }
 
+
 static void MtmInitialize()
 {
 	bool found;
@@ -2087,6 +2169,7 @@ static void MtmInitialize()
 		Mtm->nConfigChanges = 0;
 		Mtm->recoveryCount = 0;
 		Mtm->localTablesHashLoaded = false;
+		Mtm->preparedTransactionsLoaded = false;
 		Mtm->inject2PCError = 0;
 		Mtm->sendQueue = NULL;
 		Mtm->freeQueue = NULL;
@@ -2851,6 +2934,13 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 	MtmReplicationMode mode = REPLMODE_OPEN_EXISTED;
 
 	MtmLock(LW_EXCLUSIVE);
+	
+	if (!Mtm->preparedTransactionsLoaded)
+	{
+		MtmLoadPreparedTransactions();
+		Mtm->preparedTransactionsLoaded = true;
+	}
+
 	while ((Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) || BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) 
 	{ 	
 		if (*shutdown) 
@@ -3330,7 +3420,7 @@ mtm_get_csn(PG_FUNCTION_ARGS)
 	csn_t csn = INVALID_CSN;
 
 	MtmLock(LW_SHARED);
-    ts = hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+    ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
     if (ts != NULL) { 
 		csn = ts->csn;
 	}
@@ -3859,14 +3949,12 @@ static inline void MtmGucUpdate(const char *key, char *value)
 	MtmGucEntry *hentry;
 	bool found;
 
-	hentry = hash_search(MtmGucHash, key, HASH_FIND, &found);
+	hentry = (MtmGucEntry*)hash_search(MtmGucHash, key, HASH_ENTER, &found);
 	if (found)
 	{
 		pfree(hentry->value);
 		dlist_delete(&hentry->list_node);
 	}
-
-	hentry = hash_search(MtmGucHash, key, HASH_ENTER, NULL);
 	hentry->value = value;
 	dlist_push_tail(&MtmGucList, &hentry->list_node);
 }
@@ -3876,7 +3964,7 @@ static inline void MtmGucRemove(const char *key)
 	MtmGucEntry *hentry;
 	bool found;
 
-	hentry = hash_search(MtmGucHash, key, HASH_FIND, &found);
+	hentry = (MtmGucEntry*)hash_search(MtmGucHash, key, HASH_FIND, &found);
 	if (found)
 	{
 		pfree(hentry->value);
@@ -4465,7 +4553,7 @@ MtmSerializeLock(PROCLOCK* proclock, void* arg)
 }
 
 static bool 
-MtmDetectGlobalDeadLockFortXid(TransactionId xid)
+MtmDetectGlobalDeadLockForXid(TransactionId xid)
 {
 	bool hasDeadlock = false;
     if (TransactionIdIsValid(xid)) { 
@@ -4530,11 +4618,11 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 
 	MTM_LOG1("Detect global deadlock for %d by backend %d", pgxact->xid, MyProcPid);
 
-    return MtmDetectGlobalDeadLockFortXid(pgxact->xid);
+    return MtmDetectGlobalDeadLockForXid(pgxact->xid);
 }
 
 Datum mtm_check_deadlock(PG_FUNCTION_ARGS)
 {
 	TransactionId xid = PG_GETARG_INT32(0);
-    PG_RETURN_BOOL(MtmDetectGlobalDeadLockFortXid(xid));
+    PG_RETURN_BOOL(MtmDetectGlobalDeadLockForXid(xid));
 }