Fix timer initialziation

Author: knizhnik

arbiter.c

@@ -103,6 +103,7 @@ typedef struct
 static int*      sockets;
 static int       gateway;
 static bool      send_heartbeat;
+static TimeoutId heartbeat_timer;
 
 static void MtmTransSender(Datum arg);
 static void MtmTransReceiver(Datum arg);
@@ -526,7 +527,7 @@ static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, Mt
 		buf->data[buf->used].csn  = ts->csn;
 	} else { 
 		buf->data[buf->used].code = MSG_HEARTBEAT;
-		MTM_LOG3("Send HEARTBEAT message to node %d from node %d\n", node+1, MtmNodeId);
+		MTM_LOG3("Send HEARTBEAT to node %d from node %d at %ld\n", node+1, MtmNodeId, USEC_TO_MSEC(MtmGetSystemTime()));
 	}
 	buf->data[buf->used].node = MtmNodeId;
 	buf->data[buf->used].disabledNodeMask = Mtm->disabledNodeMask;
@@ -540,8 +541,9 @@ static void MtmBroadcastMessage(MtmBuffer* txBuffer, MtmTransState* ts)
 	int n = 1;
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
-		if (!BIT_CHECK(Mtm->disabledNodeMask, i) && (ts == NULL || TransactionIdIsValid(ts->xids[i]))) { 
-			Assert(i+1 != MtmNodeId);
+		if (i+1 != MtmNodeId && !BIT_CHECK(Mtm->disabledNodeMask, i) 
+		    && (ts == NULL || TransactionIdIsValid(ts->xids[i]))) 
+		{ 
 			MtmAppendBuffer(txBuffer, ts ? ts->xids[i] : InvalidTransactionId, i, ts);
 			n += 1;
 		}
@@ -553,6 +555,7 @@ static void MtmSendHeartbeat()
 {
 	send_heartbeat = true;
 	PGSemaphoreUnlock(&Mtm->votingSemaphore);
+        //enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
 }
 
 
@@ -561,15 +564,19 @@ static void MtmTransSender(Datum arg)
 	sigset_t sset;
 	int nNodes = MtmMaxNodes;
 	int i;
+
 	MtmBuffer* txBuffer = (MtmBuffer*)palloc(sizeof(MtmBuffer)*nNodes);
 
+	InitializeTimeouts();
+
 	signal(SIGINT, SetStop);
 	signal(SIGQUIT, SetStop);
 	signal(SIGTERM, SetStop);
 	sigfillset(&sset);
 	sigprocmask(SIG_UNBLOCK, &sset, NULL);
 
-	RegisterTimeout(USER_TIMEOUT, MtmSendHeartbeat);
+	heartbeat_timer = RegisterTimeout(USER_TIMEOUT, MtmSendHeartbeat);
+	enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
 
 	MtmOpenConnections();
 
@@ -584,6 +591,7 @@ static void MtmTransSender(Datum arg)
 
 		if (send_heartbeat) {
 			send_heartbeat = false;
+			enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
 			MtmBroadcastMessage(txBuffer, NULL);
 		}			
 		/* 
@@ -725,7 +733,8 @@ static void MtmTransReceiver(Datum arg)
 					Mtm->nodes[msg->node-1].oldestSnapshot = msg->oldestSnapshot;
 					Mtm->nodes[msg->node-1].lastHeartbeat = MtmGetSystemTime();
 
-					if (msg->code == MSG_HEARTBEAT) { 
+					if (msg->code == MSG_HEARTBEAT) {
+						MTM_LOG3("Receive HEARTBEAT from node %d at %ld", msg->node, USEC_TO_MSEC(MtmGetSystemTime())); 
 						continue;
 					}
 					ts = (MtmTransState*)hash_search(MtmXid2State, &msg->dxid, HASH_FIND, NULL);

multimaster.c

@@ -755,8 +755,8 @@ static void MtmWatchdog()
 			if (Mtm->nodes[i].lastHeartbeat != 0
 				&& now > Mtm->nodes[i].lastHeartbeat + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) 
 			{ 
-				elog(WARNING, "Disable node %d because last heartbeat was received %d msec ago", 
-					 i+1, (int)USEC_TO_MSEC(now - Mtm->nodes[i].lastHeartbeat));
+				elog(WARNING, "Disable node %d because last heartbeat was received %d msec ago (%ld)", 
+					 i+1, (int)USEC_TO_MSEC(now - Mtm->nodes[i].lastHeartbeat), USEC_TO_MSEC(now));
 				MtmOnNodeDisconnect(i+1);				
 			}
 		}
@@ -801,7 +801,7 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 		/* wait votes from all nodes */
 		while (!ts->votingCompleted) {
 			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;
@@ -1693,7 +1693,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,
-		2000,
+		100000,
 		1,
 		INT_MAX,
 		PGC_BACKEND,
@@ -1752,7 +1752,7 @@ _PG_init(void)
 		"Minamal amount of time (milliseconds) to wait 2PC confirmation from all nodes",
 		"Timeout for 2PC is calculated as MAX(prepare_time*2pc_prepare_ratio/100,2pc_min_timeout)",
 		&Mtm2PCMinTimeout,
-		100000, /* 100 seconds */
+		10000, /* 100 seconds */
 		0,
 		INT_MAX,
 		PGC_BACKEND,
@@ -1813,8 +1813,8 @@ _PG_init(void)
 	DefineCustomIntVariable(
 		"multimaster.max_recovery_lag",
 		"Maximal lag of replication slot of failed node after which this slot is dropped to avoid transaction log overflow",
-		"Dropping slog makes it not possible to recover node using logical replication mechanism, it will be ncessary to completely copy content of some other nodes " 
-		"usimg basebackup or similar tool. Zero value of parameter disable droipping slot.",
+		"Dropping slot makes it not possible to recover node using logical replication mechanism, it will be ncessary to completely copy content of some other nodes " 
+		"using basebackup or similar tool. Zero value of parameter disable dropping slot.",
 		&MtmMaxRecoveryLag,
 		100000000,
 		0,