Disconnect node only on heartbeat timeout; One disabled state instead of several

Author: kelvich

arbiter.c

@@ -190,7 +190,6 @@ static void MtmDisconnect(int node)
 	MtmUnregisterSocket(sockets[node]);
 	pg_closesocket(sockets[node], MtmUseRDMA);
 	sockets[node] = -1;
-	MtmOnNodeDisconnect(node+1);
 }
 
 static int MtmWaitSocket(int sd, bool forWrite, timestamp_t timeoutMsec)
@@ -541,9 +540,6 @@ static void MtmOpenConnections()
 	for (i = 0; i < nNodes; i++) {
 		if (i+1 != MtmNodeId && i < Mtm->nAllNodes) { 
 			sockets[i] = MtmConnectSocket(i, Mtm->nodes[i].con.arbiterPort);
-			if (sockets[i] < 0) { 
-				MtmOnNodeDisconnect(i+1);
-			} 
 		}
 	}
 	MtmStateProcessEvent(MTM_ARBITER_RECEIVER_START);
@@ -579,7 +575,6 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 			}
 			sockets[node] = MtmConnectSocket(node, Mtm->nodes[node].con.arbiterPort);
 			if (sockets[node] < 0) { 
-				MtmOnNodeDisconnect(node+1);
 				result = false;
 				break;
 			}

multimaster.c

@@ -218,15 +218,10 @@ static TransactionManager MtmTM =
 
 char const* const MtmNodeStatusMnem[] =
 {
-	"Initialization",
-	"Offline",
-	"Connecting",
-	"Online",
+	"Disabled",
 	"Recovery",
 	"Recovered",
-	"InMinor",
-	"OutOfClique",
-	"OutOfService"
+	"Online"
 };
 
 char const* const MtmTxnStatusMnem[] =
@@ -1961,7 +1956,7 @@ void MtmPollStatusOfPreparedTransactionsForDisabledNode(int disabledNodeId)
 				MtmBroadcastPollMessage(ts);
 			}
 		} else {
-			MTM_LOG1("Skip transaction %s (%llu) with status %s gtid.node=%d gtid.xid=%llu votedMask=%llx",
+			MTM_LOG2("Skip transaction %s (%llu) with status %s gtid.node=%d gtid.xid=%llu votedMask=%llx",
 					 ts->gid, (long64)ts->xid, MtmTxnStatusMnem[ts->status], ts->gtid.node, (long64)ts->gtid.xid, ts->votedMask);
 		}
 	}
@@ -2367,7 +2362,7 @@ static void MtmInitialize()
 	if (!found)
 	{
 		MemSet(Mtm, 0, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1));
-		Mtm->status = MTM_IN_MINORITY; //MTM_INITIALIZATION;
+		Mtm->status = MTM_DISABLED; //MTM_INITIALIZATION;
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);
 		Mtm->csn = MtmGetCurrentTime();
@@ -2774,7 +2769,7 @@ _PG_init(void)
 		"Timeout in milliseconds of sending heartbeat messages",
 		"Period of broadcasting heartbeat messages by arbiter to all nodes",
 		&MtmHeartbeatSendTimeout,
-		1000,
+		200,
 		1,
 		INT_MAX,
 		PGC_BACKEND,
@@ -2789,7 +2784,7 @@ _PG_init(void)
 		"Timeout in milliseconds of receiving heartbeat messages",
 		"If no heartbeat message is received from node within this period, it assumed to be dead",
 		&MtmHeartbeatRecvTimeout,
-		10000,
+		1000,
 		1,
 		INT_MAX,
 		PGC_BACKEND,

multimaster.h

@@ -130,15 +130,12 @@ typedef enum
 
 typedef enum
 {
-	MTM_INITIALIZATION, /* Initial status */
-	MTM_OFFLINE,        /* Node is excluded from cluster */
-	MTM_CONNECTED,      /* Arbiter is established connections with other nodes */
-	MTM_ONLINE,         /* Ready to receive client's queries */
+	MTM_DISABLED,       /* Node disabled */
 	MTM_RECOVERY,       /* Node is in recovery process */
-	MTM_RECOVERED,      /* Node is recovered by is not yet switched to ONLINE because not all sender/receivers are restarted */
-	MTM_IN_MINORITY,    /* Node is out of quorum */
-	MTM_OUT_OF_CLIQUE,  /* Node is out of cluster by clique detector */
-	MTM_OUT_OF_SERVICE  /* Node is not available to to critical, non-recoverable error */
+	MTM_RECOVERED,      /* Node is recovered by is not yet switched to ONLINE because
+						 * not all sender/receivers are restarted
+						 */
+	MTM_ONLINE          /* Ready to receive client's queries */
 } MtmNodeStatus;
 
 typedef enum

