Add multimaster.break_connection option

Author: knizhnik

arbiter.c

@@ -327,9 +327,10 @@ static void MtmCheckResponse(MtmArbiterMessage* resp)
 	{ 
 		elog(WARNING, "Node %d thinks that I am dead, while I am %s (message %s)", resp->node, MtmNodeStatusMnem[Mtm->status], MtmMessageKindMnem[resp->code]);
 		BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
+		Mtm->nConfigChanges += 1;
 		MtmSwitchClusterMode(MTM_RECOVERY);
 	} else if (BIT_CHECK(Mtm->disabledNodeMask, resp->node-1) && sockets[resp->node-1] < 0) { 
-		/* We receive heartbeat from dsiable node with 
+		/* We receive heartbeat from disabled node.
 		 * Looks like it is restarted.
 		 * Try to reconnect to it.
 		 */
@@ -1040,13 +1041,16 @@ static void MtmReceiver(Datum arg)
 							Mtm->nodes[node-1].transDelay += MtmGetCurrentTime() - ts->csn;
 							ts->xids[node-1] = msg->sxid;
 
+#if 0
+							/* This code seems to be deteriorated because now checking that distributed transaction involves all live nodes is done at replica while applying PREPARE */
 							if ((~msg->disabledNodeMask & Mtm->disabledNodeMask) != 0) { 
 								/* Coordinator's disabled mask is wider than of this node: so reject such transaction to avoid 
 								   commit on smaller subset of nodes */
 								elog(WARNING, "Coordinator of distributed transaction %s (%llu) see less nodes than node %d: %llx instead of %llx",
 									 ts->gid, (long64)ts->xid, node, Mtm->disabledNodeMask, msg->disabledNodeMask);
 								MtmAbortTransaction(ts);
 							}
+#endif
 							if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 								/* All nodes are finished their transactions */
 								if (ts->status == TRANSACTION_STATUS_ABORTED) { 

multimaster.c

@@ -246,6 +246,7 @@ static int   MtmGcPeriod;
 static bool  MtmIgnoreTablesWithoutPk;
 static int   MtmLockCount;
 static bool  MtmMajorNode;
+static bool  MtmBreakConnection;
 
 static ExecutorStart_hook_type PreviousExecutorStartHook;
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
@@ -418,16 +419,27 @@ MtmInitializeSequence(int64* start, int64* step)
  * -------------------------------------------
  */
 
-csn_t MtmTransactionSnapshot(TransactionId xid)
+/* 
+ * Get snapshot of transaction proceed by WAL sender pglogical plugin.
+ * If it is local transaction or replication node is not in participant mask, then return INVALID_CSN.
+ * Transaction should be skept by WAL sender in the following cases:
+ *   1. Transaction was replicated from some other node and it is not a recovery process.
+ *   2. State of transaction is unknown
+ *   3. Replication node is not participated in transaction
+ */
+csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, nodemask_t* participantsMask)
 {
 	csn_t snapshot = INVALID_CSN;
-	
+	*participantsMask = 0;
 	MtmLock(LW_SHARED);
 	if (Mtm->status == MTM_ONLINE) {
 		MtmTransState* ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
-		if (ts != NULL && !ts->isLocal) { 
-			snapshot = ts->snapshot;
-			Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession); 		
+		if (ts != NULL) { 
+			*participantsMask = ts->participantsMask;
+			if (!ts->isLocal && BIT_CHECK(ts->participantsMask, nodeId-1)) { 
+				snapshot = ts->snapshot;
+				Assert(ts->gtid.node == MtmNodeId || MtmIsRecoverySession); 		
+			}
 		}
 	}
 	MtmUnlock();
@@ -621,6 +633,9 @@ MtmAdjustOldestXid(TransactionId xid)
     return xid;
 }
 
+
+
+
 /*
  * -------------------------------------------
  * Transaction list manipulation.
@@ -789,7 +804,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 			 * Allow execution of transaction by bg-workers to make it possible to perform recovery.
 			 */
 			MtmUnlock();			
-			elog(ERROR, "Multimaster node is not online: current status %s", MtmNodeStatusMnem[Mtm->status]);
+			elog(MtmBreakConnection ? FATAL : ERROR, "Multimaster node is not online: current status %s", MtmNodeStatusMnem[Mtm->status]);
 		}
 		x->containsDML = false;
         x->snapshot = MtmAssignCSN();	
