fix mtm.nodes() to be function and add rest of the info needed

Author: kelvich

Cluster.pm

@@ -151,7 +151,7 @@ sub configure
 			multimaster.heartbeat_send_timeout = 250
 			multimaster.max_nodes = 6
 			# XXX try without ignore_tables_without_pk
-			multimaster.ignore_tables_without_pk = true
+			# multimaster.ignore_tables_without_pk = true
 			multimaster.volkswagen_mode = 1
 			log_line_prefix = '%t [%p]: '
 		));

multimaster--1.0.sql

@@ -62,6 +62,18 @@ CREATE TYPE mtm.node_info AS (
     "receiver_status" text
 );
 
+CREATE TYPE mtm.node AS (
+    "id" int,
+    "conninfo" text,
+    "is_self" bool,
+    "enabled" bool,
+    "connected" bool,
+    "sender_pid" int,
+    "receiver_pid" int,
+    "n_workers" int,
+    "receiver_status" text
+);
+
 ---
 --- User facing API for node info and management.
 ---
@@ -84,8 +96,13 @@ RETURNS mtm.cluster_status
 AS 'MODULE_PATHNAME','mtm_status'
 LANGUAGE C;
 
-CREATE VIEW mtm.nodes AS
-    SELECT id, conninfo, is_self, (mtm.node_info(id)).* FROM mtm.cluster_nodes;
+CREATE OR REPLACE FUNCTION mtm.nodes() RETURNS SETOF mtm.node AS
+$$
+    SELECT id, conninfo, is_self, (mtm.node_info(id)).*
+    FROM mtm.cluster_nodes
+    ORDER BY id;
+$$
+LANGUAGE sql;
 
 CREATE OR REPLACE FUNCTION mtm.add_node(connstr text) RETURNS void AS
 $$

src/include/multimaster.h

@@ -215,7 +215,12 @@ typedef struct
 	int			recoveryCount;	/* Number of completed recoveries */
 	int			donorNodeId;	/* Cluster node from which this node was
 								 * populated */
-	int			dmq_dest_ids[MTM_MAX_NODES];
+	struct {
+		MtmReplicationMode	receiver_mode;
+		pid_t				sender_pid;
+		pid_t				receiver_pid;
+		int					dmq_dest_id;
+	} peers[MTM_MAX_NODES];
 	BgwPool		pools[FLEXIBLE_ARRAY_MEMBER];		/* [Mtm->nAllNodes]: per-node data */
 } MtmState;
 

src/include/receiver.h

@@ -10,6 +10,14 @@ typedef struct
 	XLogRecPtr end_lsn;
 } MtmReceiverContext;
 
+typedef enum
+{
+	REPLMODE_RECOVERY,     /* perform recovery of the node by applying all data from the slot from specified point */
+	REPLMODE_RECOVERED    /* recovery of receiver node is completed so drop old slot and restart replication from the current position in WAL */
+} MtmReplicationMode;
+
+extern char const *const MtmReplicationModeName[];
+
 extern BackgroundWorkerHandle *MtmStartReceiver(int nodeId, Oid db_id, Oid user_id, pid_t monitor_pid);
 
 extern void MtmExecutor(void* work, size_t size, MtmReceiverContext *rctx);

src/multimaster.c

@@ -332,7 +332,7 @@ MtmStateShmemStartup()
 		MemSet(Mtm, 0, sizeof(MtmState));
 		Mtm->stop_new_commits = false;
 		Mtm->recovered = false;
-		Mtm->status = MTM_DISABLED; //MTM_INITIALIZATION;
+		Mtm->status = MTM_DISABLED;
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);
 		Mtm->disabledNodeMask = (nodemask_t) -1;
@@ -347,10 +347,13 @@ MtmStateShmemStartup()
 		Mtm->localTablesHashLoaded = false;
 		Mtm->latestSyncpoint = InvalidXLogRecPtr;
 
-		// XXX: change to dsa and make it per-receiver
 		for (i = 0; i < MtmMaxNodes; i++)
 		{
-			Mtm->dmq_dest_ids[i] = -1;
+			Mtm->peers[i].receiver_pid = InvalidPid;
+			Mtm->peers[i].sender_pid = InvalidPid;
+			Mtm->peers[i].dmq_dest_id = -1;
+
+			// XXX: change to dsa and make it per-receiver
 			BgwPoolInit(&Mtm->pools[i], MtmExecutor, MtmQueueSize, 0);
 		}
 	}