pglogical_receiver.c

@@ -404,7 +404,7 @@ pglogical_receiver_main(Datum main_arg)
 			if (rc & WL_POSTMASTER_DEATH)
 				proc_exit(1);
 
-			if (Mtm->status == MTM_OFFLINE || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId))
+			if (Mtm->status == MTM_DISABLED || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId))
 			{
 				ereport(LOG, (MTM_ERRMSG("%s: restart WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
 				break;

state.c

@@ -60,14 +60,6 @@ MtmSetClusterStatus(MtmNodeStatus status)
 	MTM_LOG1("[STATE]   Switching status from %s to %s status",
 		MtmNodeStatusMnem[Mtm->status], MtmNodeStatusMnem[status]);
 
-	/* Do some actions on specific status transitions */
-	if (status == MTM_IN_MINORITY)
-	{
-		Mtm->recoverySlot = 0;
-		Mtm->pglogicalReceiverMask = 0;
-		Mtm->pglogicalSenderMask = 0;
-	}
-
 	Mtm->status = status;
 }
 
@@ -88,27 +80,21 @@ MtmCheckState(void)
 		maskToString(Mtm->pglogicalSenderMask, Mtm->nAllNodes),
 		Mtm->nAllNodes);
 
-	/* ANY -> MTM_IN_MINORITY */
-	if (nConnected < Mtm->nAllNodes/2+1)
+	/* ANY -> MTM_DISABLED */
+	if ( nConnected < Mtm->nAllNodes/2+1 ||
+		!BIT_CHECK(Mtm->clique, MtmNodeId-1) )
 	{
 		BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
-		MtmSetClusterStatus(MTM_IN_MINORITY);
-		return;
-	}
-
-	/* ANY -> CLIQUE_DISABLE */
-	if (!BIT_CHECK(Mtm->clique, MtmNodeId-1))
-	{
-		/* Should be already disabled by clique detector */
-		// Assert(BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId-1));
-		MtmSetClusterStatus(MTM_OUT_OF_CLIQUE);
+		Mtm->recoverySlot = 0;
+		Mtm->pglogicalReceiverMask = 0;
+		Mtm->pglogicalSenderMask = 0;
+		MtmSetClusterStatus(MTM_DISABLED);
 		return;
 	}
 
 	switch (Mtm->status)
 	{
-		case MTM_IN_MINORITY:
-		case MTM_OUT_OF_CLIQUE:
+		case MTM_DISABLED:
 			if ( (nConnected >= Mtm->nAllNodes/2+1) && /* majority */
 				 BIT_CHECK(Mtm->clique, MtmNodeId-1) ) /* in clique */
 			{
@@ -173,7 +159,7 @@ MtmStateProcessNeighborEvent(int node_id, MtmNeighborEvent ev)
 
 		case MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED:
 			BIT_SET(Mtm->pglogicalSenderMask, node_id - 1);
-			MtmEnableNode(node_id);
+			MtmEnableNode(node_id); /// XXXX ?
 			break;
 
 		case MTM_NEIGHBOR_RECOVERY_CAUGHTUP:
@@ -293,14 +279,12 @@ void MtmOnNodeDisconnect(int nodeId)
 	if (BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId-1))
 		return;
 
-	MtmDisableNode(nodeId);
-
 	MTM_LOG1("[STATE] Node %i: disconnected", nodeId);
 
 	MtmLock(LW_EXCLUSIVE);
 	BIT_SET(SELF_CONNECTIVITY_MASK, nodeId-1);
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
-
+	// MtmDisableNode(nodeId);
 	MtmCheckState();
 	MtmUnlock();