@@ -799,10 +814,8 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 
 		/*
 		 * Check if there is global multimaster lock preventing new transaction from commit to make a chance to wal-senders to caught-up.
-		 * Only "own" transactions are blocked. Transactions replicated from other nodes (including recovered transaction) should be proceeded
-		 * and should not cause cluster status change.
 		 */
-		if (x->isDistributed/* && x->isReplicated*/) { 
+		if (x->isDistributed) { 
 			MtmCheckClusterLock();
 		}
 
@@ -1141,6 +1154,9 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		}
 		MtmUnlock();
 		if (x->isTwoPhase) { 
+			if (x->status == TRANSACTION_STATUS_ABORTED) { 
+				elog(ERROR, "Prepare of user's 2PC transaction %s (%llu) is aborted by DTM", x->gid, (long64)x->xid);
+			}
 			MtmResetTransaction();
 		}
 	}
@@ -2198,7 +2214,29 @@ void MtmReconnectNode(int nodeId)
 	MtmUnlock();
 }
 
-
+/*
+ * Check particioants of replicated transaction. This function is called by receiver at start of replicated transaction to
+ * check that all live nodes are participated in it.
+ */
+bool MtmCheckParticipants(GlobalTransactionId* gtid, nodemask_t participantsMask)
+{
+	bool result = true;
+	MtmLock(LW_EXCLUSIVE);		
+	if (BIT_CHECK(Mtm->disabledNodeMask, gtid->node-1)) { 
+		elog(WARNING, "Ignore transaction %llu from disabled node %d", (long64)gtid->xid, gtid->node);
+		result = false;
+	} else { 
+		nodemask_t liveMask = (((nodemask_t)1 << Mtm->nAllNodes) - 1) & ~Mtm->disabledNodeMask;
+		BIT_SET(participantsMask, gtid->node-1);
+		if (liveMask & ~participantsMask) { 
+			elog(WARNING, "Ignore transaction %llu from node %d because some of live nodes (%llx) are not participated in it (%llx)", 
+				 (long64)gtid->xid, gtid->node, liveMask, participantsMask);
+			result = false;
+		}
+	}
+	MtmUnlock();
+	return result;
+}
 
 /*
  * -------------------------------------------
@@ -2836,6 +2874,18 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomBoolVariable(
+		"multimaster.break_connection",
+		"Break connection with client when node is no online",
+		NULL,
+		&MtmBreakConnection,
+		false,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
 	DefineCustomBoolVariable(
 		"multimaster.major_node",
 		"Node which forms a majority in case of partitioning in cliques with equal number of nodes",
@@ -4358,8 +4408,8 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
 		}
 		if (!PrepareTransactionBlock(x->gid))
 		{
-			if (!MtmVolksWagenMode)
-				elog(WARNING, "Failed to prepare transaction %s", x->gid);
+			//if (!MtmVolksWagenMode)
+			elog(WARNING, "Failed to prepare transaction %s (%llu)", x->gid, (long64)x->xid);
 			/* ??? Should we do explicit rollback */
 		} else { 	
 			CommitTransactionCommand();

multimaster.h

@@ -347,7 +347,7 @@ extern timestamp_t MtmRefreshClusterStatusSchedule;
 extern void  MtmArbiterInitialize(void);
 extern void  MtmStartReceivers(void);
 extern void  MtmStartReceiver(int nodeId, bool dynamic);
-extern csn_t MtmTransactionSnapshot(TransactionId xid);
+extern csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, nodemask_t* participantsMask);
 extern csn_t MtmAssignCSN(void);
 extern csn_t MtmSyncClock(csn_t csn);
 extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot);
