Filter applied transactions

Author: knizhnik

arbiter.c

@@ -313,12 +313,14 @@ static void MtmSetSocketOptions(int sd)
 
 static void MtmCheckResponse(MtmArbiterMessage* resp)
 {
-	if (BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1) && !BIT_CHECK(Mtm->disabledNodeMask, resp->node-1)) { 
+	if (BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1) 
+		&& !BIT_CHECK(Mtm->disabledNodeMask, resp->node-1)
+		&& Mtm->status != MTM_RECOVERY
+		&& Mtm->nodes[MtmNodeId-1].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) < MtmGetSystemTime()) 
+	{ 
 		elog(WARNING, "Node %d thinks that I was dead, while I am %s (message %s)", resp->node, MtmNodeStatusMnem[Mtm->status], messageKindText[resp->code]);
-		if (Mtm->status != MTM_RECOVERY) { 
-			BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
-			MtmSwitchClusterMode(MTM_RECOVERY);
-		}
+		BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
+		MtmSwitchClusterMode(MTM_RECOVERY);
 	}
 }
 
@@ -561,7 +563,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 static int MtmReadFromNode(int node, void* buf, int buf_size)
 {
 	int rc = MtmReadSocket(sockets[node], buf, buf_size);
-	if (rc < 0) { 
+	if (rc <= 0) { 
 		elog(WARNING, "Arbiter failed to read from node=%d, rc=%d, errno=%d", node+1, rc, errno);
 		MtmDisconnect(node);
 	}
@@ -957,6 +959,7 @@ static void MtmReceiver(Datum arg)
 						elog(WARNING, "Ignore response for unexisted transaction %d from node %d", msg->dxid, node);
 						continue;
 					}
+					Assert(msg->code == MSG_ABORTED || strcmp(msg->gid, ts->gid) == 0);
 					if (BIT_CHECK(ts->votedMask, node-1)) {
 						elog(WARNING, "Receive deteriorated %s response for transaction %d (%s) from node %d",
 							 messageKindText[msg->code], ts->xid, ts->gid, node);
@@ -990,6 +993,8 @@ static void MtmReceiver(Datum arg)
 									MtmWakeUpBackend(ts);								
 								} else { 
 									Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
+									MTM_LOG1("Transaction %s is prepared (status=%d participants=%lx disabled=%lx, voted=%lx)", 
+											 ts->gid, ts->status, ts->participantsMask, Mtm->disabledNodeMask, ts->votedMask);
 									ts->isPrepared = true;
 									if (ts->isTwoPhase) { 
 										MtmWakeUpBackend(ts);										
@@ -1048,9 +1053,11 @@ static void MtmReceiver(Datum arg)
 								ts->csn = MtmAssignCSN();
 								MtmAdjustSubtransactions(ts);
 								MtmSend2PCMessage(ts, MSG_PRECOMMITTED);
-							} else {
-								Assert(ts->status == TRANSACTION_STATUS_ABORTED);
+							} else if (ts->status == TRANSACTION_STATUS_ABORTED) {
 								MtmSend2PCMessage(ts, MSG_ABORTED);
+							} else { 
+								elog(WARNING, "Transaction %s is already %s",
+									 ts->gid, ts->status == TRANSACTION_STATUS_COMMITTED ? "committed" : "prepared");
 							}
 							break;
 						  default:

multimaster.c

@@ -15,6 +15,8 @@
 #include "miscadmin.h"
 
 #include "libpq-fe.h"
+#include "lib/stringinfo.h"
+#include "libpq/pqformat.h"
 #include "common/username.h"
 
 #include "postmaster/postmaster.h"
@@ -925,7 +927,9 @@ MtmVotingCompleted(MtmTransState* ts)
 			ts->votingCompleted = true;
 			ts->status = TRANSACTION_STATUS_UNKNOWN;
 			return true;
-		} else { 
+		} else {
+			MTM_LOG1("Transaction %s is considered as prepared (status=%d participants=%lx disabled=%lx, voted=%lx)", 
+					 ts->gid, ts->status, ts->participantsMask, Mtm->disabledNodeMask, ts->votedMask);
 			ts->isPrepared = true;
 			if (ts->isTwoPhase) {
 				ts->votingCompleted = true;
@@ -979,9 +983,10 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		MtmResetTransaction();
 	} else { 
 		int result = 0;
-		
+		int nConfigChanges = Mtm->nConfigChanges;
 		/* Wait votes from all nodes until: */
-		while (!MtmVotingCompleted(ts)) 
+		while (!MtmVotingCompleted(ts)
+			   && (ts->isPrepared || nConfigChanges == Mtm->nConfigChanges))
 		{ 
 			MtmUnlock();
 			MTM_TXTRACE(x, "PostPrepareTransaction WaitLatch Start");
@@ -997,8 +1002,15 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 			MtmLock(LW_EXCLUSIVE);
 		}
 		if (ts->status != TRANSACTION_STATUS_ABORTED && !ts->votingCompleted) { 
+			if (ts->isPrepared) { 
+				elog(ERROR, "Commit of distributed transaction %s is suspended because node is switched to %s mode", ts->gid, MtmNodeStatusMnem[Mtm->status]);
+			}
+			if (Mtm->status != MTM_ONLINE) { 
+				elog(WARNING, "Commit of distributed transaction is canceled because node is switched to %s mode", MtmNodeStatusMnem[Mtm->status]);
+			} else {
+				elog(WARNING, "Commit of distributed transaction is canceled because cluster configuration was changed");
+			}
 			MtmAbortTransaction(ts);
-			elog(WARNING, "Commit of distributed transaction is canceled because node is switched to %s mode", MtmNodeStatusMnem[Mtm->status]);
 		}
 		x->status = ts->status;
 		MTM_LOG3("%d: Result of vote: %d", MyProcPid, ts->status);
@@ -1031,6 +1043,7 @@ MtmCommitPreparedTransaction(MtmCurrentTrans* x)
 		elog(WARNING, "Global transaciton ID '%s' is not found", x->gid);
 	} else {
 		int result = 0;
+		int nConfigChanges = Mtm->nConfigChanges;
 
  		Assert(tm->state != NULL);
 		MTM_LOG3("Commit prepared transaction %d with gid='%s'", x->xid, x->gid);
@@ -1045,7 +1058,8 @@ MtmCommitPreparedTransaction(MtmCurrentTrans* x)
 		MtmSend2PCMessage(ts, MSG_PRECOMMIT);
 
 		/* Wait votes from all nodes until: */
-		while (!MtmVotingCompleted(ts))
+		while (!MtmVotingCompleted(ts)
+			   && (ts->isPrepared || nConfigChanges == Mtm->nConfigChanges))
 		{
 			MtmUnlock();
 			MTM_TXTRACE(x, "CommitPreparedTransaction WaitLatch Start");
@@ -1062,8 +1076,15 @@ MtmCommitPreparedTransaction(MtmCurrentTrans* x)
 			}
 		}
 		if (ts->status != TRANSACTION_STATUS_ABORTED && !ts->votingCompleted) { 
+			if (ts->isPrepared) { 
+				elog(ERROR, "Commit of distributed transaction %s is suspended because node is switched to %s mode", ts->gid, MtmNodeStatusMnem[Mtm->status]);
+			}
+			if (Mtm->status != MTM_ONLINE) { 
+				elog(WARNING, "Commit of distributed transaction is canceled because node is switched to %s mode", MtmNodeStatusMnem[Mtm->status]);
+			} else {
+				elog(WARNING, "Commit of distributed transaction is canceled because cluster configuration was changed");
+			}
 			MtmAbortTransaction(ts);
-			elog(WARNING, "Commit of distributed transaction is canceled because node is switched to %s mode", MtmNodeStatusMnem[Mtm->status]);
 		}
 		x->status = ts->status;
 		x->xid = ts->xid;
@@ -1165,11 +1186,14 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 				}
 				ts->status = TRANSACTION_STATUS_ABORTED;
 				ts->isLocal = true;
+				ts->isPrepared = false;
 				ts->snapshot = x->snapshot;
+				ts->isTwoPhase = x->isTwoPhase;
 				ts->csn = MtmAssignCSN();	
 				ts->gtid = x->gtid;
 				ts->nSubxids = 0;
 				ts->votingCompleted = true;
+				strcpy(ts->gid, x->gid);
 				if (ts->isActive) { 
 					ts->isActive = false;
 					Assert(Mtm->nActiveTransactions != 0);
@@ -1225,8 +1249,9 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 		int i;
 		for (i = 0; i < Mtm->nAllNodes; i++)
 		{
-			if (BIT_CHECK(ts->participantsMask & ~Mtm->disabledNodeMask, i) && TransactionIdIsValid(ts->xids[i]))
+			if (BIT_CHECK(ts->participantsMask & ~Mtm->disabledNodeMask, i))
 			{
+				Assert(TransactionIdIsValid(ts->xids[i]));
 				msg.node = i+1;
 				msg.dxid = ts->xids[i];
 				MtmSendMessage(&msg);
@@ -1654,7 +1679,7 @@ MtmCheckClusterLock()
 				continue;
 			} else {  
 				/* All lockers are synchronized their logs */
-				/* Remove lock and mark them as rceovered */
+				/* Remove lock and mark them as recovered */
 				MTM_LOG1("Complete recovery of %d nodes (node mask %lx)", Mtm->nLockers, (long) Mtm->nodeLockerMask);
 				Assert(Mtm->walSenderLockerMask == 0);
 				Assert((Mtm->nodeLockerMask & Mtm->disabledNodeMask) == Mtm->nodeLockerMask);
@@ -2081,6 +2106,8 @@ static void MtmInitialize()
 			Mtm->nodes[i].timeline = 0;
 		}
 		Mtm->nodes[MtmNodeId-1].originId = DoNotReplicateId;
+		/* All transaction originated from the current node should be ignored during recovery */
+		Mtm->nodes[MtmNodeId-1].restartLsn = (XLogRecPtr)PG_UINT64_MAX;
 		PGSemaphoreCreate(&Mtm->sendSemaphore);
 		PGSemaphoreReset(&Mtm->sendSemaphore);
 		SpinLockInit(&Mtm->spinlock);
@@ -2807,12 +2834,7 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
 	Assert(!IsTransactionState());
 	MtmResetTransaction();
 	StartTransactionCommand();
-#if 0
-	if (Mtm->nodes[MtmNodeId-1].originId == InvalidRepOriginId) { 
-		/* This dummy origin is used for local commits/aborts which should not be replicated */
-		Mtm->nodes[MtmNodeId-1].originId = replorigin_create(psprintf(MULTIMASTER_SLOT_PATTERN, MtmNodeId));
-	}
-#endif
+
 	MtmBeginSession(MtmNodeId);
 	MtmSetCurrentTransactionCSN(ts->csn);
 	MtmSetCurrentTransactionGID(ts->gid);
@@ -2829,7 +2851,6 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
  */
 MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown)
 {
-	int i;
 	bool recovery = false;
 
 	while (Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) 
@@ -2851,9 +2872,6 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 				Mtm->nReceivers = 0;
 				Mtm->recoveryCount += 1;
 				Mtm->pglogicalNodeMask = 0;
-				for (i = 0; i < Mtm->nAllNodes; i++) { 
-					Mtm->nodes[i].restartLsn = InvalidXLogRecPtr;
-				}
 				MtmUnlock();
 				return REPLMODE_RECOVERY;
 			}
@@ -3070,6 +3088,67 @@ MtmReplicationRowFilterHook(struct PGLogicalRowFilterArgs* args)
 	return isDistributed;
 }
 
+bool MtmFilterTransaction(char* record, int size)
+{
+	StringInfoData s;
+	uint8       flags;
+	XLogRecPtr  origin_lsn;
+	XLogRecPtr  end_lsn;
+	int         replication_node;
+	int         origin_node;
+	char const* gid = "";
+	bool        duplicate;
+
+    s.data = record;
+    s.len = size;
+    s.maxlen = -1;
+	s.cursor = 0;
+
+	Assert(pq_getmsgbyte(&s) == 'C');
+	flags = pq_getmsgbyte(&s); /* flags */
+	replication_node = pq_getmsgbyte(&s);
+
+	/* read fields */
+	pq_getmsgint64(&s); /* commit_lsn */
+	end_lsn = pq_getmsgint64(&s); /* end_lsn */
+	pq_getmsgint64(&s); /* commit_time */
+	
+	origin_node = pq_getmsgbyte(&s);
+	origin_lsn = pq_getmsgint64(&s);
+
+	Assert(replication_node == MtmReplicationNodeId && 
+		   origin_node != 0 &&
+		   (Mtm->status == MTM_RECOVERY || origin_node == replication_node));	
+
+    switch(PGLOGICAL_XACT_EVENT(flags))
+    {
+	  case PGLOGICAL_PREPARE:
+	  case PGLOGICAL_ABORT_PREPARED:
+		gid = pq_getmsgstring(&s);
+		break;
+	  case PGLOGICAL_COMMIT_PREPARED:
+		pq_getmsgint64(&s); /* CSN */
+		gid = pq_getmsgstring(&s);
+		break;
+	  default:
+		break;
+	}
+	duplicate = Mtm->status == MTM_RECOVERY && origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLsn;
+	
+	MTM_LOG1("%s transaction %s from node %d lsn %lx, origin node %d, original lsn=%lx, current lsn=%lx", 
+			 duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, origin_node, origin_lsn, Mtm->nodes[origin_node-1].restartLsn);
+	if (Mtm->status == MTM_RECOVERY) { 
+		if (Mtm->nodes[origin_node-1].restartLsn < origin_lsn) { 
+			Mtm->nodes[origin_node-1].restartLsn = origin_lsn;
+		}
+	} else { 
+		if (Mtm->nodes[replication_node-1].restartLsn < end_lsn) { 
+			Mtm->nodes[replication_node-1].restartLsn = end_lsn;
+		}
+	}
+	return duplicate;
+}
+
 void MtmSetupReplicationHooks(struct PGLogicalHooks* hooks)
 {
 	hooks->startup_hook = MtmReplicationStartupHook;

multimaster.h

@@ -362,4 +362,6 @@ extern void MtmBeginSession(int nodeId);
 extern void MtmEndSession(int nodeId, bool unlock);
 extern void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit);
 extern void MtmRollbackPreparedTransaction(char const* gid);
+extern bool MtmFilterTransaction(char* record, int size);
+
 #endif

pglogical_apply.c

@@ -344,10 +344,10 @@ process_remote_begin(StringInfo s)
 	Assert(gtid.node > 0);
 
 	MTM_LOG2("REMOTE begin node=%d xid=%d snapshot=%ld", gtid.node, gtid.xid, snapshot);
+	MtmResetTransaction();		
 #if 1
 	if (BIT_CHECK(Mtm->disabledNodeMask, gtid.node-1)) { 
 		elog(WARNING, "Ignore transaction %d from disabled node %d", gtid.xid, gtid.node);
-		MtmResetTransaction();		
 		return false;
 	}
 #endif
@@ -603,9 +603,6 @@ process_remote_commit(StringInfo in)
 	origin_node = pq_getmsgbyte(in);
 	origin_lsn = pq_getmsgint64(in);
 
-	if (Mtm->nodes[origin_node-1].restartLsn < origin_lsn) { 
-		Mtm->nodes[origin_node-1].restartLsn = origin_lsn;
-	}
 	if (origin_node != MtmReplicationNodeId) {
 		replorigin_advance(Mtm->nodes[origin_node-1].originId, origin_lsn, GetXLogInsertRecPtr(),
 						   false /* backward */ , false /* WAL */ );

pglogical_receiver.c

@@ -339,6 +339,9 @@ pglogical_receiver_main(Datum main_arg)
 				MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
 			} else { 
 				originStartPos = replorigin_get_progress(originId, false);
+				if (Mtm->nodes[nodeId-1].restartLsn < originStartPos) { 
+					Mtm->nodes[nodeId-1].restartLsn = originStartPos;
+				}
 				MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
 			}
 			Mtm->nodes[nodeId-1].originId = originId;
@@ -535,27 +538,32 @@ pglogical_receiver_main(Datum main_arg)
 						ByteBufferAppend(&buf, stmt, rc - hdr_len);
 						if (stmt[0] == 'C') /* commit */
 						{
-							if (spill_file >= 0) { 
-								ByteBufferAppend(&buf, ")", 1);
-								pq_sendbyte(&spill_info, '(');
-								pq_sendint(&spill_info, buf.used, 4);
-								MtmSpillToFile(spill_file, buf.data, buf.used);
-								MtmCloseSpillFile(spill_file);
-								MtmExecute(spill_info.data, spill_info.len);
-								spill_file = -1;
-								resetStringInfo(&spill_info);
-							} else { 
-								if (MtmPreserveCommitOrder && buf.used == rc - hdr_len) {
-									/* Perform commit-prepared and rollback-prepared requested directly in receiver */
-									timestamp_t stop, start = MtmGetSystemTime();
-									MtmExecutor(buf.data, buf.used);
-									stop = MtmGetSystemTime();
-									if (stop - start > USECS_PER_SEC) { 
-										elog(WARNING, "Commit of prepared transaction takes %ld usec, flags=%x", stop - start, stmt[1]);
+							if (!MtmFilterTransaction(stmt, rc - hdr_len)) { 
+								if (spill_file >= 0) { 
+									ByteBufferAppend(&buf, ")", 1);
+									pq_sendbyte(&spill_info, '(');
+									pq_sendint(&spill_info, buf.used, 4);
+									MtmSpillToFile(spill_file, buf.data, buf.used);
+									MtmCloseSpillFile(spill_file);
+									MtmExecute(spill_info.data, spill_info.len);
+									spill_file = -1;
+									resetStringInfo(&spill_info);
+								} else { 
+									if (MtmPreserveCommitOrder && buf.used == rc - hdr_len) {
+										/* Perform commit-prepared and rollback-prepared requested directly in receiver */
+										timestamp_t stop, start = MtmGetSystemTime();
+										MtmExecutor(buf.data, buf.used);
+										stop = MtmGetSystemTime();
+										if (stop - start > USECS_PER_SEC) { 
+											elog(WARNING, "Commit of prepared transaction takes %ld usec, flags=%x", stop - start, stmt[1]);
+										}
+									} else {
+										MtmExecute(buf.data, buf.used);
 									}
-								} else {
-									MtmExecute(buf.data, buf.used);
 								}
+							} else if (spill_file >= 0) { 
+								MtmCloseSpillFile(spill_file);
+								spill_file = -1;
 							}
 							ByteBufferReset(&buf);
 						}

tests2/docker-entrypoint.sh

@@ -45,7 +45,6 @@ if [ "$1" = 'postgres' ]; then
 		# 	dbname=$POSTGRES_DB user=$POSTGRES_USER host=node2, \
 		# 	dbname=$POSTGRES_DB user=$POSTGRES_USER host=node3"
 
-#            log_line_prefix = '%t: '
 
 		cat <<-EOF >> $PGDATA/postgresql.conf
 			listen_addresses='*' 
@@ -58,6 +57,10 @@ if [ "$1" = 'postgres' ]; then
 			max_wal_senders = 10
 			shared_preload_libraries = 'raftable,multimaster'
 			default_transaction_isolation = 'repeatable read'
+			log_checkpoints = on
+			checkpoint_timeout = 30
+			log_autovacuum_min_duration = 0
+            log_line_prefix = '%t: '
 
 			multimaster.workers = 4
 			multimaster.max_nodes = 3