recovery in progress

Author: knizhnik

multimaster.c

@@ -695,6 +695,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	ts->votingCompleted = false;
 	ts->cmd = MSG_INVALID;
 	ts->nSubxids = xactGetCommittedChildren(&subxids);
+	Mtm->nActiveTransactions += 1;
 
 	x->isPrepared = true;
 	x->csn = ts->csn;
@@ -794,6 +795,8 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 				ts->status = TRANSACTION_STATUS_ABORTED;
 			}
 			MtmAdjustSubtransactions(ts);
+			Assert(Mtm->nActiveTransactions != 0);
+			Mtm->nActiveTransactions -= 1;
 		}
 		if (!commit && x->isReplicated && TransactionIdIsValid(x->gtid.xid)) { 
 			/* 
@@ -835,6 +838,13 @@ void MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd)
 	}
 }
 
+void MtmRecoveryCompleted(void)
+{
+	elog(WARNING, "Recevoery of node %d is completed", MtmNodeId);
+	Mtm->recoverySlot = 0;
+	MtmSwitchClusterMode(MTM_ONLINE);
+}
+
 void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 {
 	MtmLock(LW_EXCLUSIVE);
@@ -846,8 +856,7 @@ void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 		Assert(Mtm->status == MTM_RECOVERY);
 	} else if (Mtm->status == MTM_RECOVERY) { 
 		/* When recovery is completed we get normal transaction ID and switch to normal mode */
-		Mtm->recoverySlot = 0;
-		MtmSwitchClusterMode(MTM_ONLINE);
+		MtmRecoveryCompleted();
 	}
 	MtmTx.gtid = *gtid;
 	MtmTx.xid = GetCurrentTransactionId();
@@ -972,35 +981,52 @@ static int64 MtmGetSlotLag(int nodeId)
  */
 bool MtmIsRecoveredNode(int nodeId)
 {
-	return BIT_CHECK(Mtm->disabledNodeMask, nodeId-1));
+	return BIT_CHECK(Mtm->disabledNodeMask, nodeId-1);
 }
 
 
-void MtmRecoveryPorgress(XLogRecPtr lsn)
+bool MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN)
 {
-	
-		Assert(MyWalSnd != NULL); /* This function is called by WAL-sender, so it should not be NULL */
-		if (!BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)
-			&& MyWalSnd->sentPtr + MtmMinRecoveryLag > GetXLogInsertRecPtr()) 
+	bool caughtUp = false;
+	if (MtmIsRecoveredNode(nodeId)) { 
+		XLogRecPtr walLSN = GetXLogInsertRecPtr();
+		MtmLock(LW_EXCLUSIVE);
+		if (slotLSN == walLSN) {
+			if (BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)) { 
+				elog(WARNING,"Node %d is caught-up", nodeId);	
+				BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
+				BIT_CLEAR(Mtm->walSenderLockerMask, MyWalSnd - WalSndCtl->walsnds);
+				BIT_CLEAR(Mtm->nodeLockerMask, nodeId-1);
+				Mtm->nLockers -= 1;
+			} else { 
+				elog(WARNING,"Node %d is caugth-up without locking cluster", nodeId);	
+				/* We are lucky: caugth-up without locking cluster! */
+				Mtm->nNodes += 1;
+				BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
+			}
+			caughtUp = true;
+		} else if (!BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)
+				   && slotLSN + MtmMinRecoveryLag > walLSN) 
 		{ 
 			/*
 			 * Wal sender almost catched up.
 			 * Lock cluster preventing new transaction to start until wal is completely replayed.
 			 * We have to maintain two bitmasks: one is marking wal sender, another - correspondent nodes. 
 			 * Is there some better way to establish mapping between nodes ad WAL-seconder?
 			 */
-			elog(WARNING,"Node %d is catching up", nodeId);
-			MtmLock(LW_EXCLUSIVE);
+			elog(WARNING,"Node %d is almost caught-up: lock cluster", nodeId);
+			Assert(MyWalSnd != NULL); /* This function is called by WAL-sender, so it should not be NULL */
 			BIT_SET(Mtm->nodeLockerMask, nodeId-1);
 			BIT_SET(Mtm->walSenderLockerMask, MyWalSnd - WalSndCtl->walsnds);
 			Mtm->nLockers += 1;
-			MtmUnlock();
 		} else { 
-			MTM_INFO("Continue recovery of node %d, slot position %lx, WAL position %lx, lockers %d\n", nodeId, MyWalSnd->sentPtr, GetXLogInsertRecPtr(), Mtm->nLockers);			
+			MTM_INFO("Continue recovery of node %d, slot position %lx, WAL position %lx, WAL sender position %lx, lockers %d, active transactions %d\n", nodeId, slotLSN, walLSN, MyWalSnd->sentPtr, Mtm->nLockers, Mtm->nActiveTransactions);
 		}
