Handle heartbeat timesouts in arbiter receiver

Author: knizhnik

contrib/mmts/arbiter.c

@@ -675,6 +675,8 @@ static void MtmTransReceiver(Datum arg)
 	int nResponses;
 	int i, j, n, rc;
 	MtmBuffer* rxBuffer = (MtmBuffer*)palloc(sizeof(MtmBuffer)*nNodes);
+	timestamp_t lastHeartbeatCheck = MtmGetSystemTime();
+	timestamp_t now;
 
 #if USE_EPOLL
 	struct epoll_event* events = (struct epoll_event*)palloc(sizeof(struct epoll_event)*nNodes);
@@ -698,7 +700,7 @@ static void MtmTransReceiver(Datum arg)
 
 	while (!stop) {
 #if USE_EPOLL
-        n = epoll_wait(epollfd, events, nNodes, USEC_TO_MSEC(MtmKeepaliveTimeout));
+        n = epoll_wait(epollfd, events, nNodes, MtmHeartbeatRecvTimeout);
 		if (n < 0) { 
 			if (errno == EINTR) { 
 				continue;
@@ -717,8 +719,8 @@ static void MtmTransReceiver(Datum arg)
 		do { 
 			struct timeval tv;
 			events = inset;
-			tv.tv_sec = MtmKeepaliveTimeout/USECS_PER_SEC;
-			tv.tv_usec = MtmKeepaliveTimeout%USECS_PER_SEC;
+			tv.tv_sec = MtmHeartbeatRecvTimeout/1000;
+			tv.tv_usec = MtmHeartbeatRecvTimeout%1000*1000;
 			do { 
 				n = select(max_fd+1, &events, NULL, NULL, &tv);
 			} while (n < 0 && errno == EINTR);
@@ -855,8 +857,13 @@ static void MtmTransReceiver(Datum arg)
 				}
 			}
 		}
+		now = MtmGetSystemTime();
+		if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
+			MtmWatchdog();
+			lastHeartbeatCheck = now;
+		}
 		if (n == 0 && Mtm->disabledNodeMask != 0) { 
-			/* If timeout is expired and there are didabled nodes, then recheck cluster's state */
+			/* If timeout is expired and there are disabled nodes, then recheck cluster's state */
 			MtmRefreshClusterStatus(false);
 		}
 	}

contrib/mmts/multimaster.c

@@ -187,7 +187,7 @@ int   MtmReplicationNodeId;
 int   MtmArbiterPort;
 int   MtmConnectAttempts;
 int   MtmConnectTimeout;
-int   MtmKeepaliveTimeout;
+int   MtmRaftPollDelay;
 int   MtmReconnectAttempts;
 int   MtmNodeDisableDelay;
 int   MtmTransSpillThreshold;
@@ -747,7 +747,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 /*
  * Check heartbeats
  */
