Merge branch 'master' into more_tests

Author: kelvich

contrib/mmts/multimaster.c

@@ -1531,8 +1531,8 @@ static void MtmEnableNode(int nodeId)
 void MtmRecoveryCompleted(void)
 {
 	int i;
-	MTM_LOG1("Recovery of node %d is completed, disabled mask=%llx, connectivity mask=%llx, live nodes=%d",
-			 MtmNodeId, (long long) Mtm->disabledNodeMask, (long long) Mtm->connectivityMask, Mtm->nLiveNodes);
+	MTM_LOG1("Recovery of node %d is completed, disabled mask=%llx, connectivity mask=%llx, endLSN=%lx, live nodes=%d",
+			 MtmNodeId, (long long) Mtm->disabledNodeMask, (long long) Mtm->connectivityMask, GetXLogInsertRecPtr(), Mtm->nLiveNodes);
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
 	Mtm->recoveredLSN = GetXLogInsertRecPtr();
@@ -1542,7 +1542,7 @@ void MtmRecoveryCompleted(void)
 	for (i = 0; i < Mtm->nAllNodes; i++) { 
 		Mtm->nodes[i].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	}
-	/* Mode will be changed to online once all logical reciever are connected */
+	/* Mode will be changed to online once all logical receiver are connected */
 	MtmSwitchClusterMode(MTM_CONNECTED);
 	MtmUnlock();
 }
@@ -2131,7 +2131,6 @@ static void MtmInitialize()
 			Mtm->nodes[i].restartLSN = InvalidXLogRecPtr;
 			Mtm->nodes[i].originId = InvalidRepOriginId;
 			Mtm->nodes[i].timeline = 0;
-			Mtm->nodes[i].recoveredLSN = InvalidXLogRecPtr;
 		}
 		Mtm->nodes[MtmNodeId-1].originId = DoNotReplicateId;
 		/* All transaction originated from the current node should be ignored during recovery */
@@ -2884,13 +2883,14 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 {
 	MtmReplicationMode mode = REPLMODE_OPEN_EXISTED;
 
+	MtmLock(LW_EXCLUSIVE);
 	while ((Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) || BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) 
 	{ 	
 		if (*shutdown) 
 		{ 
+			MtmUnlock();
 			return REPLMODE_EXIT;
 		}
-		MtmLock(LW_EXCLUSIVE);
 		if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) { 
 			mode = REPLMODE_CREATE_NEW;
 		}
@@ -2913,6 +2913,7 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 		MtmUnlock();
 		/* delay opening of other slots until recovery is completed */
 		MtmSleep(STATUS_POLL_DELAY);
+		MtmLock(LW_EXCLUSIVE);
 	}
 	if (mode == REPLMODE_RECOVERED) { 
 		MTM_LOG1("%d: Restart replication from node %d after end of recovery", MyProcPid, nodeId);
@@ -2921,6 +2922,7 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 	} else { 
 		MTM_LOG1("%d: Continue replication from node %d", MyProcPid, nodeId);
 	}
+	MtmUnlock();
 	return mode;		
 }
 
@@ -3014,7 +3016,12 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 			}
 		} else if (strcmp("mtm_recovered_pos", elem->defname) == 0) { 
 			if (elem->arg != NULL && strVal(elem->arg) != NULL) {
-				sscanf(strVal(elem->arg), "%lx", &Mtm->nodes[MtmReplicationNodeId-1].recoveredLSN);
+				XLogRecPtr recoveredLSN;
+				sscanf(strVal(elem->arg), "%lx", &recoveredLSN);
+				MTM_LOG1("Recovered position of node %d is %lx", MtmReplicationNodeId, recoveredLSN); 
+				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) { 
+					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
+				}
 			} else { 
 				elog(ERROR, "Recovered position is not specified");
 			}
@@ -3129,16 +3136,21 @@ MtmReplicationRowFilterHook(struct PGLogicalRowFilterArgs* args)
 	return isDistributed;
 }
 
+/*
+ * Filter received transacyions at destination side.
+ * This function is executed by receiver, so there are no race conditions and it is possible to update nodes[i].restaetLSN without lock
+ */
 bool MtmFilterTransaction(char* record, int size)
 {
 	StringInfoData s;
 	uint8       flags;
 	XLogRecPtr  origin_lsn;
 	XLogRecPtr  end_lsn;
+	XLogRecPtr  restart_lsn;
 	int         replication_node;
 	int         origin_node;
 	char const* gid = "";
-	bool        duplicate;
+	bool        duplicate = false;
 
     s.data = record;
     s.len = size;
@@ -3174,11 +3186,17 @@ bool MtmFilterTransaction(char* record, int size)
 	  default:
 		break;
 	}
+	restart_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
+    if (Mtm->nodes[origin_node-1].restartLSN < restart_lsn) {
+		Mtm->nodes[origin_node-1].restartLSN = restart_lsn;
+    } else {
+		duplicate = true;
+	}
+		
 	//duplicate = Mtm->status == MTM_RECOVERY && origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLSN;
-	duplicate = origin_lsn != InvalidXLogRecPtr && origin_lsn <= Mtm->nodes[origin_node-1].restartLSN;
 
 	MTM_LOG1("%s transaction %s from node %d lsn %lx, flags=%x, origin node %d, original lsn=%lx, current lsn=%lx", 
-			 duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, flags, origin_node, origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
+			 duplicate ? "Ignore" : "Apply", gid, replication_node, end_lsn, flags, origin_node, origin_lsn, restart_lsn);
 	return duplicate;
 }
 

