Eliminate dupplicated messages

Author: knizhnik

Dockerfile

@@ -31,7 +31,7 @@ USER postgres
 ENV CFLAGS -O0
 WORKDIR /pg
 
-ENV REBUILD 1
+ENV REBUILD 2
 
 RUN cd /pg && \
 	git clone https://github.com/postgrespro/postgres_cluster.git --depth 1 && \

arbiter.c

@@ -81,12 +81,13 @@
 typedef struct
 {
 	MtmMessageCode code; /* Message code: MSG_READY, MSG_PREPARE, MSG_COMMIT, MSG_ABORT */
-    int            node; /* Sender node ID */
+    int            node; /* Sender node ID */	
 	TransactionId  dxid; /* Transaction ID at destination node */
 	TransactionId  sxid; /* Transaction ID at sender node */  
 	csn_t          csn;  /* Local CSN in case of sending data from replica to master, global CSN master->replica */
 	nodemask_t     disabledNodeMask; /* Bitmask of disabled nodes at the sender of message */
 	csn_t          oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
+	uint64         seqno;/* Message sequence number (used to eliminate duplicated messages) */
 } MtmArbiterMessage;
 
 typedef struct 
@@ -112,6 +113,7 @@ static int         busy_socket;
 static void MtmTransSender(Datum arg);
 static void MtmTransReceiver(Datum arg);
 static void MtmSendHeartbeat(void);
+static bool MtmSendToNode(int node, void const* buf, int size);
 
 
 static char const* const messageText[] = 
@@ -248,6 +250,7 @@ static bool MtmWriteSocket(int sd, void const* buf, int size)
 		if (rc == 1) { 
 			int n = send(sd, src, size, 0);
 			if (n < 0) {
+				Assert(errno != EINTR); /* should not happen in non-blocking call */
 				busy_socket = -1;
 				return false;
 			}
@@ -266,6 +269,7 @@ static int MtmReadSocket(int sd, void* buf, int buf_size)
 {
 	int rc = recv(sd, buf, buf_size, 0);
 	if (rc <= 0) { 
+		Assert(errno != EINTR); /* should not happen in non-blocking call */
 		return -1;
 	}
 	return rc;
@@ -346,9 +350,8 @@ static void MtmSendHeartbeat()
 	{
 		if (sockets[i] >= 0 && sockets[i] != busy_socket && !BIT_CHECK(Mtm->disabledNodeMask|Mtm->reconnectMask, i))
 		{ 
-			size_t rc = send(sockets[i], &msg, sizeof(msg), 0);
-			if ((size_t)rc != sizeof(msg)) { 
-				elog(LOG, "Failed to send heartbeat to node %d: %d", i+1, errno);
+			if (!MtmSendToNode(i, &msg, sizeof(msg))) { 
+				elog(LOG, "Arbiter failed to send heartbeat to node %d", i+1);
 			}
 		}
 	}
@@ -629,6 +632,7 @@ static void MtmAppendBuffer(MtmBuffer* txBuffer, TransactionId xid, int node, Mt
 	MTM_LOG3("Send %s message CSN=%ld to node %d from node %d for global transaction %d/local transaction %d", 
 			 messageText[ts->cmd], ts->csn, node+1, MtmNodeId, ts->gtid.xid, ts->xid);
 	Assert(ts->cmd != MSG_INVALID);
+	buf->data[buf->used].seqno = ++Mtm->nodes[node].sendSeqNo;
 	buf->data[buf->used].code = ts->cmd;
 	buf->data[buf->used].sxid = ts->xid;
 	buf->data[buf->used].csn  = ts->csn;
@@ -845,10 +849,17 @@ static void MtmTransReceiver(Datum arg)
 						elog(WARNING, "Ignore message from dead node %d\n", msg->node);
 						continue;
 					}
+					if (msg->seqno <= Mtm->nodes[msg->node-1].recvSeqNo) { 
+						elog(WARNING, "Ignore duplicated message %ld from node %d", msg->seqno, msg->node);
+						continue;
+					}
+					Mtm->nodes[msg->node-1].recvSeqNo = msg->seqno;
 
 					ts = (MtmTransState*)hash_search(MtmXid2State, &msg->dxid, HASH_FIND, NULL);
-					Assert(ts != NULL);
-
+					if (ts == NULL) { 
+						elog(WARNING, "Ignore response for unexisted transaction %d from node %d", msg->dxid, msg->node);
+						continue;
+					}
 					if (BIT_CHECK(msg->disabledNodeMask, MtmNodeId-1) && Mtm->status != MTM_RECOVERY) { 
 						elog(PANIC, "Node %d thinks that I was dead: perform hara-kiri not to be a zombie", msg->node);
 					}

multimaster.c

@@ -814,6 +814,10 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 	MtmTransState* ts;
 	MTM_TXTRACE(x, "PostPrepareTransaction Start");
 
+	if (!x->isDistributed) {
+		return;
+	}
+
 	if (Mtm->inject2PCError == 2) { 
 		Mtm->inject2PCError = 0;
 		elog(ERROR, "ERROR INJECTION for transaction %d (%s)", x->xid, x->gid);
@@ -1662,6 +1666,8 @@ static void MtmInitialize()
 			Mtm->nodes[i].con = MtmConnections[i];
 			Mtm->nodes[i].flushPos = 0;
 			Mtm->nodes[i].lastHeartbeat = 0;
+			Mtm->nodes[i].sendSeqNo = 0;
+			Mtm->nodes[i].recvSeqNo = 0;
 		}
 		PGSemaphoreCreate(&Mtm->votingSemaphore);
 		PGSemaphoreReset(&Mtm->votingSemaphore);

multimaster.h

@@ -136,7 +136,9 @@ typedef struct
 	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 */	
+	uint64      sendSeqNo;
+	uint64      recvSeqNo;
 } MtmNodeInfo;
 
 typedef struct MtmTransState

pglogical_receiver.c

@@ -40,6 +40,7 @@
 #include "spill.h"
 
 #define ERRCODE_DUPLICATE_OBJECT_STR  "42710"
+#define RECEIVER_SUSPEND_TIMEOUT (10*USECS_PER_SEC)
 
 /* Signal handling */
 static volatile sig_atomic_t got_sigterm = false;
@@ -355,8 +356,9 @@ pglogical_receiver_main(Datum main_arg)
 			proc_exit(1);
 
 		if (Mtm->status == MTM_OFFLINE || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId)) {
-			ereport(LOG, (errmsg("%s: terminating WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
-			proc_exit(0);
+			ereport(LOG, (errmsg("%s: suspending WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
+			MtmSleep(RECEIVER_SUSPEND_TIMEOUT);
+			continue;
 		}