Support read-committed isolation level

Author: knizhnik

multimaster.c

@@ -504,10 +504,22 @@ csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, nodemask_
     return snapshot;
 }
 
+void MtmSetSnapshot(csn_t globalSnapshot)
+{
+	MtmLock(LW_EXCLUSIVE);
+	MtmSyncClock(globalSnapshot);	
+	MtmTx.snapshot = globalSnapshot;	
+	MtmUnlock();
+}
 
+		
 Snapshot MtmGetSnapshot(Snapshot snapshot)
 {
 	snapshot = PgGetSnapshotData(snapshot);
+	if (XactIsoLevel == XACT_READ_COMMITTED && MtmTx.snapshot != INVALID_CSN && TransactionIdIsValid(GetCurrentTransactionIdIfAny())) { 
+		MtmTx.snapshot = MtmGetCurrentTime();
+		LogLogicalMessage("S", (char*)&MtmTx.snapshot, sizeof(MtmTx.snapshot), true);
+	}
 	RecentGlobalDataXmin = RecentGlobalXmin = Mtm->oldestXid;
 	return snapshot;
 }
@@ -540,7 +552,7 @@ bool MtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 
     Assert(xid != InvalidTransactionId);
 
-	if (!MtmUseDtm) { 
+	if (!MtmUseDtm || TransactionIdPrecedes(xid, Mtm->oldestXid)) { 
 		return PgXidInMVCCSnapshot(xid, snapshot);
 	}
 	MtmLock(LW_SHARED);
@@ -681,6 +693,10 @@ MtmAdjustOldestXid(TransactionId xid)
 				hash_search(MtmGid2State, &prev->gid, HASH_REMOVE, NULL);
 			}
 		}
+		if (ts != NULL) { 
+			MTM_LOG2("Adjust(%lld) stop at snashot %lld, xid %lld, pinned=%d, oldestSnaphsot=%lld\n",
+					 (long64)xid, ts->csn, (long64)ts->xid, ts->isPinned, oldestSnapshot);
+		}					 
 	} 
 
 	if (MtmUseDtm && !MtmVolksWagenMode) 
@@ -2827,11 +2843,13 @@ static bool ConfigIsSane(void)
 {
 	bool ok = true;
 
+#if 0
 	if (DefaultXactIsoLevel != XACT_REPEATABLE_READ)
 	{
 		MTM_ELOG(WARNING, "multimaster requires default_transaction_isolation = 'repeatable read'");
 		ok = false;
 	}
+#endif
 
 	if (MtmMaxNodes < 1)
 	{
@@ -5123,11 +5141,11 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		standard_ProcessUtility(parsetree, queryString, context,
 								params, dest, completionTag);
 	}
-	
+#if 0
     if (!MtmVolksWagenMode && MtmTx.isDistributed && XactIsoLevel != XACT_REPEATABLE_READ) { 
 		MTM_ELOG(ERROR, "Isolation level %s is not supported by multimaster", isoLevelStr[XactIsoLevel]);
 	}
-
+#endif
 	if (MyXactAccessedTempRel)
 	{
 		MTM_LOG1("Xact accessed temp table, stopping replication");

multimaster.h

@@ -424,5 +424,6 @@ extern bool MtmTransIsActive(void);
 extern MtmTransState* MtmGetActiveTransaction(MtmL2List* list);
 extern void MtmReleaseLocks(void);
 extern void MtmInitMessage(MtmArbiterMessage* msg, MtmMessageCode code);
+extern void MtmSetSnapshot(csn_t snapshot);
 
 #endif

pglogical_apply.c

@@ -483,8 +483,15 @@ process_remote_message(StringInfo s)
 			standalone = true;
 			break;
 		}
-
-	}
+	    case 'S':
+		{
+  		    Assert(messageSize == sizeof(csn_t));
+		    MtmSetSnapshot(*(csn_t*)messageBody);
+			break;
+		}
+	    default:
+		    Assert(false);
+ 	}
 	return standalone;
 }
 

pglogical_proto.c

@@ -180,6 +180,11 @@ pglogical_write_message(StringInfo out,
 			MTM_LOG1("Send deadlock message to node %d", MtmReplicationNodeId);
 		}
 		break;
+	  case 'S':
+		if (MtmIsFilteredTxn) {
+			return;
+		}
+		break;
 	  case 'D':
 		if (MtmIsFilteredTxn) {
 			MTM_LOG2("%d: pglogical_write_message filtered", MyProcPid);

tests/dtmbench.cpp

@@ -69,6 +69,7 @@ struct config
     int updatePercent;
     vector<string> connections;
 	bool scatter;
+	bool avoidDeadlocks;
 
     config() {
         nReaders = 1;
@@ -77,6 +78,7 @@ struct config
         nAccounts = 100000;
         updatePercent = 100;
 		scatter = false;
+		avoidDeadlocks = false;
     }
 };
 
@@ -157,6 +159,12 @@ void* writer(void* arg)
 		if (cfg.scatter) { 
 			srcAcc = srcAcc/cfg.nWriters*cfg.nWriters + t.id;
 			dstAcc = dstAcc/cfg.nWriters*cfg.nWriters + t.id;
+		} else if (cfg.avoidDeadlocks) { 
+			if (dstAcc < srcAcc) { 
+				int tmp = srcAcc;
+				srcAcc = dstAcc;
+				dstAcc = tmp;
+			}
 		}
         try {            
             if (random() % 100 < cfg.updatePercent) { 
@@ -240,6 +248,9 @@ int main (int argc, char* argv[])
             case 'i':
                 initialize = true;
                 continue;
+            case 'd':
+			    cfg.avoidDeadlocks = true;
+                continue;
             }
         }
         printf("Options:\n"
@@ -249,6 +260,8 @@ int main (int argc, char* argv[])
                "\t-n N\tnumber of iterations (1000)\n"
                "\t-p N\tupdate percent (100)\n"
                "\t-c STR\tdatabase connection string\n"
+               "\t-s\tscattern avoid deadlocks\n"
+               "\t-d\tavoid deadlocks\n"
                "\t-i\tinitialize database\n");
         return 1;
     }