-static void MtmWatchdog()
+void MtmWatchdog(void)
 {
 	int i, n = Mtm->nAllNodes;
 	timestamp_t now = MtmGetSystemTime();
@@ -795,33 +795,27 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		MtmResetTransaction(x);
 	} else { 
 		time_t transTimeout = Max(Mtm2PCMinTimeout, (ts->csn - ts->snapshot)*Mtm2PCPrepareRatio/100000); /* usec->msec and percents */ 
-		time_t timeout = Min(transTimeout, MtmHeartbeatRecvTimeout);
-		timestamp_t deadline = MtmGetSystemTime() + MSEC_TO_USEC(transTimeout);
 		int result = 0;
 		int nConfigChanges = Mtm->nConfigChanges;
 		/* wait votes from all nodes */
-		while (!ts->votingCompleted) {
+		while (!ts->votingCompleted && !(result & WL_TIMEOUT)) 
+		{
 			MtmUnlock();
-			//MtmWatchdog();
+			MtmWatchdog();
 			if (ts->status == TRANSACTION_STATUS_ABORTED) {
 				elog(WARNING, "Transaction %d(%s) is aborted by watchdog", x->xid, x->gid);				
 				x->status = TRANSACTION_STATUS_ABORTED;
 				return;
 			}
-			result = WaitLatch(&MyProc->procLatch, WL_LATCH_SET|WL_TIMEOUT, timeout);
+			result = WaitLatch(&MyProc->procLatch, WL_LATCH_SET|WL_TIMEOUT, transTimeout);
 			if (result & WL_LATCH_SET) { 
 				ResetLatch(&MyProc->procLatch);			
-			} else if (result & WL_TIMEOUT) { 
-				if (MtmGetSystemTime() > deadline) { 
-					MtmLock(LW_SHARED);
-					break;
-				}
 			} 
 			MtmLock(LW_SHARED);
 		}
 		if (!ts->votingCompleted) { 
 			ts->status = TRANSACTION_STATUS_ABORTED;
-			elog(WARNING, "Transaction is aborted because of %d msec timeout expiration, prepare time %d msec", (int)timeout, (int)USEC_TO_MSEC(ts->csn - x->snapshot));
+			elog(WARNING, "Transaction is aborted because of %d msec timeout expiration, prepare time %d msec", (int)transTimeout, (int)USEC_TO_MSEC(ts->csn - x->snapshot));
 		} else if (nConfigChanges != Mtm->nConfigChanges) {
 			ts->status = TRANSACTION_STATUS_ABORTED;
 			elog(WARNING, "Transaction is aborted because cluster configuration is changed during commit");
@@ -1369,8 +1363,7 @@ void MtmOnNodeDisconnect(int nodeId)
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
 	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false);
 
-	/* Wait more than socket KEEPALIVE timeout to let other nodes update their statuses */
-	MtmSleep(MtmKeepaliveTimeout);
+	MtmSleep(MtmRaftPollDelay);
 
 	if (!MtmRefreshClusterStatus(false)) { 
 		MtmLock(LW_EXCLUSIVE);
@@ -1970,10 +1963,10 @@ _PG_init(void)
 	);
 
 	DefineCustomIntVariable(
-		"multimaster.keepalive_timeout",
-		"Multimaster keepalive interval for sockets",
+		"multimaster.raft_poll_delay",
+		"Multimaster delay of polling cluster state from Raftable after updating local node status",
 		"Timeout in microseconds before polling state of nodes",
-		&MtmKeepaliveTimeout,
+		&MtmRaftPollDelay,
 		1000000,
 		1,
 		INT_MAX,

contrib/mmts/multimaster.h

@@ -217,7 +217,7 @@ extern char* MtmDatabaseName;
 extern int   MtmConnectAttempts;
 extern int   MtmConnectTimeout;
 extern int   MtmReconnectAttempts;
-extern int   MtmKeepaliveTimeout;
+extern int   MtmRaftPollDelay;
 extern int   MtmNodeDisableDelay;
 extern int   MtmTransSpillThreshold;
 extern int   MtmHeartbeatSendTimeout;
@@ -266,6 +266,8 @@ extern void  MtmRecoveryCompleted(void);
 extern void  MtmMakeTableLocal(char* schema, char* name);
 extern void  MtmHandleApplyError(void);
 extern void  MtmUpdateLsnMapping(int nodeId, XLogRecPtr endLsn);
-extern 	XLogRecPtr MtmGetFlushPosition(int nodeId);
+extern XLogRecPtr MtmGetFlushPosition(int nodeId);
+extern void MtmWatchdog(void);
+
 
 #endif

contrib/raftable/state.c

@@ -6,7 +6,7 @@
 #include "blockmem.h"
 #include "state.h"
 
-#define RAFTABLE_BLOCK_MEM (1024 * 1024)
+#define RAFTABLE_BLOCK_MEM (8*1024 * 1024)
 #define RAFTABLE_HASH_SIZE (127)
 
 typedef struct State {