Support subtransactions

knizhnik · knizhnik · commit fc142bad326a · 2015-11-26T23:36:30.000+03:00
diff --git a/pg_dtm.c b/pg_dtm.c
@@ -21,6 +21,7 @@
 #include "access/xact.h"
 #include "access/xtm.h"
 #include "access/transam.h"
+#include "access/subtrans.h"
 #include "access/xlog.h"
 #include "access/clog.h"
 #include "access/twophase.h"
@@ -455,7 +456,10 @@ bool DtmXidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 #endif
     while (true)
     {
-        DtmTransStatus* ts = (DtmTransStatus*)hash_search(xid2status, &xid, HASH_FIND, NULL);
+        DtmTransStatus* ts;
+        TransactionId subxid = xid;
+        while ((ts = (DtmTransStatus*)hash_search(xid2status, &subxid, HASH_FIND, NULL)) == NULL
+               && TransactionIdIsValid(subxid = SubTransGetParent(subxid)));        
         if (ts != NULL)
         {
             if (ts->cid > dtm_tx.snapshot) { 
diff --git a/tests/dtmbench.cpp b/tests/dtmbench.cpp
@@ -68,7 +68,7 @@ struct config
         nWriters = 10;
         nIterations = 1000;
         nAccounts = 100000;  
-        startId = 1;
+        startId = 0;
         diapason = 100000;
     }
 };
@@ -177,7 +177,10 @@ void* writer(void* arg)
         csn_t snapshot = execQuery(srcTx, "select dtm_extend('%s')", gtid);
         snapshot = execQuery(dstTx, "select dtm_access(%ld, '%s')", snapshot, gtid);
 
-        exec(srcTx, "update t set v = v - 1 where u=%d", srcAcc);
+        exec(srcTx, "savepoint c1");        
+        exec(dstTx, "savepoint c2");        
+
+        exec(srcTx, "update t set v = v - 1 where u=%d", srcAcc);        
         exec(dstTx, "update t set v = v + 1 where u=%d", dstAcc);
 
         exec(srcTx, "prepare transaction '%s'", gtid);
@@ -260,7 +263,7 @@ int main (int argc, char* argv[])
         return 1;
     }
 
-    if (cfg.startId + cfg.diapason - 1 > cfg.nAccounts) {
+    if (cfg.startId + cfg.diapason > cfg.nAccounts) {
         printf("startId + diapason should be less that nAccounts. Exiting.\n");
         return 1;
     }
@@ -305,7 +308,7 @@ int main (int argc, char* argv[])
     printf(
         "{\"update_tps\":%f, \"read_tps\":%f,"
         " \"readers\":%d, \"writers\":%d,"
-        " \"accounts\":%d, \"iterations\":%d, \"hosts\":%d}\n",
+        " \"accounts\":%d, \"iterations\":%d, \"hosts\":%ld}\n",
         (double)(nWrites*USEC)/elapsed,
         (double)(nReads*USEC)/elapsed,
         cfg.nReaders,
