Switch from recovery to normal mode

Author: knizhnik

multimaster.c

@@ -60,11 +60,19 @@ typedef struct {
 	bool  isReplicated;   /* transaction on replica */
 	bool  isDistributed;  /* transaction performed INSERT/UPDATE/DELETE and has to be replicated to other nodes */
 	bool  containsDML;    /* transaction contains DML statements */
+	bool  isPrepared;     /* transaction was prepared for commit */
     csn_t snapshot;       /* transaction snaphsot   */
 } MtmCurrentTrans;
 
 /* #define USE_SPINLOCK 1 */
 
+typedef enum 
+{
+	HASH_LOCK_ID,
+	COMMIT_LOCK_ID,
+	N_LOCKS
+} MtmLockIds;
+
 #define MTM_SHMEM_SIZE (64*1024*1024)
 #define MTM_HASH_SIZE  100003
 #define USEC 1000000
@@ -150,7 +158,7 @@ void MtmLock(LWLockMode mode)
 #ifdef USE_SPINLOCK
 	SpinLockAcquire(&dtm->spinlock);
 #else
-	LWLockAcquire(dtm->hashLock, mode);
+	LWLockAcquire(dtm->locks[HASH_LOCK_ID], mode);
 #endif
 }
 
@@ -159,7 +167,7 @@ void MtmUnlock(void)
 #ifdef USE_SPINLOCK
 	SpinLockRelease(&dtm->spinlock);
 #else
-	LWLockRelease(dtm->hashLock);
+	LWLockRelease(dtm->locks[HASH_LOCK_ID]);
 #endif
 }
 
@@ -412,7 +420,7 @@ static void MtmInitialize()
 	{
 		dtm->status = MTM_INITIALIZATION;
 		dtm->recoverySlot = 0;
-		dtm->hashLock = (LWLock*)GetNamedLWLockTranche(MULTIMASTER_NAME);
+		dtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);
 		dtm->csn = MtmGetCurrentTime();
 		dtm->oldestXid = FirstNormalTransactionId;
         dtm->nNodes = MtmNodes;
@@ -423,6 +431,7 @@ static void MtmInitialize()
         dtm->transListTail = &dtm->transListHead;		
         dtm->nReceivers = 0;
 		dtm->timeShift = 0;
+		pg_atomic_write_u32(&dtm->nCommittingTrans, 0);
 		PGSemaphoreCreate(&dtm->votingSemaphore);
 		PGSemaphoreReset(&dtm->votingSemaphore);
 		SpinLockInit(&dtm->spinlock);
@@ -467,6 +476,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 		x->xid = GetCurrentTransactionIdIfAny();
         x->isReplicated = false;
         x->isDistributed = IsNormalProcessingMode() && dtm->status == MTM_ONLINE && MtmDoReplication && !am_walsender && !IsBackgroundWorker && !IsAutoVacuumWorkerProcess();
+		x->isPrepared = false;
 		x->containsDML = false;
         x->snapshot = MtmAssignCSN();	
 		x->gtid.xid = InvalidTransactionId;
@@ -476,6 +486,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
     }
 }
 
+
 /* 
  * We need to pass snapshot to WAL-sender, so create record in transaction status hash table 
  * before commit
@@ -488,8 +499,14 @@ static void MtmPrepareTransaction(MtmCurrentTrans* x)
 	if (!x->isDistributed) {
 		return;
 	}
-	x->xid = GetCurrentTransactionId();
+	/* Check that commits are not disabled */
+	LWLockAcquire(dtm->locks[COMMIT_LOCK_ID], LW_SHARED);		
+	LWLockRelease(dtm->locks[COMMIT_LOCK_ID]);
 
+	pg_atomic_fetch_add_u32(dtm->nCommittingTransactions, 1);
+	x->isPrepared = true;
+	x->xid = GetCurrentTransactionId();
+				  
 	MtmLock(LW_EXCLUSIVE);
 	ts = hash_search(xid2state, &x->xid, HASH_ENTER, NULL);
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;	
@@ -500,6 +517,7 @@ static void MtmPrepareTransaction(MtmCurrentTrans* x)
 	ts->procno = MyProc->pgprocno;
 	ts->nVotes = 0; 
 	ts->done = false;
