Add mtm.get_csn function

Author: knizhnik

multimaster--1.0.sql

@@ -27,6 +27,10 @@ CREATE FUNCTION mtm.get_snapshot() RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.get_csn(tid xid) RETURNS bigint
+AS 'MODULE_PATHNAME','mtm_get_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);
 

multimaster.c

@@ -106,6 +106,7 @@ PG_FUNCTION_INFO_V1(mtm_add_node);
 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_nodes_state);
 PG_FUNCTION_INFO_V1(mtm_get_cluster_state);
 PG_FUNCTION_INFO_V1(mtm_get_cluster_info);
@@ -451,12 +452,13 @@ static TransactionId
 MtmAdjustOldestXid(TransactionId xid)
 {
 	int i;   
+	csn_t oldestSnapshot = INVALID_CSN;
 	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;
+		oldestSnapshot = ts->snapshot;
 		Mtm->nodes[MtmNodeId-1].oldestSnapshot = oldestSnapshot;
 		for (i = 0; i < Mtm->nAllNodes; i++) { 
 			if (!BIT_CHECK(Mtm->disabledNodeMask, i)
@@ -486,6 +488,7 @@ MtmAdjustOldestXid(TransactionId xid)
 		if (prev != NULL) { 
 			Mtm->transListHead = prev;
 			Mtm->oldestXid = xid = prev->xid;            
+			MTM_LOG1("%d: MtmAdjustOldestXid: oldestXid=%d, olderstSnapshot=%ld", MyProcPid, xid, oldestSnapshot);
 		} else if (TransactionIdPrecedes(Mtm->oldestXid, xid)) {  
 			xid = Mtm->oldestXid;
 		}
@@ -2291,6 +2294,23 @@ mtm_get_snapshot(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(MtmTx.snapshot);
 }
 
+Datum
+mtm_get_csn(PG_FUNCTION_ARGS)
+{
+	TransactionId xid = PG_GETARG_INT32(0);
+	MtmTransState* ts;
+	csn_t csn = INVALID_CSN;
+
+	MtmLock(LW_SHARED);
+    ts = hash_search(MtmXid2State, &xid, HASH_FIND, NULL);
+    if (ts != NULL) { 
+		csn = ts->csn;
+	}
+	MtmUnlock();
+
+    return csn;
+}
+
 typedef struct
 {
 	int       nodeId;

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 from t order by u");
-        result r2 = txn2.exec("select v from t order by u");
+        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");
 		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,7 @@ void* reader(void* arg)
 					delta = diff;
 					if (delta < 0) lt++; else gt++;
 				} else if (delta != diff) { 
-					printf("Inconsistency found for record %d\n", i);
+					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()));
 				}
 			}
 		}