-		return true;
+		MtmUnlock();
+	} else { 
+		MTM_INFO("Node %d is not in recovery mode\n", nodeId);
 	}
-	return false;
+	return caughtUp;
 }
 
 void MtmSwitchClusterMode(MtmNodeStatus mode)
@@ -1019,22 +1045,24 @@ void MtmSwitchClusterMode(MtmNodeStatus mode)
 static void 
 MtmCheckClusterLock()
 {	
+	timestamp_t delay = MIN_WAIT_TIMEOUT;
 	while (true)
 	{
 		nodemask_t mask = Mtm->walSenderLockerMask;
 		if (mask != 0) {
-			XLogRecPtr currLogPos = GetXLogInsertRecPtr();
-			int i;
-			timestamp_t delay = MIN_WAIT_TIMEOUT;
-			for (i = 0; mask != 0; i++, mask >>= 1) { 
-				if (mask & 1) { 
-					if (WalSndCtl->walsnds[i].sentPtr != currLogPos) {
-						/* recovery is in progress */
-						break;
-					} else { 
-						/* recovered replica catched up with master */
-						elog(WARNING, "WAL-sender %d complete recovery", i);
-						BIT_CLEAR(Mtm->walSenderLockerMask, i);
+			if (Mtm->nActiveTransactions == 0) { 
+				XLogRecPtr currLogPos = GetXLogInsertRecPtr();
+				int i;
+				for (i = 0; mask != 0; i++, mask >>= 1) { 
+					if (mask & 1) { 
+						if (WalSndCtl->walsnds[i].sentPtr != currLogPos) {
+							/* recovery is in progress */
+							break;
+						} else { 
+							/* recovered replica catched up with master */
+							elog(WARNING, "WAL-sender %d complete recovery", i);
+							BIT_CLEAR(Mtm->walSenderLockerMask, i);
+						}
 					}
 				}
 			}
@@ -1294,6 +1322,7 @@ static void MtmInitialize()
 		Mtm->walSenderLockerMask = 0;
 		Mtm->nodeLockerMask = 0;
 		Mtm->nLockers = 0;
+		Mtm->nActiveTransactions = 0;
 		Mtm->votingTransactions = NULL;
         Mtm->transListHead = NULL;
         Mtm->transListTail = &Mtm->transListHead;		
@@ -1734,12 +1763,31 @@ void MtmDropNode(int nodeId, bool dropSlot)
 static void 
 MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 {
+	ListCell *param;
+	bool isRecoverySession = false;
+	foreach(param, args->in_params)
+	{
+		DefElem    *elem = lfirst(param);
+		if (strcmp("mtm_replication_mode", elem->defname) == 0) { 
+			isRecoverySession = elem->arg != NULL && strVal(elem->arg) != NULL && strcmp(strVal(elem->arg), "recovery") == 0;
+			break;
+		}
+	}
 	MtmLock(LW_EXCLUSIVE);
-	if (BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
-		elog(WARNING, "Recovery of node %d is completed: start normal replication", MtmReplicationNodeId); 
+	if (isRecoverySession) {
+		elog(WARNING, "Node %d start recovery of node %d", MtmNodeId, MtmReplicationNodeId);
+		if (!BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
+			BIT_SET(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
+			Mtm->nNodes -= 1;			
+			MtmCheckQuorum();
+		}
+	} else if (BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
+		elog(WARNING, "Node %d consider that recovery of node %d is completed: start normal replication", MtmNodeId, MtmReplicationNodeId); 
 		BIT_CLEAR(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
 		Mtm->nNodes += 1;
 		MtmCheckQuorum();
+	} else {
+		elog(NOTICE, "Node %d start logical replication to node %d in normal mode", MtmNodeId, MtmReplicationNodeId); 
 	}
 	MtmUnlock();
 }
@@ -1757,7 +1805,7 @@ MtmReplicationTxnFilterHook(struct PGLogicalTxnFilterArgs* args)
 	bool res = Mtm->status != MTM_RECOVERY
 		&& (args->origin_id == InvalidRepOriginId 
 			|| MtmIsRecoveredNode(MtmReplicationNodeId));
-	MTM_TRACE("%d: MtmReplicationTxnFilterHook->%d\n", MyProcPid, res);
+	MTM_INFO("%d: MtmReplicationTxnFilterHook->%d\n", MyProcPid, res);
 	return res;
 }
 

multimaster.h

@@ -130,18 +130,19 @@ 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 */
-    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 */
-	long   timeShift;                  /* local time correction */
-	csn_t  csn;                        /* last obtained CSN: used to provide unique acending CSNs based on system time */
+    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 */
+	int    nActiveTransactions;        /* Nunmber of active 2PC transactions */
+	long   timeShift;                  /* Local time correction */
+	csn_t  csn;                        /* Last obtained CSN: used to provide unique acending CSNs based on system time */
 	MtmTransState* votingTransactions; /* L1-list of replicated transactions sendings notifications to coordinator.
 									 	 This list is used to pass information to mtm-sender BGW */
     MtmTransState* transListHead;      /* L1 list of all finished transactions present in xid2state hash.
 									 	  It is cleanup by MtmGetOldestXmin */
     MtmTransState** transListTail;     /* Tail of L1 list of all finished transactionds, used to append new elements.
 								  		  This list is expected to be in CSN ascending order, by strict order may be violated */
-	uint64 transCount;                 /* Counter of transactions perfromed by this node */
+	uint64 transCount;                 /* Counter of transactions perfromed by this node */	
 	time_t nodeTransDelay[MAX_NODES];  /* Time of waiting transaction acknowledgment from node */
     BgwPool pool;                      /* Pool of background workers for applying logical replication patches */
 	MtmNodeInfo nodes[1];              /* [MtmNodes]: per-node data */ 
@@ -200,5 +201,6 @@ extern void  MtmSwitchClusterMode(MtmNodeStatus mode);
 extern void  MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr);
 extern void  MtmSetupReplicationHooks(struct PGLogicalHooks* hooks);
 extern void  MtmCheckQuorum(void);
-
+extern bool  MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN);
+extern void  MtmRecoveryCompleted(void);
 #endif

pglogical_apply.c

@@ -497,10 +497,12 @@ process_remote_commit(StringInfo in)
 	uint8 		flags;
 	csn_t       csn;
 	const char *gid = NULL;	
+	bool        caughtUp;
 
 	/* read flags */
 	flags = pq_getmsgbyte(in);
 	MtmReplicationNode = pq_getmsgbyte(in);
+	caughtUp = pq_getmsgbyte(in) != 0;
 
 	/* read fields */
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
@@ -568,7 +570,10 @@ process_remote_commit(StringInfo in)
 		default:
 			Assert(false);
 	}
-	MtmEndSession();
+	MtmEndSession(true);
+	if (caughtUp) {
+		MtmRecoveryCompleted();
+	}
 }
 
 static void

pglogical_proto.c

@@ -150,11 +150,12 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
-	MTM_TRACE("PGLOGICAL_SEND commit: event=%d, gid=%s\n", flags, txn->gid);
+	MTM_INFO("PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx\n", flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());
 
     /* send the flags field */
     pq_sendbyte(out, flags);
     pq_sendbyte(out, MtmNodeId);
+    pq_sendbyte(out, MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn));
 
     /* send fixed fields */
     pq_sendint64(out, commit_lsn);
@@ -167,7 +168,6 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     if (flags != PGLOGICAL_COMMIT) { 
     	pq_sendstring(out, txn->gid);
 	}
-
 }
 
 /*

pglogical_receiver.c

@@ -292,12 +292,13 @@ pglogical_receiver_main(Datum main_arg)
 	}
 	CommitTransactionCommand();
 
-	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %u/%u (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d')",
+	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %u/%u (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"mtm_replication_mode\" '%s')",
 					  args->receiver_slot,
 					  (uint32) (originStartPos >> 32),
 					  (uint32) originStartPos,
 					  MULTIMASTER_MAX_PROTO_VERSION,
-					  MULTIMASTER_MIN_PROTO_VERSION
+					  MULTIMASTER_MIN_PROTO_VERSION,
+					  mode == SLOT_OPEN_EXISTED ? "recovery" : "normal"					  
 		);
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_COPY_BOTH)