Add active snapshot

knizhnik · knizhnik · commit 8ed9197ec83d · 2015-09-25T16:30:39.000+03:00
diff --git a/contrib/pg_xtm/dtmd/src/main.c b/contrib/pg_xtm/dtmd/src/main.c
@@ -227,17 +227,17 @@ static void gen_snapshot(Transaction *t) {
 
 static xid_t get_global_xmin() {
 	int i, j;
-	xid_t xmin = MAX_XID;
+	xid_t xmin = INVALID_XID;
 	Transaction *t;
 	for (i = 0; i < transactions_count; i++) {
 		t = transactions + i;
-        j = t->snapshots_count > MAX_SNAPSHOTS_PER_TRANS ? MAX_SNAPSHOTS_PER_TRANS : t->snapshots_count; 
-        while (--j >= 0) { 
-            Snapshot* s = transaction_snapshot(t, j);
-            if (s->xmin < xmin) {
-                xmin = s->xmin;
-            }
-            // minor TODO: Use 'times_sent' to generate a bit greater xmin?
+		j = t->snapshots_count > MAX_SNAPSHOTS_PER_TRANS ? MAX_SNAPSHOTS_PER_TRANS : t->snapshots_count; 
+		while (--j >= 0) { 
+			Snapshot* s = transaction_snapshot(t, j);
+			if ((xmin == INVALID_XID) || (s->xmin < xmin)) {
+				xmin = s->xmin;
+			}
+			// minor TODO: Use 'times_sent' to generate a bit greater xmin?
 		}
 	}
 	return xmin;
diff --git a/contrib/pg_xtm/tests/transfers.go b/contrib/pg_xtm/tests/transfers.go
@@ -15,6 +15,8 @@ const (
     N_ACCOUNTS = TRANSFER_CONNECTIONS//100000
     //ISOLATION_LEVEL = "repeatable read"
     ISOLATION_LEVEL = "read committed"
+    GLOBAL_UPDATES = true
+    LOCAL_UPDATES = false
 )
 
 
@@ -115,13 +117,15 @@ func transfer(id int, cCommits chan int, cAborts chan int, wg *sync.WaitGroup) {
     var nCommits = 0
     var myCommits = 0
 
-    conn1, err := pgx.Connect(cfg1)
+    var conn [2]*pgx.Conn
+
+    conn[0], err = pgx.Connect(cfg1)
     checkErr(err)
-    defer conn1.Close()
+    defer conn[0].Close()
 
-    conn2, err := pgx.Connect(cfg2)
+    conn[1], err = pgx.Connect(cfg2)
     checkErr(err)
-    defer conn2.Close()
+    defer conn[1].Close()
 
     start := time.Now()
     for myCommits < N_ITERATIONS {
@@ -130,23 +134,57 @@ func transfer(id int, cCommits chan int, cAborts chan int, wg *sync.WaitGroup) {
         account1 := rand.Intn(N_ACCOUNTS)
         account2 := rand.Intn(N_ACCOUNTS)
 
-        xid = execQuery(conn1, "select dtm_begin_transaction(2)")
-        exec(conn2, "select dtm_join_transaction($1)", xid)
+        if (account1 >= account2) {
+            continue
+        }
+
+        src := conn[rand.Intn(2)]
+        dst := conn[rand.Intn(2)]
 
-        // start transaction
-        exec(conn1, "begin transaction isolation level " + ISOLATION_LEVEL)
-        exec(conn2, "begin transaction isolation level " + ISOLATION_LEVEL)
+        if src == dst {
+            // local update
+            if !LOCAL_UPDATES {
+                // which we do not want
+                continue
+            }
 
-        ok1 := execUpdate(conn1, "update t set v = v + $1 where u=$2", amount, account1)
-        ok2 := execUpdate(conn2, "update t set v = v - $1 where u=$2", amount, account2)
-        if !ok1 || !ok2 {
-            exec(conn1, "rollback")
-            exec(conn2, "rollback")
-            nAborts += 1
+            exec(src, "begin transaction isolation level " + ISOLATION_LEVEL)
+            ok1 := execUpdate(src, "update t set v = v - $1 where u=$2", amount, account1)
+            ok2 := execUpdate(src, "update t set v = v + $1 where u=$2", amount, account2)
+            if !ok1 || !ok2 {
+                exec(src, "rollback")
+                nAborts += 1
+            } else {
+                exec(src, "commit")
+                nCommits += 1
+                myCommits += 1
+            }
         } else {
-            commit(conn1, conn2)
-            nCommits += 1
-            myCommits += 1
+            // global update
+            if !GLOBAL_UPDATES {
+                // which we do not want
+                continue
+            }
+
+            xid = execQuery(src, "select dtm_begin_transaction(2)")
+            exec(dst, "select dtm_join_transaction($1)", xid)
+
+            // start transaction
+            exec(src, "begin transaction isolation level " + ISOLATION_LEVEL)
+            exec(dst, "begin transaction isolation level " + ISOLATION_LEVEL)
+
+            ok1 := execUpdate(src, "update t set v = v - $1 where u=$2", amount, account1)
+            ok2 := execUpdate(dst, "update t set v = v + $1 where u=$2", amount, account2)
+
+            if !ok1 || !ok2 {
+                exec(src, "rollback")
+                exec(dst, "rollback")
+                nAborts += 1
+            } else {
+                commit(src, dst)
+                nCommits += 1
+                myCommits += 1
+            }
         }
 
         if time.Since(start).Seconds() > 1 {
@@ -241,6 +279,9 @@ func execUpdate(conn *pgx.Conn, stmt string, arguments ...interface{}) bool {
     var err error
     // fmt.Println(stmt)
     _, err = conn.Exec(stmt, arguments... )
+    if err != nil {
+        fmt.Println(err)
+    }
     return err == nil
 }
 