+				  
 	if (TransactionIdIsValid(x->gtid.xid)) { 
 		ts->gtid = x->gtid;
 	} else { 
@@ -528,6 +546,9 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		MtmAdjustSubtransactions(ts);
 		MtmUnlock();
 	}
+	if (x->isPrepared) { 
+		pg_atomic_fetch_add_u32(dtm->nCommittingTransactions, -1);
+	}
 	x->snapshot = INVALID_CSN;
 	x->xid = InvalidTransactionId;
 	x->gtid.xid = InvalidTransactionId;
@@ -547,6 +568,39 @@ void MtmSendNotificationMessage(MtmTransState* ts)
 	}
 }
 
+void  MtmUpdateStatus(bool recovered)
+{
+	if (dtm->status == MTM_RECOVERY) { 
+		MtmLock(LW_EXCLUSIVE);		
+		dtm->status = MTM_ONLINE; /* Is it all we shoudl do t switch to nortmal state */
+		MtmUnlock();
+	}
+}
+
+void MtmRecoveryCompleted(int nodeId)
+{
+	if (BIT_CHECK(dtm->pglogicalNodeMask, nodeId-1)) { 
+		if (MyWalSnd->sentPtr == GetXLogInsertRecPtr()) { 
+			/* Ok, now we done with recovery of node */
+			MtmLock(LW_EXCLUSIVE);
+			dtm->pglogicalNodeMask &= (int64)1 << (nodeId-1); /* now node is assumed as recovered */
+			dtm->nNodes += 1;
+			MtmUnlock();
+
+			LWLockRelease(dtm->locks[COMMIT_LOCK_ID]); /* enable commits */
+
+			return true;
+		} else if (MyWalSnd->sentPtr + MtmSlotDelayThreashold > GetXLogInsertRecPtr()) { 
+			/* we almost done with recovery of node.. */
+			LWLockAcquire(dtm->locks[COMMIT_LOCK_ID], LW_EXCLUSIVE); /* disable new commits */
+		}
+		return false;
+	} else { 
+		return true;
+	}
+}
+
+
 static bool
 MtmCommitTransaction(TransactionId xid, int nsubxids, TransactionId *subxids)
 {
@@ -803,7 +857,7 @@ _PG_init(void)
 	 * resources in mtm_shmem_startup().
 	 */
 	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmQueueSize);