@@ -402,4 +402,6 @@ extern void MtmRollbackPreparedTransaction(int nodeId, char const* gid);
 extern bool MtmFilterTransaction(char* record, int size);
 extern void MtmPrecommitTransaction(char const* gid);
 extern char* MtmGucSerialize(void);
+extern bool MtmCheckParticipants(GlobalTransactionId* gtid, nodemask_t participants);
+
 #endif

pglogical_apply.c

@@ -337,22 +337,21 @@ process_remote_begin(StringInfo s)
 {
 	GlobalTransactionId gtid;
 	csn_t snapshot;
+	nodemask_t participantsMask;
 	int rc;
 
 	gtid.node = pq_getmsgint(s, 4); 
 	gtid.xid = pq_getmsgint(s, 4); 
 	snapshot = pq_getmsgint64(s);    
-
+	participantsMask = pq_getmsgint64(s);
 	Assert(gtid.node > 0);
 
-	MTM_LOG2("REMOTE begin node=%d xid=%d snapshot=%lld", gtid.node, gtid.xid, snapshot);
+	MTM_LOG2("REMOTE begin node=%d xid=%lu snapshot=%lld participantsMask=%llx", gtid.node, (long64)gtid.xid, snapshot, participantsMask);
 	MtmResetTransaction();		
-#if 1
-	if (BIT_CHECK(Mtm->disabledNodeMask, gtid.node-1)) { 
-		elog(WARNING, "Ignore transaction %llu from disabled node %d", (long64)gtid.xid, gtid.node);
+
+	if (!MtmCheckParticipants(&gtid, participantsMask)) {
 		return false;
 	}
-#endif
     SetCurrentStatementStartTimestamp();     
 	StartTransactionCommand();
     MtmJoinTransaction(&gtid, snapshot);

pglogical_proto.c

@@ -119,7 +119,8 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 					  ReorderBufferTXN *txn)
 {
 	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
-	csn_t csn = MtmTransactionSnapshot(txn->xid);
+	nodemask_t participantsMask;
+	csn_t csn = MtmDistributedTransactionSnapshot(txn->xid, MtmReplicationNodeId, &participantsMask);
 
 	if (!isRecovery && csn == INVALID_CSN) { 
 		MtmIsFilteredTxn = true;
@@ -136,6 +137,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 		pq_sendint(out, MtmNodeId, 4);
 		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
 		pq_sendint64(out, csn);
+		pq_sendint64(out, participantsMask);
 
 		MtmTransactionRecords = 0;
 	}
@@ -205,7 +207,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 			return;
 		}
 	} else { 
-		csn_t csn = MtmTransactionSnapshot(txn->xid);
+		nodemask_t partisipantsMask;
+		csn_t csn = MtmDistributedTransactionSnapshot(txn->xid, MtmReplicationNodeId, &partisipantsMask);
 		bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 
 		if (!isRecovery && csn == INVALID_CSN && (event != PGLOGICAL_ABORT_PREPARED || txn->origin_id != InvalidRepOriginId))

tests2/docker-entrypoint.sh

@@ -69,7 +69,7 @@ if [ "$1" = 'postgres' ]; then
 			multimaster.conn_strings = '$CONNSTRS'
 			multimaster.heartbeat_recv_timeout = 1100
 			multimaster.heartbeat_send_timeout = 250
-			multimaster.min_2pc_timeout = 100000
+			multimaster.min_2pc_timeout = 1000000
 		EOF
 
 		cat $PGDATA/postgresql.conf

tests2/lib/bank_client.py

@@ -185,8 +185,8 @@ def exec_tx(self, tx_block, aggname_prefix, conn_i):
                 # back to event loop and block it
                 yield from asyncio.sleep(0.01)
             except BaseException as e:
+                print('Catch exception: ', e)
                 agg.finish_tx(str(e).strip())
-                print('Catch exception ', str(e).strip())
                 # Give evloop some free time.
                 # In case of continuous excetions we can loop here without returning
                 # back to event loop and block it