Fix bug in flushed_lsn reporting

Author: knizhnik

multimaster--1.0.sql

@@ -35,7 +35,7 @@ AS 'MODULE_PATHNAME','mtm_get_nodes_state'
 LANGUAGE C;
 
 CREATE TYPE mtm.cluster_state AS ("status" text, "disabledNodeMask" bigint, "disconnectedNodeMask" bigint, "catchUpNodeMask" bigint, "liveNodes" integer, "allNodes" integer, "nActiveQueries" integer, "nPendingQueries" integer, "queueSize" bigint, "transCount" bigint, "timeShift" bigint, "recoverySlot" integer,
-"xidHashSize" bigint, "gidHashSize" bigint, "oldestSnapshot" bigint, "configChanges" integer);
+"xidHashSize" bigint, "gidHashSize" bigint, "oldestXid" integer, "configChanges" integer);
 
 CREATE FUNCTION mtm.get_cluster_state() RETURNS mtm.cluster_state 
 AS 'MODULE_PATHNAME','mtm_get_cluster_state'

multimaster.c

@@ -199,6 +199,7 @@ static int   MtmMinRecoveryLag;
 static int   MtmMaxRecoveryLag;
 static int   Mtm2PCPrepareRatio;
 static int   Mtm2PCMinTimeout;
+static int   MtmGcPeriod;
 static bool  MtmIgnoreTablesWithoutPk;
 
 static ExecutorFinish_hook_type PreviousExecutorFinishHook;
@@ -341,16 +342,20 @@ csn_t MtmTransactionSnapshot(TransactionId xid)
 Snapshot MtmGetSnapshot(Snapshot snapshot)
 {
     snapshot = PgGetSnapshotData(snapshot);
-	RecentGlobalDataXmin = RecentGlobalXmin = Mtm->oldestXid;//MtmAdjustOldestXid(RecentGlobalDataXmin);
+	RecentGlobalDataXmin = RecentGlobalXmin = Mtm->oldestXid;
     return snapshot;
 }
 
 
 TransactionId MtmGetOldestXmin(Relation rel, bool ignoreVacuum)
 {
     TransactionId xmin = PgGetOldestXmin(NULL, false); /* consider all backends */
-    xmin = MtmAdjustOldestXid(xmin);
-    return xmin;
+	if (TransactionIdIsValid(xmin)) { 
+		MtmLock(LW_EXCLUSIVE);
+		xmin = MtmAdjustOldestXid(xmin);
+		MtmUnlock();
+	}
+	return xmin;
 }
 
 bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
