Add switch from recovery to normal mode

Author: knizhnik

arbiter.c

@@ -83,7 +83,7 @@ typedef struct
 	TransactionId  dxid; /* Transaction ID at destination node */
 	TransactionId  sxid; /* Transaction IO at sender node */  
 	csn_t          csn;  /* local CSN in case of sending data from replica to master, global CSN master->replica */
-	int64          disabledNodeMask; /* bitmask of disabled nodes at the sender of message */
+	nodemask_t     disabledNodeMask; /* bitmask of disabled nodes at the sender of message */
 } MtmArbiterMessage;
 
 typedef struct 

multimaster.c

@@ -45,6 +45,7 @@
 #include "storage/proc.h"
 #include "utils/syscache.h"
 #include "replication/walsender.h"
+#include "replication/walsender_private.h"
 #include "replication/slot.h"
 #include "port/atomics.h"
 #include "tcop/utility.h"
@@ -60,16 +61,14 @@ 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,
+	MTM_STATE_LOCK_ID,
 	N_LOCKS
 } MtmLockIds;
 
@@ -142,6 +141,7 @@ int   MtmReconnectAttempts;
 static int MtmQueueSize;
 static int MtmWorkers;
 static int MtmVacuumDelay;
+static int MtmMinRecoveryLag;
 
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
 static ProcessUtility_hook_type PreviousProcessUtilityHook;
@@ -158,7 +158,7 @@ void MtmLock(LWLockMode mode)
 #ifdef USE_SPINLOCK
 	SpinLockAcquire(&dtm->spinlock);
 #else
-	LWLockAcquire(dtm->locks[HASH_LOCK_ID], mode);
+	LWLockAcquire((LWLockId)&dtm->locks[MTM_STATE_LOCK_ID], mode);
 #endif
 }
 
@@ -167,7 +167,7 @@ void MtmUnlock(void)
 #ifdef USE_SPINLOCK
 	SpinLockRelease(&dtm->spinlock);
 #else
-	LWLockRelease(dtm->locks[HASH_LOCK_ID]);
+	LWLockRelease((LWLockId)&dtm->locks[MTM_STATE_LOCK_ID]);
 #endif
 }
 
@@ -426,12 +426,14 @@ static void MtmInitialize()
         dtm->nNodes = MtmNodes;
 		dtm->disabledNodeMask = 0;
 		dtm->pglogicalNodeMask = 0;
+		dtm->walSenderLockerMask = 0;
+		dtm->nodeLockerMask = 0;
+		dtm->nLockers = 0;
 		dtm->votingTransactions = NULL;
         dtm->transListHead = NULL;
         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);
@@ -476,7 +478,6 @@ 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;
@@ -487,6 +488,53 @@ MtmBeginTransaction(MtmCurrentTrans* x)
 }
 
 