@@ -838,6 +841,8 @@ mtm_init_cluster(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+// XXX: During evaluation of (mtm.node_info(id)).* this function called
+// once each columnt for every row. So may be just rewrite to SRF.
 Datum
 mtm_node_info(PG_FUNCTION_ARGS)
 {
@@ -847,14 +852,38 @@ mtm_node_info(PG_FUNCTION_ARGS)
 	bool		nulls[Natts_mtm_node_info] = {false};
 
 	MtmLock(LW_SHARED);
+
 	values[Anum_mtm_node_info_enabled - 1] =
 		BoolGetDatum(!BIT_CHECK(Mtm->disabledNodeMask, node_id - 1));
 	values[Anum_mtm_node_info_connected - 1] =
 		BoolGetDatum(!BIT_CHECK(Mtm->selfConnectivityMask, node_id - 1));
-	values[Anum_mtm_node_info_sender_pid - 1] = Int32GetDatum(0);
-	values[Anum_mtm_node_info_receiver_pid - 1] = Int32GetDatum(0);
-	values[Anum_mtm_node_info_n_workers - 1] = Int32GetDatum(0);
-	values[Anum_mtm_node_info_receiver_status - 1] = CStringGetTextDatum("unimplemented");
+
+	if (Mtm->peers[node_id - 1].sender_pid != InvalidPid)
+	{
+		values[Anum_mtm_node_info_sender_pid - 1] =
+			Int32GetDatum(Mtm->peers[node_id - 1].sender_pid);
+	}
+	else
+	{
+		nulls[Anum_mtm_node_info_sender_pid - 1] = true;
+	}
+
+	if (Mtm->peers[node_id - 1].receiver_pid != InvalidPid)
+	{
+		values[Anum_mtm_node_info_receiver_pid - 1] =
+			Int32GetDatum(Mtm->peers[node_id - 1].receiver_pid);
+		values[Anum_mtm_node_info_n_workers - 1] =
+			Int32GetDatum(Mtm->pools[node_id - 1].nWorkers);
+		values[Anum_mtm_node_info_receiver_status - 1] =
+			CStringGetTextDatum(MtmReplicationModeName[Mtm->peers[node_id - 1].receiver_mode]);
+	}
+	else
+	{
+		nulls[Anum_mtm_node_info_receiver_pid - 1] = true;
+		nulls[Anum_mtm_node_info_n_workers - 1] = true;
+		nulls[Anum_mtm_node_info_receiver_status - 1] = true;
+	}
+
 	MtmUnlock();
 
 	get_call_result_type(fcinfo, NULL, &desc);

src/pglogical_apply.c

@@ -852,7 +852,7 @@ mtm_send_reply(TransactionId xid, int node_id, MtmMessageCode msg_code)
 	MtmArbiterMessage msg;
 
 	MtmLock(LW_SHARED);
-	dest_id = Mtm->dmq_dest_ids[node_id - 1];
+	dest_id = Mtm->peers[node_id - 1].dmq_dest_id;
 	MtmUnlock();
 	Assert(dest_id >= 0);
 

src/pglogical_proto.c

@@ -783,6 +783,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 		mtm_log(ERROR, "Out-of-clique node %d tries to connect",
 				 MtmReplicationNodeId);
 	}
+	Mtm->peers[MtmReplicationNodeId - 1].sender_pid = MyProcPid;
 	MtmUnlock();
 
 	if (hooks_data->is_recovery)
@@ -843,6 +844,7 @@ MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
 	Assert(MtmReplicationNodeId >= 0);
 
 	MtmLock(LW_EXCLUSIVE);
+	Mtm->peers[MtmReplicationNodeId - 1].sender_pid = InvalidPid;
 	if (BIT_CHECK(Mtm->pglogicalSenderMask, MtmReplicationNodeId-1))
 	{
 		BIT_CLEAR(Mtm->pglogicalSenderMask, MtmReplicationNodeId-1);

src/pglogical_receiver.c

@@ -66,15 +66,6 @@
 
 bool MtmIsReceiver;
 
-typedef enum
-{
-	REPLMODE_EXIT,         /* receiver should exit */
-	REPLMODE_RECOVERED,    /* recovery of receiver node is completed so drop old slot and restart replication from the current position in WAL */
-	REPLMODE_RECOVERY,     /* perform recovery of the node by applying all data from the slot from specified point */
-	REPLMODE_CREATE_NEW,   /* destination node is recovered: drop old slot and restart from roveredLsn position */
-	REPLMODE_OPEN_EXISTED  /* normal mode: use existed slot or create new one and start receiving data from it from the remembered position */
-} MtmReplicationMode;
-
 typedef struct MtmFlushPosition
 {
 	dlist_node node;
@@ -83,13 +74,10 @@ typedef struct MtmFlushPosition
 	XLogRecPtr	remote_end;
 } MtmFlushPosition;
 
-static char const* const MtmReplicationModeName[] =
+char const* const MtmReplicationModeName[] =
 {
-	"exit",
-	"recovered",   /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
-	"recovery",	   /* perform recorvery of the node by applying all data from theslot from specified point */
-	"create_new",  /* destination node is recovered: drop old slot and restart from roveredLsn position */
-	"open_existed" /* normal mode: use existed slot or create new one and start receiving data from it from the rememered position */
+	"recovery",    /* perform recorvery of the node by applying all data from theslot from specified point */
+	"recovered"    /* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
 };
 
 static dlist_head MtmLsnMapping = DLIST_STATIC_INIT(MtmLsnMapping);
@@ -550,6 +538,9 @@ pglogical_receiver_at_exit(int status, Datum arg)
 {
 	int node_id = DatumGetInt32(arg);
 	BgwPoolCancel(&Mtm->pools[node_id - 1]);
+	MtmLock(LW_EXCLUSIVE);
+	Mtm->peers[node_id - 1].receiver_pid = InvalidPid;
+	MtmUnlock();
 }
 
 void
@@ -639,6 +630,11 @@ pglogical_receiver_main(Datum main_arg)
 		 */
 		mode = MtmGetReplicationMode(nodeId);
 
+		MtmLock(LW_EXCLUSIVE);
+		Mtm->peers[nodeId - 1].receiver_pid = MyProcPid;
+		Mtm->peers[nodeId - 1].receiver_mode = mode;
+		MtmUnlock();
+
 		// XXX: delete unnecessary modes
 		Assert(mode == REPLMODE_RECOVERY || mode == REPLMODE_RECOVERED);
 

src/resolver.c

@@ -407,7 +407,7 @@ scatter_status_requests(MtmConfig *mtm_cfg)
 
 				// XXX: we need here to await destination
 				MtmLock(LW_SHARED);
-				dest_id = Mtm->dmq_dest_ids[node_id - 1];
+				dest_id = Mtm->peers[node_id - 1].dmq_dest_id;
 				MtmUnlock();
 				Assert(dest_id >= 0);
 

src/state.c

@@ -981,7 +981,7 @@ check_status_requests(MtmConfig *mtm_cfg)
 		msg->node = Mtm->my_node_id;
 
 		MtmLock(LW_SHARED);
-		dest_id = Mtm->dmq_dest_ids[sender_node_id - 1];
+		dest_id = Mtm->peers[sender_node_id - 1].dmq_dest_id;
 		MtmUnlock();
 		Assert(dest_id >= 0);
 
@@ -1128,7 +1128,7 @@ start_node_workers(int node_id, MtmConfig *new_cfg, Datum arg)
 	dest = dmq_destination_add(dmq_connstr, dmq_my_name, dmq_node_name,
 							   MtmHeartbeatSendTimeout);
 	MtmLock(LW_EXCLUSIVE);
-	Mtm->dmq_dest_ids[node_id - 1] = dest;
+	Mtm->peers[node_id - 1].dmq_dest_id = dest;
 	MtmUnlock();
 
 	/* Attach receiver so we can collect tx requests */
@@ -1179,7 +1179,7 @@ stop_node_workers(int node_id, MtmConfig *new_cfg, Datum arg)
 	dmq_destination_drop(dmq_name);
 
 	MtmLock(LW_EXCLUSIVE);
-	Mtm->dmq_dest_ids[node_id - 1] = -1;
+	Mtm->peers[node_id - 1].dmq_dest_id = -1;
 	MtmUnlock();
 
 	/*
@@ -1354,7 +1354,7 @@ MtmMonitor(Datum arg)
 		}
 
 		/* Launch resolver after we added dmq destinations */
-		// XXX: that's because of current use of Mtm->dmq_dest_ids[]
+		// XXX: that's because of current use of Mtm->peers[].dmq_dest_id
 		if (resolver == NULL)
 			resolver = ResolverStart(db_id, user_id);