Fix recovery

Author: knizhnik

arbiter.c

@@ -289,8 +289,7 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 		elog(ERROR, "Arbiter failed to resolve host '%s' by name", host);
 	}
 
-Retry:
-
+  Retry:
     while (1) {
 		int rc = -1;
 
@@ -384,20 +383,29 @@ static void MtmOpenConnections()
 
 
 static bool MtmSendToNode(int node, void const* buf, int size)
-{
-	while (sockets[node] < 0 || !MtmWriteSocket(sockets[node], buf, size)) { 
-		elog(WARNING, "Arbiter failed to write to node %d: %d", node+1, errno);
-		if (sockets[node] >= 0) { 
+{	
+	while (true) {
+		if (sockets[node] >= 0 && BIT_CHECK(Mtm->reconnectMask, node)) {
+			elog(WARNING, "Arbiter is forced to reconnect to node %d", node+1); 
+			BIT_CLEAR(Mtm->reconnectMask, node);
 			close(sockets[node]);
+			sockets[node] = -1;
 		}
-		sockets[node] = MtmConnectSocket(Mtm->nodes[node].con.hostName, MtmArbiterPort + node + 1, MtmReconnectAttempts);
-		if (sockets[node] < 0) { 
-			MtmOnNodeDisconnect(node+1);
-			return false;
+		if (sockets[node] < 0 || !MtmWriteSocket(sockets[node], buf, size)) { 
+			if (sockets[node] >= 0) { 
+				elog(WARNING, "Arbiter failed to write to node %d: %d", node+1, errno);
+				close(sockets[node]);
+			}
+			sockets[node] = MtmConnectSocket(Mtm->nodes[node].con.hostName, MtmArbiterPort + node + 1, MtmReconnectAttempts);
+			if (sockets[node] < 0) { 
+				MtmOnNodeDisconnect(node+1);
+				return false;
+			}
+			MTM_TRACE("Arbiter restablished connection with node %d\n", node+1);
+		} else { 
+			return true;
 		}
-		elog(NOTICE, "Arbiter restablish connection with node %d", node+1);
 	}
-	return true;
 }
 
 static int MtmReadFromNode(int node, void* buf, int buf_size)
@@ -477,10 +485,6 @@ static void MtmAcceptIncomingConnections()
 
 	sockets[MtmNodeId-1] = gateway;
 	MtmRegisterSocket(gateway, MtmNodeId-1);
-
-	for (i = 0; i < MtmNodes-1; i++) {
-		MtmAcceptOneConnection();
-	}
 }
 
 
@@ -693,6 +697,7 @@ static void MtmTransReceiver(Datum arg)
 									 msg->node, Mtm->disabledNodeMask, msg->disabledNodeMask);
 								ts->status = TRANSACTION_STATUS_ABORTED;
 								MtmAdjustSubtransactions(ts);
+								Mtm->nActiveTransactions -= 1;
 							}
 
 							if (++ts->nVotes == Mtm->nNodes) { 
@@ -712,6 +717,7 @@ static void MtmTransReceiver(Datum arg)
 								Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
 								ts->status = TRANSACTION_STATUS_ABORTED;
 								MtmAdjustSubtransactions(ts);
+								Mtm->nActiveTransactions -= 1;
 							}
 							if (++ts->nVotes == Mtm->nNodes) {
 								MtmWakeUpBackend(ts);

bgwpool.h

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

multimaster.c

@@ -736,11 +736,15 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 	ts = hash_search(MtmXid2State, &x->xid, HASH_FIND, NULL);
 	Assert(ts != NULL);
 
-	if (!MtmIsCoordinator(ts)) {
+	if (!MtmIsCoordinator(ts) || Mtm->status == MTM_RECOVERY) {
 		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
 		Assert(x->gid[0]);
 		tm->state = ts;
-		MtmSendNotificationMessage(ts, MSG_READY); /* send notification to coordinator */
+		if (Mtm->status != MTM_RECOVERY) { 
+			MtmSendNotificationMessage(ts, MSG_READY); /* send notification to coordinator */
+		} else {
+			ts->status = TRANSACTION_STATUS_UNKNOWN;
+		}
 		MtmUnlock();
 		MtmResetTransaction(x);
 	} else { 
@@ -769,6 +773,7 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 		Assert(tm != NULL);
 		tm->state->status = TRANSACTION_STATUS_ABORTED;
 		MtmAdjustSubtransactions(tm->state);
+		Mtm->nActiveTransactions -= 1;
 		MtmUnlock();
 		x->status = TRANSACTION_STATUS_ABORTED;
 	}
@@ -809,6 +814,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 			Mtm->nActiveTransactions -= 1;
 		}
 		if (!commit && x->isReplicated && TransactionIdIsValid(x->gtid.xid)) { 
+			Assert(Mtm->status != MTM_RECOVERY);
 			/* 
 			 * Send notification only if ABORT happens during transaction processing at replicas, 
 			 * do not send notification if ABORT is receiver from master 
@@ -1205,6 +1211,7 @@ void MtmCheckQuorum(void)
 void MtmOnNodeDisconnect(int nodeId)
 {
 	BIT_SET(Mtm->connectivityMask, nodeId-1);
+	BIT_SET(Mtm->reconnectMask, nodeId-1);
 	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false);
 
 	/* Wait more than socket KEEPALIVE timeout to let other nodes update their statuses */
@@ -1292,6 +1299,7 @@ static void MtmInitialize()
 		Mtm->pglogicalNodeMask = 0;
 		Mtm->walSenderLockerMask = 0;
 		Mtm->nodeLockerMask = 0;
+		Mtm->reconnectMask = 0;
 		Mtm->nLockers = 0;
 		Mtm->nActiveTransactions = 0;
 		Mtm->votingTransactions = NULL;

multimaster.h

@@ -133,6 +133,8 @@ typedef struct
 	nodemask_t pglogicalNodeMask;      /* bitmask of started pglogic receivers */
 	nodemask_t walSenderLockerMask;    /* Mask of WAL-senders IDs locking the cluster */
 	nodemask_t nodeLockerMask;         /* Mask of node IDs which WAL-senders are locking the cluster */
+	nodemask_t reconnectMask; 	       /* Mask of nodes connection to which has to be reestablished by sender */
+
     int    nNodes;                     /* Number of active nodes */
     int    nReceivers;                 /* Number of initialized logical receivers (used to determine moment when Mtm intialization is completed */
 	int    nLockers;                   /* Number of lockers */

pglogical_apply.c

@@ -902,7 +902,7 @@ void MtmExecutor(int id, void* work, size_t size)
     {    
         while (true) { 
             char action = pq_getmsgbyte(&s);
-            MTM_TRACE("%d: REMOTE process actiob %c\n", MyProcPid, action);
+            MTM_TRACE("%d: REMOTE process action %c\n", MyProcPid, action);
             switch (action) {
                 /* BEGIN */
             case 'B':

pglogical_proto.c

@@ -163,10 +163,10 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     pq_sendint64(out, txn->end_lsn);
     pq_sendint64(out, txn->commit_time);
 
-	if (flags == PGLOGICAL_COMMIT_PREPARED) { 
+	if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED) { 
 		pq_sendint64(out, MtmGetTransactionCSN(txn->xid));
 	}
-    if (flags != PGLOGICAL_COMMIT) { 
+    if (txn->xact_action != XLOG_XACT_COMMIT) { 
     	pq_sendstring(out, txn->gid);
 	}
 }

raftable.c

@@ -1,21 +1,9 @@
 #include <dlfcn.h>
 #include "postgres.h"
+#include "raftable.h"
 #include "raftable_wrapper.h"
 
 
-static raftable_get_t raftable_get_impl;
-static raftable_set_t raftable_set_impl;
-
-static void RaftableResolve()
-{
-	if (raftable_get_impl == NULL) { 
-		void* dll = dlopen(NULL, RTLD_NOW);
-		raftable_get_impl = dlsym(dll, "raftable_get");
-		raftable_set_impl = dlsym(dll, "raftable_set");
-		Assert(raftable_get_impl != NULL && raftable_set_impl != NULL);
-	}
-}
-
 /*
  * Raftable function proxies
  */
@@ -24,16 +12,14 @@ void* RaftableGet(char const* key, size_t* size, RaftableTimestamp* ts, bool now
 	if (!MtmUseRaftable) { 
 		return NULL;
 	}
-	RaftableResolve();
-	return (*raftable_get_impl)(key, size);
+	return raftable_get(key, size);
 }
 
 
 void RaftableSet(char const* key, void const* value, size_t size, bool nowait)
 {
 	if (MtmUseRaftable) { 
-		RaftableResolve();
-		(*raftable_set_impl)(key, value, size, nowait ? 0 : -1);
+		raftable_set(key, value, size, nowait ? 0 : -1);
 	}		
 }
 

raftable_wrapper.h

@@ -30,9 +30,6 @@ extern void  RaftableSet(char const* key, void const* value, size_t size, bool n
  */
 extern bool  RaftableCAS(char const* key, char const* value, bool nowait);
 
-typedef void* (*raftable_get_t)(char const* key, size_t* size);
-typedef void (*raftable_set_t)(char const* key, void const* value, size_t size, int timeout_ms);
-
 extern bool MtmUseRaftable;
 
 #endif

tests/postgresql.conf.mm

@@ -144,7 +144,7 @@
 
 #max_files_per_process = 1000		# min 25
 					# (change requires restart)
-shared_preload_libraries = 'multimaster'		# (change requires restart)
+shared_preload_libraries = 'raftable,multimaster'		# (change requires restart)
 
 # - Cost-Based Vacuum Delay -