Add recover_node function

Author: knizhnik

contrib/mmts/arbiter.c

@@ -119,15 +119,15 @@ static BackgroundWorker MtmSender = {
 	"mtm-sender",
 	BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION, /* do not need connection to the database */
 	BgWorkerStart_ConsistentState,
-	1, /* restart in one second (is it possible to restart immediately?) */
+	MULTIMASTER_BGW_RESTART_TIMEOUT,
 	MtmTransSender
 };
 
 static BackgroundWorker MtmRecevier = {
 	"mtm-receiver",
 	BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION, /* do not need connection to the database */
 	BgWorkerStart_ConsistentState,
-	1, /* restart in one second (is it possible to restart immediately?) */
+	MULTIMASTER_BGW_RESTART_TIMEOUT,
 	MtmTransReceiver
 };
 
@@ -297,6 +297,7 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 
 			/* Some node considered that I am dead, so switch to recovery mode */
 			if (BIT_CHECK(msg.disabledNodeMask, MtmNodeId-1)) { 
+				elog(WARNING, "Node %d think that I am dead", msg.node);
 				MtmSwitchClusterMode(MTM_RECOVERY);
 			}
 			/* Combine disable masks from all node. Is it actually correct or we should better check availability of nodes ourselves? */

contrib/mmts/bgwpool.c

@@ -85,7 +85,7 @@ void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor)
     worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_main = BgwPoolMainLoop;
-	worker.bgw_restart_time = 10; /* Wait 10 seconds for restart before crash */
+	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
 
     for (i = 0; i < nWorkers; i++) { 
         BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)malloc(sizeof(BgwPoolExecutorCtx));

contrib/mmts/bgwpool.h

@@ -8,6 +8,7 @@
 typedef void(*BgwPoolExecutor)(int id, void* work, size_t size);
 
 #define MAX_DBNAME_LEN 30
+#define MULTIMASTER_BGW_RESTART_TIMEOUT 10 /* seconds */
 
 typedef struct
 {

contrib/mmts/multimaster--1.0.sql

@@ -13,6 +13,12 @@ CREATE FUNCTION mtm.drop_node(node integer, drop_slot bool default false) RETURN
 AS 'MODULE_PATHNAME','mtm_drop_node'
 LANGUAGE C;
 
+-- Create replication slot for the node which was previously dropped together with it's slot 
+CREATE FUNCTION mtm.recover_node(node integer) RETURNS void
+AS 'MODULE_PATHNAME','mtm_recover_node'
+LANGUAGE C;
+
+
 CREATE FUNCTION mtm.get_snapshot() RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;

contrib/mmts/multimaster.c

@@ -89,6 +89,7 @@ PG_MODULE_MAGIC;
 PG_FUNCTION_INFO_V1(mtm_start_replication);
 PG_FUNCTION_INFO_V1(mtm_stop_replication);
 PG_FUNCTION_INFO_V1(mtm_drop_node);
+PG_FUNCTION_INFO_V1(mtm_recover_node);
 PG_FUNCTION_INFO_V1(mtm_get_snapshot);
 
 static Snapshot MtmGetSnapshot(Snapshot snapshot);
@@ -1182,6 +1183,22 @@ MtmSlotMode MtmReceiverSlotMode(int nodeId)
 	return dtm->recoverySlot ? SLOT_CREATE_NEW : SLOT_OPEN_ALWAYS;
 }
 
+void MtmRecoverNode(int nodeId)
+{
+	if (nodeId <= 0 || nodeId > dtm->nNodes) 
+	{ 
+		elog(ERROR, "NodeID %d is out of range [1,%d]", nodeId, dtm->nNodes);
+	}
+	if (!BIT_CHECK(dtm->disabledNodeMask, nodeId-1)) { 
+		elog(ERROR, "Node %d was not disabled", nodeId);
+	}
+	if (!IsTransactionBlock())
+	{
+		MtmBroadcastUtilityStmt(psprintf("select pg_create_logical_replication_slot('" MULTIMASTER_SLOT_PATTERN "', '" MULTIMASTER_NAME "')", nodeId), true);
+	}
+}
+	
+	
 void MtmDropNode(int nodeId, bool dropSlot)
 {
 	if (!BIT_CHECK(dtm->disabledNodeMask, nodeId-1))
@@ -1227,6 +1244,14 @@ mtm_drop_node(PG_FUNCTION_ARGS)
     PG_RETURN_VOID();
 }
 