contrib/mmts/multimaster.h

@@ -204,7 +204,6 @@ typedef struct
 	void*       lockGraphData;
 	int         lockGraphAllocated;
 	int         lockGraphUsed;
-	XLogRecPtr  recoveredLSN;
 } MtmNodeInfo;
 
 typedef struct MtmTransState

contrib/mmts/pglogical_apply.c

@@ -430,11 +430,17 @@ process_remote_message(StringInfo s)
 			MtmAbortLogicalMessage* msg = (MtmAbortLogicalMessage*)messageBody;
 			int origin_node = msg->origin_node;
 			Assert(messageSize == sizeof(MtmAbortLogicalMessage));
+			/* This function is called directly by receiver, so there is no race condition and we can update
+			 * restartLSN without locks
+			 */
 			if (Mtm->nodes[origin_node-1].restartLSN < msg->origin_lsn) { 
 				Mtm->nodes[origin_node-1].restartLSN = msg->origin_lsn;
+				replorigin_session_origin_lsn = msg->origin_lsn; 
+				MtmRollbackPreparedTransaction(origin_node, msg->gid);
+			} else { 
+				MTM_LOG1("Ignore rollback of transaction %s from node %d because it's LSN %lx <= %lx", 
+						 msg->gid, origin_node, msg->origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
 			}
-			replorigin_session_origin_lsn = msg->origin_lsn; 
-			MtmRollbackPreparedTransaction(origin_node, msg->gid);
 			standalone = true;
 			break;
 		}
@@ -611,9 +617,6 @@ process_remote_commit(StringInfo in)
 	origin_lsn = pq_getmsgint64(in);
 
 	replorigin_session_origin_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
-	if (Mtm->nodes[origin_node-1].restartLSN < replorigin_session_origin_lsn) { 
-		Mtm->nodes[origin_node-1].restartLSN = replorigin_session_origin_lsn;
-	}
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 
 	switch(PGLOGICAL_XACT_EVENT(flags))

contrib/mmts/pglogical_receiver.c

@@ -260,9 +260,8 @@ pglogical_receiver_main(Datum main_arg)
 	{ 
 		int  count;
 		ConnStatusType status;
-		XLogRecPtr originStartPos = InvalidXLogRecPtr;
+		XLogRecPtr originStartPos = Mtm->nodes[nodeId-1].restartLSN;
 		int timeline;
-		bool newTimeline = false;
 
 		/* 
 		 * Determine when and how we should open replication slot.
@@ -291,12 +290,12 @@ pglogical_receiver_main(Datum main_arg)
 		if ((mode == REPLMODE_OPEN_EXISTED && timeline != Mtm->nodes[nodeId-1].timeline)
 			|| mode == REPLMODE_CREATE_NEW) 
 		{ /* recreate slot */
+			elog(LOG, "Recreate replication slot %s", slotName);
 			appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"", slotName);
 			res = PQexec(conn, query->data);
 			PQclear(res);
 			resetPQExpBuffer(query);
 			timeline = Mtm->nodes[nodeId-1].timeline;
-			newTimeline = true;
 		}
 		/* My original assumption was that we can perfrom recovery only from existed slot, 
 		 * but unfortunately looks like slots can "disapear" together with WAL-sender.
@@ -322,11 +321,7 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		/* Start logical replication at specified position */
-		if (mode == REPLMODE_RECOVERED) {
-			originStartPos = Mtm->nodes[nodeId-1].restartLSN;
-			MTM_LOG1("Restart replication from node %d from position %lx", nodeId, originStartPos);
-		} 
-		if (originStartPos == InvalidXLogRecPtr && !newTimeline) { 
+		if (originStartPos == InvalidXLogRecPtr) { 
 			StartTransactionCommand();
 			originName = psprintf(MULTIMASTER_SLOT_PATTERN, nodeId);
 			originId = replorigin_by_name(originName, true);
@@ -349,10 +344,11 @@ pglogical_receiver_main(Datum main_arg)
 			}
 			Mtm->nodes[nodeId-1].originId = originId;
 			CommitTransactionCommand();
-		} else if (mode == REPLMODE_CREATE_NEW) { 
-			originStartPos = Mtm->nodes[nodeId-1].recoveredLSN;
-		}
+		}		
 
+		MTM_LOG1("Start replication on slot %s from node %d at position %lx, mode %s, recovered lsn %lx", 
+				 slotName, nodeId, originStartPos, MtmReplicationModeName[mode], Mtm->recoveredLSN);
+
 		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s', \"mtm_restart_pos\" '%lx', \"mtm_recovered_pos\" '%lx')",
 						  slotName,
 						  (uint32) (originStartPos >> 32),
@@ -409,13 +405,16 @@ pglogical_receiver_main(Datum main_arg)
 				ereport(LOG, (errmsg("%s: restart WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
 				break;
 			}
-			if (count != Mtm->recoveryCount) { 
-				
+			if (count != Mtm->recoveryCount) { 				
 				ereport(LOG, (errmsg("%s: restart WAL receiver because node was recovered", worker_proc)));
 				break;
 			}
 
-
+			if (timeline != Mtm->nodes[nodeId-1].timeline) {
+                ereport(LOG, (errmsg("%s: restart WAL receiver because node %d timeline is changed", worker_proc, nodeId)));
+                break;
+            }
+				
 			/*
 			 * Receive data.
 			 */