@@ -445,53 +450,50 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 static TransactionId 
 MtmAdjustOldestXid(TransactionId xid)
 {
-    if (TransactionIdIsValid(xid)) { 
-        MtmTransState *ts, *prev = NULL;
-        int i;
-
-		MtmLock(LW_EXCLUSIVE);
-        ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
-        if (ts != NULL) { 
-			csn_t oldestSnapshot = ts->snapshot;
-			Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
-			for (i = 0; i < Mtm->nAllNodes; i++) { 
-				if (!BIT_CHECK(Mtm->disabledNodeMask, i)
-					&& Mtm->nodes[i].oldestSnapshot < oldestSnapshot) 
-				{ 
-					oldestSnapshot = Mtm->nodes[i].oldestSnapshot;
-				}
-			}
-			oldestSnapshot -= MtmVacuumDelay*USECS_PER_SEC;
-
-			for (ts = Mtm->transListHead; 
-				 ts != NULL 
-					 && ts->csn < oldestSnapshot
-					 && TransactionIdPrecedes(ts->xid, xid)
-					 && (ts->status == TRANSACTION_STATUS_COMMITTED ||
-						 ts->status == TRANSACTION_STATUS_ABORTED);
-				 prev = ts, ts = ts->next) 
+	int i;   
+	MtmTransState *prev = NULL;
+	MtmTransState *ts = (MtmTransState*)hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+	MTM_LOG1("%d: MtmAdjustOldestXid(%d): snapshot=%ld, csn=%ld, status=%d", MyProcPid, xid, ts != NULL ? ts->snapshot : 0, ts != NULL ? ts->csn : 0, ts != NULL ? ts->status : -1);
+	Mtm->gcCount = 0;
+	if (ts != NULL) { 
+		csn_t oldestSnapshot = ts->snapshot;
+		Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
+		for (i = 0; i < Mtm->nAllNodes; i++) { 
+			if (!BIT_CHECK(Mtm->disabledNodeMask, i)
+				&& Mtm->nodes[i].oldestSnapshot < oldestSnapshot) 
 			{ 
-				if (prev != NULL) { 
-					/* Remove information about too old transactions */
-					hash_search(MtmXid2State, &prev->xid, HASH_REMOVE, NULL);
-				}
+				oldestSnapshot = Mtm->nodes[i].oldestSnapshot;
 			}
-        } 
-		if (MtmUseDtm) 
+		}
+		oldestSnapshot -= MtmVacuumDelay*USECS_PER_SEC;
+		
+		for (ts = Mtm->transListHead; 
+			 ts != NULL 
+				 && ts->csn < oldestSnapshot
+				 && TransactionIdPrecedes(ts->xid, xid)
+				 && (ts->status == TRANSACTION_STATUS_COMMITTED ||
+					 ts->status == TRANSACTION_STATUS_ABORTED);
+			 prev = ts, ts = ts->next) 
 		{ 
 			if (prev != NULL) { 
-				Mtm->transListHead = prev;
-				Mtm->oldestXid = xid = prev->xid;            
-			} else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) {  
-				xid = Mtm->oldestXid;
-			}
-		} else { 
-			if (prev != NULL) { 
-				Mtm->transListHead = prev;
+				/* Remove information about too old transactions */
+				hash_search(MtmXid2State, &prev->xid, HASH_REMOVE, NULL);
 			}
 		}
-		MtmUnlock();
-    }
+	} 
+	if (MtmUseDtm) 
+	{ 
+		if (prev != NULL) { 
+			Mtm->transListHead = prev;
+			Mtm->oldestXid = xid = prev->xid;            
+		} else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) {  
+			xid = Mtm->oldestXid;
+		}
+	} else { 
+		if (prev != NULL) { 
+			Mtm->transListHead = prev;
+		}
+	}
     return xid;
 }
 /*
@@ -613,7 +615,12 @@ static void
 MtmBeginTransaction(MtmCurrentTrans* x)
 {
     if (x->snapshot == INVALID_CSN) { 
-		MtmLock(LW_EXCLUSIVE);
+		TransactionId xmin = (Mtm->gcCount >= MtmGcPeriod) ? PgGetOldestXmin(NULL, false) : InvalidTransactionId; /* Get oldest xmin outside critical section */
+
+		MtmLock(LW_EXCLUSIVE);	
+		if (TransactionIdIsValid(xmin) && Mtm->gcCount >= MtmGcPeriod) {
+			MtmAdjustOldestXid(xmin);
+		}
 		x->xid = GetCurrentTransactionIdIfAny();
         x->isReplicated = false;
         x->isDistributed = MtmIsUserTransaction();
@@ -689,7 +696,6 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	}
 
 	MtmLock(LW_EXCLUSIVE);
