reconnectMask maks modifications to state.c; disable MTM_RECOVERY_FINISH1 event

Author: kelvich

contrib/mmts/multimaster.c

@@ -1146,7 +1146,7 @@ bool MtmWatchdog(timestamp_t now)
 			{
 				MTM_ELOG(WARNING, "Heartbeat is not received from node %d during %d msec",
 					 i+1, (int)USEC_TO_MSEC(now - Mtm->nodes[i].lastHeartbeat));
-				MtmOnNodeDisconnect(i+1);
+				MtmStateProcessNeighborEvent(i+1, MTM_NEIGHBOR_HEARTBEAT_TIMEOUT);
 				allAlive = false;
 			}
 		}
@@ -1757,16 +1757,18 @@ void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot, nodemas
 	} else {
 		globalSnapshot = MtmTx.snapshot;
 	}
-	if (!TransactionIdIsValid(gtid->xid) && Mtm->status != MTM_RECOVERY)
-	{
-		/* In case of recovery InvalidTransactionId is passed */
-		MtmStateProcessEvent(MTM_RECOVERY_START1);
-	}
-	else if (Mtm->status == MTM_RECOVERY)
-	{
-		/* When recovery is completed we get normal transaction ID and switch to normal mode */
-		MtmStateProcessEvent(MTM_RECOVERY_FINISH1);
-	}
+
+
+	// if (!TransactionIdIsValid(gtid->xid) && Mtm->status != MTM_RECOVERY)
+	// {
+	// 	/* In case of recovery InvalidTransactionId is passed */
+	// 	MtmStateProcessEvent(MTM_RECOVERY_START1);
+	// }
+	// else if (Mtm->status == MTM_RECOVERY)
+	// {
+	// 	/* When recovery is completed we get normal transaction ID and switch to normal mode */
+	// 	MtmStateProcessEvent(MTM_RECOVERY_FINISH1);
+	// }
 }
 
 void  MtmSetCurrentTransactionGID(char const* gid)