+/* This function is called at transaction start with multimaster ock set */
+static void 
+MtmCheckClusterLock()
+{	
+	while (true)
+	{
+		nodemask_t mask = dtm->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 */
+						dtm->walSenderLockerMask &= ~((nodemask_t)1 << i);
+					}
+				}
+			}
+			if (mask != 0) { 
+				/* some "almost catch-up" wal-senders are still working */
+				/* Do not start new transactions until them complete */
+				MtmUnlock();
+				MtmSleep(delay);
+				if (delay*2 <= MAX_WAIT_TIMEOUT) { 
+					delay *= 2;
+				}
+				MtmLock(LW_EXCLUSIVE);
+				continue;
+			} else {  
+				/* All lockers are synchronized their logs */
+				/* Remove lock and mark them as receovered */
+				Assert(dtm->walSenderLockerMask == 0);
+				Assert((dtm->nodeLockerMask & dtm->disabledNodeMask) == dtm->nodeLockerMask);
+				dtm->disabledNodeMask &= ~dtm->nodeLockerMask;
+				dtm->nNodes += dtm->nLockers;
+				dtm->nLockers = 0;
+				dtm->nodeLockerMask = 0;
+			}
+		}
+		break;
+	}
+}	
+
 /* 
  * We need to pass snapshot to WAL-sender, so create record in transaction status hash table 
  * before commit
@@ -499,15 +547,12 @@ static void MtmPrepareTransaction(MtmCurrentTrans* x)
 	if (!x->isDistributed) {
 		return;
 	}
-	/* 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);
+	MtmCheckClusterLock();
+
 	ts = hash_search(xid2state, &x->xid, HASH_ENTER, NULL);
 	ts->status = TRANSACTION_STATUS_IN_PROGRESS;	
 	ts->snapshot = x->isReplicated || !x->containsDML ? INVALID_CSN : x->snapshot;
@@ -546,9 +591,6 @@ 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;
@@ -568,39 +610,29 @@ 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)
+/*
+ * This function is called by WAL sender when start sending new transaction
+ */
+bool MtmIsRecoveredNode(int nodeId)
 {
-	if (BIT_CHECK(dtm->pglogicalNodeMask, nodeId-1)) { 
-		if (MyWalSnd->sentPtr == GetXLogInsertRecPtr()) { 
-			/* Ok, now we done with recovery of node */
+	if (BIT_CHECK(dtm->disabledNodeMask, nodeId-1)) { 
+		Assert(MyWalSnd != NULL);		
+		if (!BIT_CHECK(dtm->nodeLockerMask, nodeId-1)
+			&& MyWalSnd->sentPtr + MtmMinRecoveryLag > GetXLogInsertRecPtr()) 
+		{ 
+			/* Wal sender almost catched up */
+			/* Lock cluster preventing new transaction to start until wal is completely replayed */
 			MtmLock(LW_EXCLUSIVE);
-			dtm->pglogicalNodeMask &= (int64)1 << (nodeId-1); /* now node is assumed as recovered */
-			dtm->nNodes += 1;
+			dtm->nodeLockerMask |= (nodemask_t)1 << (nodeId-1);
+			dtm->walSenderLockerMask |= (nodemask_t)1 << (MyWalSnd - WalSndCtl->walsnds);
+			dtm->nLockers += 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;
 	}
+	return false;
 }
 
-
 static bool
 MtmCommitTransaction(TransactionId xid, int nsubxids, TransactionId *subxids)
 {
@@ -717,6 +749,21 @@ _PG_init(void)
 	if (!process_shared_preload_libraries_in_progress)
 		return;
 
+	DefineCustomIntVariable(
+		"multimaster.min_recovery_lag",
+		"Minamal lag of WAL-sender performing recovery after which cluster is locked until recovery is completed",
+		NULL,
+		&MtmMinRecoveryLag,
+		100000,
+		1,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.vacuum_delay",
 		"Minimal age of records which can be vacuumed (seconds)",
@@ -897,6 +944,12 @@ _PG_fini(void)
  *  ***************************************************************************
  */
 
+static void MtmSwitchFromRecoveryToNormalMode()
+{
+	dtm->status = MTM_ONLINE;
+	/* ??? Something else to do here? */
+}
+
 void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 {
 	csn_t localSnapshot;
@@ -910,6 +963,11 @@ void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 		elog(ERROR, "Too old snapshot: requested %ld, current %ld", globalSnapshot, localSnapshot);
 	}
 
+	if (!TransactionIdIsValid(gtid->xid)) { 
+		Assert(dtm->status == MTM_RECOVERY);
+	} else if (dtm->status == MTM_RECOVERY) { 
+		MtmSwitchFromRecoveryToNormalMode();
+	}
 	dtmTx.gtid = *gtid;
 	dtmTx.xid = GetCurrentTransactionId();
 	dtmTx.snapshot = globalSnapshot;	

multimaster.h

@@ -12,7 +12,7 @@
 #define MTM_TUPLE_TRACE(fmt, ...) fprintf(stderr, fmt, ## __VA_ARGS__) 
 */
 
-#define BIT_CHECK(mask, bit) ((mask) & ((int64)1 << (bit)))
+#define BIT_CHECK(mask, bit) (((mask) & ((nodemask_t)1 << (bit))) != 0)
 
 #define MULTIMASTER_NAME                "mtm"
 #define MULTIMASTER_SCHEMA_NAME         "mtm"
@@ -30,6 +30,7 @@ typedef uint64 csn_t; /* commit serial number */
 #define INVALID_CSN  ((csn_t)-1)
 
 typedef uint64 timestamp_t;
+typedef uint64_t nodemask_t;
 
 /* Identifier of global transaction */
 typedef struct 
@@ -95,13 +96,15 @@ typedef struct
 	PGSemaphoreData votingSemaphore;   /* semaphore used to notify mtm-sender about new responses to coordinator */
 	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 */
+	nodemask_t disabledNodeMask;       /* bitmask of disabled nodes (so no more than 64 nodes in multimaster:) */
+	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 */
-	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.
@@ -141,6 +144,5 @@ extern void  MtmDropNode(int nodeId, bool dropSlot);
 extern MtmState* MtmGetState(void);
 extern timestamp_t MtmGetCurrentTime(void);
 extern void  MtmSleep(timestamp_t interval);
-extern void  MtmRecoveryCompleted(int nodeId);
-extern void  MtmUpdateStatus(bool recovered);
+extern bool  MtmIsRecoveredNode(int nodeId);
 #endif

pglogical_apply.c

@@ -466,9 +466,7 @@ 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

@@ -39,7 +39,6 @@
 typedef struct PGLogicalProtoMM
 {
     PGLogicalProtoAPI api;
-	MtmState* state;
 	int  nodeId;
     bool isLocal;
 } PGLogicalProtoMM;
@@ -107,14 +106,15 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 {
     PGLogicalProtoMM* mm = (PGLogicalProtoMM*)data->api;
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
+	bool isRecovery = MtmIsRecoveredNode(mm->nodeId);
 	MTM_TRACE("pglogical_write_begin %d CSN=%ld\n", txn->xid, csn);
-    if (csn == INVALID_CSN || BIT_CHECK(mm->state->disabledNodeMask, mm->nodeId-1)) {
+    if (csn == INVALID_CSN && !isRecovery) {
         mm->isLocal = true;
     } else { 
         mm->isLocal = false;        
         pq_sendbyte(out, 'B');		/* BEGIN */
 		pq_sendint(out, MtmNodeId, 4);
-		pq_sendint(out, txn->xid, 4);
+		pq_sendint(out, isRecovery ? InvalidTransactionId : txn->xid, 4);
         pq_sendint64(out, csn);
     }
 }
@@ -129,7 +129,6 @@ 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));
     }
 }
 
@@ -380,7 +379,6 @@ pglogical_init_api(PGLogicalProtoType typ)
 	PGLogicalProtoMM* pmm = palloc0(sizeof(PGLogicalProtoMM));
     PGLogicalProtoAPI* res = &pmm->api;
     pmm->isLocal = false;
-	pmm->state = MtmGetState();
 	sscanf(MyReplicationSlot->data.name.data, MULTIMASTER_SLOT_PATTERN, &pmm->nodeId);
     res->write_rel = pglogical_write_rel;
     res->write_begin = pglogical_write_begin;