-	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1);
+	RequestNamedLWLockTranche(MULTIMASTER_NAME, N_LOCKS);
 
     MtmNodes = MtmStartReceivers(MtmConnStrs, MtmNodeId);
     if (MtmNodes < 2) { 

multimaster.h

@@ -14,10 +14,12 @@
 
 #define BIT_CHECK(mask, bit) ((mask) & ((int64)1 << (bit)))
 
-#define MULTIMASTER_NAME           "mtm"
-#define MULTIMASTER_SCHEMA_NAME    "mtm"
-#define MULTIMASTER_DDL_TABLE      "ddl_log"
-#define MULTIMASTER_SLOT_PATTERN   "mtm_slot_%d"
+#define MULTIMASTER_NAME                "mtm"
+#define MULTIMASTER_SCHEMA_NAME         "mtm"
+#define MULTIMASTER_DDL_TABLE           "ddl_log"
+#define MULTIMASTER_SLOT_PATTERN        "mtm_slot_%d"
+#define MULTIMASTER_MIN_PROTO_VERSION   1
+#define MULTIMASTER_MAX_PROTO_VERSION   1
 
 #define Natts_mtm_ddl_log 2
 #define Anum_mtm_ddl_log_issued		1
@@ -91,14 +93,15 @@ typedef struct
 	int recoverySlot;                  /* NodeId of recovery slot or 0 if none */
 	volatile slock_t spinlock;         /* spinlock used to protect access to hash table */
 	PGSemaphoreData votingSemaphore;   /* semaphore used to notify mtm-sender about new responses to coordinator */
-	LWLockId hashLock;                 /* lock to synchronize access to hash table */
+	LWLockPadded *locks;               /* multimaster lock tranche */
 	TransactionId oldestXid;           /* XID of oldest transaction visible by any active transaction (local or global) */
 	int64  disabledNodeMask;           /* bitmask of disabled nodes (so no more than 64 nodes in multimaster:) */
 	int64  pglogicalNodeMask;          /* bitmask of started pglogic receviers */
     int    nNodes;                     /* number of active nodes */
     int    nReceivers;                 /* number of initialized logical receivers (used to determine moment when Mtm intialization is completed */
 	long   timeShift;                  /* local time correction */
 	csn_t  csn;                        /* last obtained CSN: used to provide unique acending CSNs based on system time */
+	pg_atomic_uint32 nCommittingTrans; /* nubmer of transactions i process of commit */
 	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.
@@ -137,6 +140,7 @@ extern void  MtmUnlock(void);
 extern void  MtmDropNode(int nodeId, bool dropSlot);
 extern MtmState* MtmGetState(void);
 extern timestamp_t MtmGetCurrentTime(void);
-extern void MtmSleep(timestamp_t interval);
-
+extern void  MtmSleep(timestamp_t interval);
+extern void  MtmRecoveryCompleted(int nodeId);
+extern void  MtmUpdateStatus(bool recovered);
 #endif

pglogical_apply.c

@@ -466,7 +466,9 @@ read_rel(StringInfo s, LOCKMODE mode)
 static void
 process_remote_commit(StringInfo s)
 {
+	bool recovered = pq_getmsgbyte(s);
 	CommitTransactionCommand();
+	MtmUpdateStatus(recovered);
 }
 
 static void

pglogical_proto.c

@@ -129,6 +129,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
     if (!mm->isLocal) { 
         pq_sendbyte(out, 'C');		/* sending COMMIT */
+        pq_sendbyte(out, MtmRecoveryCompleted(mm->nodeId));
     }
 }
 

pglogical_receiver.c

@@ -31,6 +31,7 @@
 #include "utils/guc.h"
 #include "utils/snapmgr.h"
 #include "executor/spi.h"
+#include "replication/origin.h"
 
 #include "multimaster.h"
 
@@ -203,10 +204,12 @@ pglogical_receiver_main(Datum main_arg)
 	PGconn *conn;
 	PGresult *res;
 	MtmSlotMode mode;
+	MtmState* ds;
 #ifndef USE_PGLOGICAL_OUTPUT
     bool insideTrans = false;
 #endif
     ByteBuffer buf;
+	XLogRecPtr originStartPos;
 
 	/* Register functions for SIGTERM/SIGHUP management */
 	pqsignal(SIGHUP, receiver_raw_sighup);
@@ -258,8 +261,14 @@ pglogical_receiver_main(Datum main_arg)
 		resetPQExpBuffer(query);
 	}
 	/* Start logical replication at specified position */
-	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL 0/0 (\"startup_params_format\" '1', \"max_proto_version\" '1',  \"min_proto_version\" '1')",
-					  args->receiver_slot);
+	originStartPos = replorigin_session_get_progress(false);
+	appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %u/%u (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d')",
+					  args->receiver_slot,
+					  (uint32) (originStartPos >> 32),
+					  (uint32) originStartPos,
+					  MULTIMASTER_MAX_PROTO_VERSION,
+					  MULTIMASTER_MIN_PROTO_VERSION
+		);
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_COPY_BOTH)
 	{
@@ -273,6 +282,7 @@ pglogical_receiver_main(Datum main_arg)
 
     MtmReceiverStarted(args->receiver_node);
     ByteBufferAlloc(&buf);
+	ds = MtmGetState();
 
 	while (!got_sigterm)
 	{
@@ -366,7 +376,7 @@ pglogical_receiver_main(Datum main_arg)
 				 * If sync mode is sent reply in all cases to ensure that
 				 * server knows how far replay has been done.
 				 */
-				if (replyRequested || receiver_sync_mode)
+				if (replyRequested || receiver_sync_mode || ds->status == MTM_RECOVERY)
 				{
 					int64 now = feGetCurrentTimestamp();