@@ -2073,7 +2075,7 @@ bool MtmRecoveryCaughtUp(int nodeId, lsn_t walEndPtr)
 	MtmLock(LW_EXCLUSIVE);
 	if (MtmIsRecoveredNode(nodeId) && Mtm->nActiveTransactions == 0) {
 		if (BIT_CHECK(Mtm->originLockNodeMask, nodeId-1)) {
-			MtmStateProcessNeighborEvent(nodeId-1, MTM_NEIGHBOR_RECOVERY_CAUGHTUP);
+			MtmStateProcessNeighborEvent(nodeId, MTM_NEIGHBOR_RECOVERY_CAUGHTUP);
 		} else {
 			MTM_LOG1("Node %d is caught-up at WAL position %llx without locking cluster", nodeId, walEndPtr);
 			/* We are lucky: caught-up without locking cluster! */
@@ -2274,7 +2276,6 @@ void MtmRefreshClusterStatus()
 		disabled = ~newClique & (((nodemask_t)1 << Mtm->nAllNodes)-1) & ~Mtm->disabledNodeMask; /* new disabled nodes mask */
 
 		if (disabled) {
-			// timestamp_t now = MtmGetSystemTime();
 			for (i = 0, mask = disabled; mask != 0; i++, mask >>= 1) {
 				if (mask & 1) {
 					if ( i+1 == MtmNodeId )
@@ -2295,58 +2296,6 @@ void MtmRefreshClusterStatus()
 	}
 }
 
-
-/*
- * This function is called in case of non-recoverable connection failure with this node.
- * Non-recoverable means that connections can not be reestablish using specified number of attempts.
- * It sets bit in connectivity mask and register delayed refresh of cluster status which build connectivity matrix
- * and determine clique of connected nodes. Timeout here is needed to allow all nodes to exchanges their connectivity masks (them
- * are sent together with any arbiter message, including heartbeats.
- */
-void MtmOnNodeDisconnect(int nodeId)
-{
-	timestamp_t now = MtmGetSystemTime();
-	if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
-	{
-		/* Node is already disabled */
-		return;
-	}
-	if (Mtm->nodes[nodeId-1].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) > now)
-	{
-		/* Avoid false detection of node failure and prevent node status blinking */
-		return;
-	}
-	MtmLock(LW_EXCLUSIVE);
-	BIT_SET(SELF_CONNECTIVITY_MASK, nodeId-1);
-	BIT_SET(Mtm->reconnectMask, nodeId-1);
-	MTM_ELOG(LOG, "Disconnect node %d connectivity mask %llx",
-		 nodeId, SELF_CONNECTIVITY_MASK);
-	MtmUnlock();
-}
-
-/*
- * This method is called when connection with node is reestablished
- */
-void MtmOnNodeConnect(int nodeId)
-{
-	MtmLock(LW_EXCLUSIVE);
-	MTM_ELOG(LOG, "Connect node %d connectivity mask %llx", nodeId, SELF_CONNECTIVITY_MASK);
-	BIT_CLEAR(SELF_CONNECTIVITY_MASK, nodeId-1);
-	BIT_SET(Mtm->reconnectMask, nodeId-1); /* force sender to reestablish connection and send heartbeat */
-	MtmUnlock();
-}
-
-/*
- * Set reconnect mask to force reconnection attempt to the node
- */
-void MtmReconnectNode(int nodeId)
-{
-	MtmLock(LW_EXCLUSIVE);
-	MTM_ELOG(LOG, "Reconnect node %d connectivity mask %llx", nodeId, SELF_CONNECTIVITY_MASK);
-	BIT_SET(Mtm->reconnectMask, nodeId-1);
-	MtmUnlock();
-}
-
 /*
  * -------------------------------------------
  * Node initialization
@@ -3659,11 +3608,9 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	} else {
 		MTM_LOG1("Node %d start logical replication to node %d in normal mode", MtmNodeId, MtmReplicationNodeId);
 	}
-	if (!BIT_CHECK(Mtm->pglogicalSenderMask, MtmReplicationNodeId-1)) {
-		MTM_ELOG(LOG, "Start %d senders and %d receivers from %d cluster status %s", Mtm->nSenders+1, Mtm->nReceivers, Mtm->nLiveNodes-1, MtmNodeStatusMnem[Mtm->status]);
-		BIT_SET(Mtm->pglogicalSenderMask, MtmReplicationNodeId-1);
-		MtmStateProcessEvent(MTM_WAL_SENDER_START);
-	}
+
+	MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START);
+
 	BIT_SET(Mtm->reconnectMask, MtmReplicationNodeId-1); /* arbiter should try to reestablish connection with this node */
 	MtmUnlock();
 	on_shmem_exit(MtmOnProcExit, 0);

contrib/mmts/multimaster.h

@@ -401,11 +401,8 @@ extern void  MtmLockNode(int nodeId, LWLockMode mode);
 extern bool  MtmTryLockNode(int nodeId, LWLockMode mode);
 extern void  MtmUnlockNode(int nodeId);
 extern void  MtmStopNode(int nodeId, bool dropSlot);
-extern void  MtmReconnectNode(int nodeId);
 extern void  MtmRecoverNode(int nodeId);
 extern void  MtmResumeNode(int nodeId);
-extern void  MtmOnNodeDisconnect(int nodeId);
-extern void  MtmOnNodeConnect(int nodeId);
 extern void  MtmWakeUpBackend(MtmTransState* ts);
 extern void  MtmSleep(timestamp_t interval);
 extern void  MtmAbortTransaction(MtmTransState* ts);

contrib/mmts/pglogical_receiver.c

@@ -368,7 +368,7 @@ pglogical_receiver_main(Datum main_arg)
 		PQclear(res);
 		resetPQExpBuffer(query);
 
-		MtmStateProcessNeighborEvent(nodeId, MTM_NEIGHBOR_RECEIVER_START);
+		MtmStateProcessNeighborEvent(nodeId, MTM_NEIGHBOR_WAL_RECEIVER_START);
 
 		while (!got_sigterm)
 		{

contrib/mmts/state.c

@@ -6,7 +6,9 @@
 char const* const MtmNeighborEventMnem[] =
 {
 	"MTM_NEIGHBOR_CLIQUE_DISABLE",
-	"MTM_NEIGHBOR_RECEIVER_START",
+	"MTM_NEIGHBOR_WAL_RECEIVER_START",
+	"MTM_NEIGHBOR_WAL_SENDER_START",
+	"MTM_NEIGHBOR_HEARTBEAT_TIMEOUT",
 	"MTM_NEIGHBOR_RECOVERY_CAUGHTUP"
 };
 
@@ -23,9 +25,6 @@ char const* const MtmEventMnem[] =
 	"MTM_RECOVERY_FINISH1",
 	"MTM_RECOVERY_FINISH2",
 
-	"MTM_WAL_RECEIVER_START",
-	"MTM_WAL_SENDER_START",
-
 	"MTM_NONRECOVERABLE_ERROR"
 };
 
@@ -37,39 +36,61 @@ static void MtmSwitchClusterMode(MtmNodeStatus mode);
 void
 MtmStateProcessNeighborEvent(int node_id, MtmNeighborEvent ev)
 {
-	MTM_LOG1("[STATE] Processing %s", MtmNeighborEventMnem[ev]);
+	MTM_LOG1("[STATE] Processing %s for node %u", MtmNeighborEventMnem[ev], node_id);
 
 	switch(ev)
 	{
 		case MTM_NEIGHBOR_CLIQUE_DISABLE:
 			MtmDisableNode(node_id);
 			break;
 
-		case MTM_NEIGHBOR_RECEIVER_START:
-			/*
-			* This functions is called by pglogical receiver main function when receiver background worker is started.
-			* We switch to ONLINE mode when all receivers are connected.
-			* As far as background worker can be restarted multiple times, use node bitmask.
-			*/
+		case MTM_NEIGHBOR_HEARTBEAT_TIMEOUT:
+			MtmLock(LW_EXCLUSIVE);
+			BIT_SET(SELF_CONNECTIVITY_MASK, node_id - 1);
+			BIT_SET(Mtm->reconnectMask, node_id - 1);
+			MtmUnlock();
+			break;
+
+		case MTM_NEIGHBOR_WAL_RECEIVER_START:
 			MtmLock(LW_EXCLUSIVE);
-			if (!BIT_CHECK(Mtm->pglogicalReceiverMask, node_id)) {
-				BIT_SET(Mtm->pglogicalReceiverMask, node_id);
-				if (BIT_CHECK(Mtm->disabledNodeMask, node_id)) {
+			if (!BIT_CHECK(Mtm->pglogicalReceiverMask, node_id - 1)) {
+				BIT_SET(Mtm->pglogicalReceiverMask, node_id - 1);
+				if (BIT_CHECK(Mtm->disabledNodeMask, node_id - 1)) {
 					MtmEnableNode(node_id);
 				}
-				MtmStateProcessEvent(MTM_WAL_RECEIVER_START);
+
+				if (++Mtm->nReceivers == Mtm->nLiveNodes-1 && Mtm->nSenders == Mtm->nLiveNodes-1
+					&& (Mtm->status == MTM_RECOVERED || Mtm->status == MTM_CONNECTED))
+				{
+					BIT_CLEAR(Mtm->originLockNodeMask, MtmNodeId-1); /* recovery is completed: release cluster lock */
+					MtmSwitchClusterMode(MTM_ONLINE);
+				}
 			}
 			MtmUnlock();
 			break;
 
+		case MTM_NEIGHBOR_WAL_SENDER_START:
+			if (!BIT_CHECK(Mtm->pglogicalSenderMask, node_id - 1)) {
+				BIT_SET(Mtm->pglogicalSenderMask, node_id - 1);
+				if (++Mtm->nSenders == Mtm->nLiveNodes-1 && Mtm->nReceivers == Mtm->nLiveNodes-1
+						&& (Mtm->status == MTM_RECOVERED || Mtm->status == MTM_CONNECTED))
+					{
+						/* All logical replication connections from and to this node are established, so we can switch cluster to online mode */
+						BIT_CLEAR(Mtm->originLockNodeMask, MtmNodeId-1); /* recovery is completed: release cluster lock */
+						MtmSwitchClusterMode(MTM_ONLINE);
+					}
+			}
+			break;
+
 		case MTM_NEIGHBOR_RECOVERY_CAUGHTUP:
-			Assert(BIT_CHECK(Mtm->disabledNodeMask, node_id));
-			BIT_CLEAR(Mtm->originLockNodeMask, node_id);
-			BIT_CLEAR(Mtm->disabledNodeMask, node_id);
-			BIT_SET(Mtm->recoveredNodeMask, node_id);
+			Assert(BIT_CHECK(Mtm->disabledNodeMask, node_id - 1));
+			BIT_CLEAR(Mtm->originLockNodeMask, node_id - 1);
+			BIT_CLEAR(Mtm->disabledNodeMask, node_id - 1);
+			BIT_SET(Mtm->recoveredNodeMask, node_id - 1);
 			Mtm->nLiveNodes += 1;
 			MtmCheckQuorum();
 			break;
+
 	}
 }
 
@@ -114,7 +135,7 @@ MtmStateProcessEvent(MtmEvent ev)
 			if (Mtm->nLiveNodes < Mtm->nAllNodes/2+1)
 			{
 				/* no quorum */
-				MTM_ELOG(WARNING, "Node is out of quorum: only %d nodes of %d are accessible", Mtm->nLiveNodes, Mtm->nAllNodes);
+				// MTM_ELOG(WARNING, "Node is out of quorum: only %d nodes of %d are accessible", Mtm->nLiveNodes, Mtm->nAllNodes);
 				MtmSwitchClusterMode(MTM_IN_MINORITY);
 			}
 			else if (Mtm->status == MTM_INITIALIZATION)
@@ -185,28 +206,6 @@ MtmStateProcessEvent(MtmEvent ev)
 			break;
 
 
-		case MTM_WAL_RECEIVER_START:
-			MTM_ELOG(LOG, "[STATE] Start %d receivers and %d senders from %d cluster status %s", Mtm->nReceivers+1, Mtm->nSenders, Mtm->nLiveNodes-1, MtmNodeStatusMnem[Mtm->status]);
-			if (++Mtm->nReceivers == Mtm->nLiveNodes-1 && Mtm->nSenders == Mtm->nLiveNodes-1
-				&& (Mtm->status == MTM_RECOVERED || Mtm->status == MTM_CONNECTED))
-			{
-				BIT_CLEAR(Mtm->originLockNodeMask, MtmNodeId-1); /* recovery is completed: release cluster lock */
-				MtmSwitchClusterMode(MTM_ONLINE);
-			}
-			break;
-
-
-		case MTM_WAL_SENDER_START:
-			if (++Mtm->nSenders == Mtm->nLiveNodes-1 && Mtm->nReceivers == Mtm->nLiveNodes-1
-				&& (Mtm->status == MTM_RECOVERED || Mtm->status == MTM_CONNECTED))
-			{
-				/* All logical replication connections from and to this node are established, so we can switch cluster to online mode */
-				BIT_CLEAR(Mtm->originLockNodeMask, MtmNodeId-1); /* recovery is completed: release cluster lock */
-				MtmSwitchClusterMode(MTM_ONLINE);
-			}
-			break;
-
-
 		case MTM_NONRECOVERABLE_ERROR:
 			// MTM_ELOG(WARNING, "Node is excluded from cluster because of non-recoverable error %d, %s, pid=%u",
 				// edata->sqlerrcode, edata->message, getpid());
@@ -234,13 +233,10 @@ MtmSwitchClusterMode(MtmNodeStatus mode)
  */
 void MtmDisableNode(int nodeId)
 {
-	timestamp_t now = MtmGetSystemTime();
-	MTM_ELOG(WARNING, "Disable node %d at xlog position %llx, last status change time %d msec ago", nodeId, (long64)GetXLogInsertRecPtr(),
-		 (int)USEC_TO_MSEC(now - Mtm->nodes[nodeId-1].lastStatusChangeTime));
 	BIT_SET(Mtm->disabledNodeMask, nodeId-1);
 	Mtm->nConfigChanges += 1;
 	Mtm->nodes[nodeId-1].timeline += 1;
-	Mtm->nodes[nodeId-1].lastStatusChangeTime = now;
+	Mtm->nodes[nodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	if (nodeId != MtmNodeId) {
 		Mtm->nLiveNodes -= 1;
@@ -269,7 +265,6 @@ void MtmEnableNode(int nodeId)
 		if (nodeId != MtmNodeId) {
 			Mtm->nLiveNodes += 1;
 		}
-		// MTM_ELOG(WARNING, "Enable node %d at xlog position %llx", nodeId, (long64)GetXLogInsertRecPtr());
 	}
 	MtmCheckQuorum();
 }
@@ -282,15 +277,49 @@ static void MtmCheckQuorum(void)
 {
 	int nVotingNodes = MtmGetNumberOfVotingNodes();
 
-	if (Mtm->nLiveNodes >= nVotingNodes/2+1 || (Mtm->nLiveNodes == (nVotingNodes+1)/2 && MtmMajorNode)) { /* have quorum */
-		if (Mtm->status == MTM_IN_MINORITY) {
-			// MTM_LOG1("Node is in majority: disabled mask %llx", Mtm->disabledNodeMask);
+	if (Mtm->nLiveNodes >= nVotingNodes/2+1 || (Mtm->nLiveNodes == (nVotingNodes+1)/2 && MtmMajorNode))
+	{
+		if (Mtm->status == MTM_IN_MINORITY)
 			MtmSwitchClusterMode(MTM_ONLINE);
-		}
-	} else {
-		if (Mtm->status == MTM_ONLINE) { /* out of quorum */
-			// MTM_ELOG(WARNING, "Node is in minority: disabled mask %llx", Mtm->disabledNodeMask);
+	}
+	else
+	{
+		if (Mtm->status == MTM_ONLINE)
 			MtmSwitchClusterMode(MTM_IN_MINORITY);
-		}
 	}
 }
+
+void MtmOnNodeDisconnect(int nodeId)
+{
+	if (BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId-1))
+		return;
+
+	MTM_LOG1("[STATE] NodeDisconnect for node %u", nodeId);
+
+	MtmLock(LW_EXCLUSIVE);
+	BIT_SET(SELF_CONNECTIVITY_MASK, nodeId-1);
+	BIT_SET(Mtm->reconnectMask, nodeId-1);
+	MtmUnlock();
+}
+
+void MtmOnNodeConnect(int nodeId)
+{
+	if (!BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId-1))
+		return;
+
+	MTM_LOG1("[STATE] NodeConnect for node %u", nodeId);
+
+	MtmLock(LW_EXCLUSIVE);
+	BIT_CLEAR(SELF_CONNECTIVITY_MASK, nodeId-1);
+	BIT_SET(Mtm->reconnectMask, nodeId-1);
+	MtmUnlock();
+}
+
+void MtmReconnectNode(int nodeId)
+{
+	MTM_LOG1("[STATE] ReconnectNode for node %u", nodeId);
+	MtmLock(LW_EXCLUSIVE);
+	BIT_SET(Mtm->reconnectMask, nodeId-1);
+	MtmUnlock();
+}
+

contrib/mmts/state.h

@@ -2,7 +2,9 @@
 typedef enum
 {
 	MTM_NEIGHBOR_CLIQUE_DISABLE,
-	MTM_NEIGHBOR_RECEIVER_START,
+	MTM_NEIGHBOR_WAL_RECEIVER_START,
+	MTM_NEIGHBOR_WAL_SENDER_START,
+	MTM_NEIGHBOR_HEARTBEAT_TIMEOUT,
 	MTM_NEIGHBOR_RECOVERY_CAUGHTUP
 } MtmNeighborEvent;
 
@@ -16,8 +18,6 @@ typedef enum
 	MTM_RECOVERY_START2,
 	MTM_RECOVERY_FINISH1,
 	MTM_RECOVERY_FINISH2,
-	MTM_WAL_RECEIVER_START,
-	MTM_WAL_SENDER_START,
 	MTM_NONRECOVERABLE_ERROR
 } MtmEvent;
 
@@ -26,3 +26,7 @@ extern void MtmStateProcessEvent(MtmEvent ev);
 extern void MtmDisableNode(int nodeId);
 extern void MtmEnableNode(int nodeId);
 
+extern void MtmOnNodeDisconnect(int nodeId);
+extern void MtmOnNodeConnect(int nodeId);
+extern void MtmReconnectNode(int nodeId);
+