+Datum
+mtm_recover_node(PG_FUNCTION_ARGS)
+{
+	int nodeId = PG_GETARG_INT32(0);
+	MtmRecoverNode(nodeId);
+    PG_RETURN_VOID();
+}
+	
 Datum
 mtm_get_snapshot(PG_FUNCTION_ARGS)
 {
@@ -1599,7 +1624,7 @@ MtmSerializeLock(PROCLOCK* proclock, void* arg)
                         {
                             if ((proclock->holdMask & LOCKBIT_ON(lm)) && (conflictMask & LOCKBIT_ON(lm)))
                             {
-                                MTM_TRACE("%d: %u(%u) waits for %u(%u)\n", getpid(), srcPgXact->xid, proc->pid, dstPgXact->xid, proclock->tag.myProc->pid);
+                                MTM_TRACE("%d: %u(%u) waits for %u(%u)\n", MyProcPid, srcPgXact->xid, proc->pid, dstPgXact->xid, proclock->tag.myProc->pid);
                                 MtmGetGtid(srcPgXact->xid, &gtid); /* transaction holding lock */
 								ByteBufferAppendInt32(buf, gtid.node); 
 								ByteBufferAppendInt32(buf, gtid.xid); 
@@ -1689,6 +1714,7 @@ void MtmRefreshClusterStatus(bool nowait)
 
 	clique = MtmFindMaxClique(matrix, MtmNodes, &clique_size);
 	if (clique_size >= MtmNodes/2+1) { /* have quorum */
+		elog(WARNING, "Find clique %lx, disabledNodeMask %lx", clique, dtm->disabledNodeMask);
 		MtmLock(LW_EXCLUSIVE);
 		mask = ~clique & (((nodemask_t)1 << MtmNodes)-1) & ~dtm->disabledNodeMask; /* new disabled nodes mask */
 		for (i = 0; mask != 0; i++, mask >>= 1) {

contrib/mmts/multimaster.h

@@ -13,7 +13,7 @@
 #define MTM_TUPLE_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 */
 
-#define MULTIMASTER_NAME                "mtm"
+#define MULTIMASTER_NAME                "multimaster"
 #define MULTIMASTER_SCHEMA_NAME         "mtm"
 #define MULTIMASTER_DDL_TABLE           "ddl_log"
 #define MULTIMASTER_SLOT_PATTERN        "mtm_slot_%d"
@@ -146,6 +146,7 @@ extern void  MtmAdjustSubtransactions(MtmTransState* ts);
 extern void  MtmLock(LWLockMode mode);
 extern void  MtmUnlock(void);
 extern void  MtmDropNode(int nodeId, bool dropSlot);
+extern void  MtmRecoverNode(int nodeId);
 extern void  MtmOnNodeDisconnect(int nodeId);
 extern void  MtmOnNodeConnect(int nodeId);
 extern MtmState* MtmGetState(void);

contrib/mmts/pglogical_apply.c

@@ -830,7 +830,7 @@ void MtmExecutor(int id, void* work, size_t size)
     PG_CATCH();
     {
         FlushErrorState();
-		MTM_TRACE("%d: REMOTE abort transaction %d\n", getpid(), GetCurrentTransactionId());
+		MTM_TRACE("%d: REMOTE abort transaction %d\n", MyProcPid, GetCurrentTransactionId());
         AbortCurrentTransaction();
     }
     PG_END_TRY();

contrib/mmts/pglogical_receiver.c

@@ -56,7 +56,6 @@ static int receiver_idle_time = 0;
 static bool receiver_sync_mode = false;
 
 /* Worker name */
-static char *worker_name = "multimaster";
 char worker_proc[BGW_MAXLEN];
 
 /* Lastly written positions */
@@ -252,7 +251,7 @@ pglogical_receiver_main(Datum main_arg)
 		resetPQExpBuffer(query);
 	}
 	if (mode != SLOT_OPEN_EXISTED) { 
-		appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", args->receiver_slot, worker_name);
+		appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", args->receiver_slot, MULTIMASTER_NAME);
 		res = PQexec(conn, query->data);
 		if (PQresultStatus(res) != PGRES_TUPLES_OK)
 		{
@@ -568,7 +567,7 @@ int MtmStartReceivers(char* conns, int node_id)
     worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
 	worker.bgw_main = pglogical_receiver_main; 
-	worker.bgw_restart_time = 10; /* Wait 10 seconds for restart before crash */
+	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
 
     while (conn_str < conn_str_end) { 
         char* p = strchr(conn_str, ',');