Add mtm.get_last_csn function

Author: knizhnik

contrib/mmts/multimaster--1.0.sql

@@ -31,6 +31,10 @@ CREATE FUNCTION mtm.get_csn(tid xid) RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_csn'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.get_last_csn() RETURNS bigint
+AS 'MODULE_PATHNAME','mtm_get_last_csn'
+LANGUAGE C;
+
 
 CREATE TYPE mtm.node_state AS ("id" integer, "disabled" bool, "disconnected" bool, "catchUp" bool, "slotLag" bigint, "avgTransDelay" bigint, "lastStatusChange" timestamp, "oldestSnapshot" bigint, "connStr" text);
 

contrib/mmts/multimaster.c

@@ -108,6 +108,7 @@ PG_FUNCTION_INFO_V1(mtm_poll_node);
 PG_FUNCTION_INFO_V1(mtm_recover_node);
 PG_FUNCTION_INFO_V1(mtm_get_snapshot);
 PG_FUNCTION_INFO_V1(mtm_get_csn);
+PG_FUNCTION_INFO_V1(mtm_get_last_csn);
 PG_FUNCTION_INFO_V1(mtm_get_nodes_state);
 PG_FUNCTION_INFO_V1(mtm_get_cluster_state);
 PG_FUNCTION_INFO_V1(mtm_get_cluster_info);
@@ -823,11 +824,11 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		if (ts != NULL) { 
 			if (commit) {
 				Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
-				ts->status = TRANSACTION_STATUS_COMMITTED;
 				if (x->csn > ts->csn) {
 					ts->csn = x->csn;
 					MtmSyncClock(ts->csn);
 				}
+				ts->status = TRANSACTION_STATUS_COMMITTED;
 			} else { 
 				ts->status = TRANSACTION_STATUS_ABORTED;
 			}
@@ -1462,6 +1463,7 @@ static void MtmInitialize()
 		Mtm->recoverySlot = 0;
 		Mtm->locks = GetNamedLWLockTranche(MULTIMASTER_NAME);
 		Mtm->csn = MtmGetCurrentTime();
+		Mtm->lastCsn = INVALID_CSN;
 		Mtm->oldestXid = FirstNormalTransactionId;
         Mtm->nLiveNodes = MtmNodes;
         Mtm->nAllNodes = MtmNodes;
@@ -2295,6 +2297,12 @@ mtm_get_snapshot(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(MtmTx.snapshot);
 }
 
+Datum
+mtm_get_last_csn(PG_FUNCTION_ARGS)
+{
+	PG_RETURN_INT64(Mtm->lastCsn);
+}
+
 Datum
 mtm_get_csn(PG_FUNCTION_ARGS)
 {

contrib/mmts/multimaster.h

@@ -172,7 +172,8 @@ typedef struct
 	int    nActiveTransactions;        /* Nunmber of active 2PC transactions */
 	int    nConfigChanges;             /* Number of cluster configuration changes */
 	int64  timeShift;                  /* Local time correction */
-	csn_t  csn;                        /* Last obtained CSN: used to provide unique acending CSNs based on system time */
+	csn_t  csn;                        /* Last obtained timestamp: used to provide unique acending CSNs based on system time */
+	csn_t  lastCsn;                    /* CSN of last committed transaction */
 	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.

contrib/mmts/tests/dtmacid.cpp

@@ -130,8 +130,8 @@ void* reader(void* arg)
 		while ((c2 = random() % conns.size()) == c1);
         work txn1(*conns[c1]);
         work txn2(*conns[c2]);
-        result r1 = txn1.exec("select v,xmin,xmax,mtm.get_csn(xmin) from t order by u");
-        result r2 = txn2.exec("select v,xmin,xmax,mtm.get_csn(xmin) from t order by u");
+        result r1 = txn1.exec("select v,xmin,xmax,mtm.get_csn(xmin),mtm.get_csn(xmax),mtm.get_snapshot(),mtm.get_last_csn() from t order by u");
+        result r2 = txn2.exec("select v,xmin,xmax,mtm.get_csn(xmin),mtm.get_csn(xmax),mtm.get_snapshot(),mtm.get_last_csn() from t order by u");
 		int delta = 0;
 		for (int i=0; i < cfg.nAccounts; i++) { 
 			int diff = r1[i][0].as(int()) - r2[i][0].as(int());
@@ -140,7 +140,9 @@ void* reader(void* arg)
 					delta = diff;
 					if (delta < 0) lt++; else gt++;
 				} else if (delta != diff) { 
-					printf("Inconsistency found for record %d: [%d,%d]->%ld vs [%d,%d]->%ld\n", i, r1[i][1].as(int()), r1[i][2].as(int()), r1[i][3].as(int64_t()), r2[i][1].as(int()), r2[i][2].as(int()), r2[i][3].as(int64_t()));
+					printf("Inconsistency found for record %d: [%d,%d]->[%ld,%ld] (snapshot %ld, last CSN %ld) vs. [%d,%d]->[%ld,%ld] (snapshot %ld, last CSN %ld)\n", i, 
+						   r1[i][1].as(int()), r1[i][2].as(int()), r1[i][3].as(int64_t()), r1[i][4].as(int64_t()), r1[i][5].as(int64_t()), r1[i][6].as(int64_t()), 
+						   r2[i][1].as(int()), r2[i][2].as(int()), r2[i][3].as(int64_t()), r2[i][4].as(int64_t()), r2[i][5].as(int64_t()), r2[i][6].as(int64_t()));
 				}
 			}
 		}