Count number of received heartbeast for statustic

Author: knizhnik

arbiter.c

@@ -949,13 +949,14 @@ static void MtmReceiver(Datum arg)
 					Mtm->nodes[node-1].lastHeartbeat = MtmGetSystemTime();
 
 					MtmCheckResponse(msg);
-					MTM_LOG2("Receive response %s for transaction %s from node %d", MtmMessageKindMnem[msg->code], msg->gid, msg->node);
+					MTM_LOG2("Receive response %s for transaction %s from node %d", MtmMessageKindMnem[msg->code], msg->gid, node);
 
 					switch (msg->code) {
 					  case MSG_HEARTBEAT:
 						MTM_LOG4("Receive HEARTBEAT from node %d with timestamp %lld delay %lld", 
 								 node, msg->csn, USEC_TO_MSEC(MtmGetSystemTime() - msg->csn)); 
-						continue;
+						Mtm->nodes[node-1].nHeartbeats += 1;
+						continue;						
 					  case MSG_POLL_REQUEST:
 						Assert(*msg->gid);
 						tm = (MtmTransMap*)hash_search(MtmGid2State, msg->gid, HASH_FIND, NULL);
@@ -965,7 +966,7 @@ static void MtmReceiver(Datum arg)
 						} else {
 							msg->status = tm->state->status;
 							msg->csn = tm->state->csn;
-							MTM_LOG1("Send response %s for transaction %s to node %d", MtmTxnStatusMnem[msg->status], msg->gid, msg->node);
+							MTM_LOG1("Send response %s for transaction %s to node %d", MtmTxnStatusMnem[msg->status], msg->gid, node);
 						}
 						msg->disabledNodeMask = Mtm->disabledNodeMask;
 						msg->connectivityMask = SELF_CONNECTIVITY_MASK;

multimaster--1.0.sql

@@ -36,7 +36,7 @@ AS 'MODULE_PATHNAME','mtm_get_last_csn'
 LANGUAGE C;
 
 
-CREATE TYPE mtm.node_state AS ("id" integer, "disabled" bool, "disconnected" bool, "catchUp" bool, "slotLag" bigint, "avgTransDelay" bigint, "lastStatusChange" timestamp, "oldestSnapshot" bigint, "SenderPid" integer, "SenderStartTime" timestamp, "ReceiverPid" integer, "ReceiverStartTime" timestamp, "connStr" text, "connectivityMask" bigint, "stalled" bool, "stopped" bool);
+CREATE TYPE mtm.node_state AS ("id" integer, "disabled" bool, "disconnected" bool, "catchUp" bool, "slotLag" bigint, "avgTransDelay" bigint, "lastStatusChange" timestamp, "oldestSnapshot" bigint, "SenderPid" integer, "SenderStartTime" timestamp, "ReceiverPid" integer, "ReceiverStartTime" timestamp, "connStr" text, "connectivityMask" bigint, "stalled" bool, "stopped" bool, "nHeartbeats" bigint);
 
 CREATE FUNCTION mtm.get_nodes_state() RETURNS SETOF mtm.node_state
 AS 'MODULE_PATHNAME','mtm_get_nodes_state'

multimaster.c

@@ -2487,6 +2487,7 @@ static void MtmInitialize()
 			Mtm->nodes[i].restartLSN = INVALID_LSN;
 			Mtm->nodes[i].originId = InvalidRepOriginId;
 			Mtm->nodes[i].timeline = 0;
+			Mtm->nodes[i].nHeartbeats = 0;
 		}
 		Mtm->nodes[MtmNodeId-1].originId = DoNotReplicateId;
 		/* All transaction originated from the current node should be ignored during recovery */
@@ -3975,6 +3976,7 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	usrfctx->values[13] = Int64GetDatum(Mtm->nodes[usrfctx->nodeId-1].connectivityMask);
 	usrfctx->values[14] = BoolGetDatum(BIT_CHECK(Mtm->stalledNodeMask, usrfctx->nodeId-1));
 	usrfctx->values[15] = BoolGetDatum(BIT_CHECK(Mtm->stoppedNodeMask, usrfctx->nodeId-1));
+	usrfctx->values[16] = Int64GetDatum(Mtm->nodes[usrfctx->nodeId-1].nHeartbeats);
 	usrfctx->nodeId += 1;
 
 	SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(heap_form_tuple(usrfctx->desc, usrfctx->values, usrfctx->nulls)));

multimaster.h

@@ -84,7 +84,7 @@
 #define Anum_mtm_local_tables_rel_name	 2
 
 #define Natts_mtm_trans_state   15
-#define Natts_mtm_nodes_state   16
+#define Natts_mtm_nodes_state   17
 #define Natts_mtm_cluster_state 19
 
 typedef ulong64 csn_t; /* commit serial number */
@@ -226,6 +226,7 @@ typedef struct
 	void*       lockGraphData;
 	int         lockGraphAllocated;
 	int         lockGraphUsed;
+	uint64      nHeartbeats;
 } MtmNodeInfo;
 
 typedef struct MtmL2List