Add information about BGW to node status

Author: knizhnik

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, "connStr" text);
+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);
 
 CREATE FUNCTION mtm.get_nodes_state() RETURNS SETOF mtm.node_state
 AS 'MODULE_PATHNAME','mtm_get_nodes_state'

multimaster.c

@@ -2073,7 +2073,8 @@ void MtmDropNode(int nodeId, bool dropSlot)
 static void
 MtmOnProcExit(int code, Datum arg)
 {
-	if (MtmReplicationNodeId >= 0) { 
+	if (MtmReplicationNodeId > 0) { 
+		Mtm->nodes[MtmReplicationNodeId-1].senderPid = -1;
 		MTM_LOG1("WAL-sender to %d is terminated", MtmReplicationNodeId); 
 		MtmOnNodeDisconnect(MtmReplicationNodeId);
 	}
@@ -2085,6 +2086,8 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	ListCell *param;
 	bool recoveryCompleted = false;
 	MtmIsRecoverySession = false;
+	Mtm->nodes[MtmReplicationNodeId-1].senderPid = MyProcPid;
+	Mtm->nodes[MtmReplicationNodeId-1].senderStartTime = MtmGetSystemTime();
 	foreach(param, args->in_params)
 	{
 		DefElem    *elem = lfirst(param);
@@ -2377,7 +2380,11 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	usrfctx->values[5] = Int64GetDatum(Mtm->transCount ? Mtm->nodes[usrfctx->nodeId-1].transDelay/Mtm->transCount : 0);
 	usrfctx->values[6] = TimestampTzGetDatum(time_t_to_timestamptz(Mtm->nodes[usrfctx->nodeId-1].lastStatusChangeTime/USECS_PER_SEC));
 	usrfctx->values[7] = Int64GetDatum(Mtm->nodes[usrfctx->nodeId-1].oldestSnapshot);
-	usrfctx->values[8] = CStringGetTextDatum(Mtm->nodes[usrfctx->nodeId-1].con.connStr);
+	usrfctx->values[8] = Int32GetDatum(Mtm->nodes[usrfctx->nodeId-1].senderPid);
+	usrfctx->values[9] = Int64GetDatum(Mtm->nodes[usrfctx->nodeId-1].senderStartTime);
+	usrfctx->values[10] = Int32GetDatum(Mtm->nodes[usrfctx->nodeId-1].receiverPid);
+	usrfctx->values[11] = Int64GetDatum(Mtm->nodes[usrfctx->nodeId-1].receiverStartTime);   
+	usrfctx->values[12] = CStringGetTextDatum(Mtm->nodes[usrfctx->nodeId-1].con.connStr);
 	usrfctx->nodeId += 1;
 
 	SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(heap_form_tuple(usrfctx->desc, usrfctx->values, usrfctx->nulls)));

multimaster.h

@@ -60,7 +60,7 @@
 #define Anum_mtm_local_tables_rel_name	 2
 
 #define Natts_mtm_cluster_state 16
-#define Natts_mtm_nodes_state   9
+#define Natts_mtm_nodes_state   13
 
 typedef uint64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)
@@ -125,8 +125,12 @@ typedef struct
 	MtmConnectionInfo con;
 	timestamp_t transDelay;
 	timestamp_t lastStatusChangeTime;
+	timestamp_t receiverStartTime;
+	timestamp_t senderStartTime;
+	int         senderPid;
+	int         receiverPid;
 	XLogRecPtr  flushPos;
-	csn_t  oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
+	csn_t  oldestSnapshot; /* Oldest snapshot used by active transactions at this node */	
 } MtmNodeInfo;
 
 typedef struct MtmTransState

pglogical_proto.c

@@ -146,6 +146,10 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	} else { 
 		csn_t csn = MtmTransactionSnapshot(txn->xid);
 		bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
+		/* 
+		 * INVALID_CSN means replicated transaction (transaction initiated by some other nodes).
+		 * We do not need to send such transactions unless we perform recovery
+		 */
 		if (csn == INVALID_CSN && !isRecovery) {
 			return;
 		}

pglogical_receiver.c

@@ -226,6 +226,9 @@ pglogical_receiver_main(Datum main_arg)
 
 	MtmCreateSpillDirectory(nodeId);
 
+	Mtm->nodes[nodeId-1].senderPid = MyProcPid;
+	Mtm->nodes[nodeId-1].senderStartTime = MtmGetSystemTime();
+
     sprintf(worker_proc, "mtm_pglogical_receiver_%d_%d", MtmNodeId, nodeId);
 
 	/* We're now ready to receive signals */