-
 	/*
 	 * Check if there is global multimaster lock preventing new transaction from commit to make a chance to wal-senders to catch-up.
 	 * Only "own" transactions are blacked. Transactions replicated from other nodes (including recovered transaction) should be proceeded
@@ -715,8 +721,10 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 
 	x->isPrepared = true;
 	x->csn = ts->csn;
-
+	
 	Mtm->transCount += 1;
+	Mtm->gcCount += 1;
+
 	MtmTransactionListAppend(ts);
 	MtmAddSubtransactions(ts, subxids, ts->nSubxids);
 	MTM_LOG3("%d: MtmPrePrepareTransaction prepare commit of %d (gtid.xid=%d, gtid.node=%d, CSN=%ld)", 
@@ -1465,8 +1473,9 @@ static void MtmInitialize()
         Mtm->transListHead = NULL;
         Mtm->transListTail = &Mtm->transListHead;		
         Mtm->nReceivers = 0;
-		Mtm->timeShift = 0;
+		Mtm->timeShift = 0;		
 		Mtm->transCount = 0;
+		Mtm->gcCount = 0;
 		Mtm->nConfigChanges = 0;
 		Mtm->localTablesHashLoaded = false;
 		for (i = 0; i < MtmNodes; i++) {
@@ -1599,6 +1608,21 @@ _PG_init(void)
 	if (!process_shared_preload_libraries_in_progress)
 		return;
 
+	DefineCustomIntVariable(
+		"multimaster.gc_period",
+		"Number of distributed transactions after which garbage collection is started",
+		"Multimaster is building xid->csn hash map which has to be cleaned to avoid hash overflow. This parameter specifies interval of invoking garbage collector for this map",
+		&MtmGcPeriod,
+		MTM_HASH_SIZE/10,
+		1,
+	    INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.max_nodes",
 		"Maximal number of cluster nodes",
@@ -2338,7 +2362,7 @@ mtm_get_cluster_state(PG_FUNCTION_ARGS)
 	values[11] = Int32GetDatum(Mtm->recoverySlot);
 	values[12] = Int64GetDatum(hash_get_num_entries(MtmXid2State));
 	values[13] = Int64GetDatum(hash_get_num_entries(MtmGid2State));
-	values[14] = Int64GetDatum(Mtm->oldestSnapshot);
+	values[14] = Int32GetDatum(Mtm->oldestXid);
 	values[15] = Int32GetDatum(Mtm->nConfigChanges);
 
 	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(desc, values, nulls)));

multimaster.h

@@ -180,6 +180,7 @@ typedef struct
     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 gcCount;                    /* Number of global transactions performed since last GC */
     BgwPool pool;                      /* Pool of background workers for applying logical replication patches */
 	MtmNodeInfo nodes[1];              /* [Mtm->nAllNodes]: per-node data */ 
 } MtmState;

pglogical_apply.c

@@ -75,8 +75,6 @@ static void process_remote_insert(StringInfo s, Relation rel);
 static void process_remote_update(StringInfo s, Relation rel);
 static void process_remote_delete(StringInfo s, Relation rel);
 
-static int MtmReplicationNode;
-
 /*
  * Search the index 'idxrel' for a tuple identified by 'skey' in 'rel'.
  *
@@ -481,8 +479,8 @@ static void
 MtmBeginSession(void)
 {
 	char slot_name[MULTIMASTER_MAX_SLOT_NAME_SIZE];
-	MtmLockNode(MtmReplicationNode);
-	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, MtmReplicationNode);
+	MtmLockNode(MtmReplicationNodeId);
+	sprintf(slot_name, MULTIMASTER_SLOT_PATTERN, MtmReplicationNodeId);
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 	replorigin_session_origin = replorigin_by_name(slot_name, false); 
 	MTM_LOG3("%d: Begin setup replorigin session: %d", MyProcPid, replorigin_session_origin);
@@ -498,7 +496,7 @@ MtmEndSession(void)
 		replorigin_session_origin = InvalidRepOriginId;
 		replorigin_session_reset();
 		if (unlock) { 
-			MtmUnlockNode(MtmReplicationNode);
+			MtmUnlockNode(MtmReplicationNodeId);
 		}
 		MTM_LOG3("%d: End reset replorigin session: %d", MyProcPid, replorigin_session_origin);
 	}
@@ -513,7 +511,7 @@ process_remote_commit(StringInfo in)
 	XLogRecPtr end_lsn;
 	/* read flags */
 	flags = pq_getmsgbyte(in);
-	MtmReplicationNode = pq_getmsgbyte(in);
+	MtmReplicationNodeId = pq_getmsgbyte(in);
 
 	/* read fields */
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */

pglogical_receiver.c

@@ -47,7 +47,7 @@ static volatile sig_atomic_t got_sighup = false;
 
 /* GUC variables */
 static int receiver_idle_time = 0;
-static bool receiver_sync_mode = false;
+static bool receiver_sync_mode = true; /* We need sync mode to have up-to-date values of catalog_xmin in replication slots */
 
 /* Worker name */
 static char worker_proc[BGW_MAXLEN];

tests/dtmacid.cpp

@@ -285,7 +285,6 @@ int main (int argc, char* argv[])
     for (int i = 0; i < cfg.nReaders; i++) { 
         readers[i].wait();
         nSelects += readers[i].selects;
-        nTransactions += writers[i].transactions;
     }
 
     time_t elapsed = getCurrentTime() - start;