Merge branch 'master' of https://github.com/postgrespro/postgres_cluster

Author: knizhnik

contrib/mmts/Makefile

@@ -47,7 +47,7 @@ xcheck:
 	cd tests2 && docker network rm tests2_net || true
 	cd tests2 && docker network rm tests2_net || true
 	cd tests2 && blockade up
-	sleep 15 # wait for mmts init
+	sleep 20 # wait for mmts init
 	cd tests2 && python test_recovery.py || true
 	#cd tests2 && blockade destroy
 

contrib/mmts/multimaster.c

@@ -927,7 +927,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 			 * Send notification only if ABORT happens during transaction processing at replicas, 
 			 * do not send notification if ABORT is received from master 
 			 */
-			MTM_LOG2("%d: send ABORT notification abort transaction %d to coordinator %d", MyProcPid, x->gtid.xid, x->gtid.node);
+			MTM_LOG1("%d: send ABORT notification abort transaction %d to coordinator %d", MyProcPid, x->gtid.xid, x->gtid.node);
 			if (ts == NULL) { 
 				Assert(TransactionIdIsValid(x->xid));
 				ts = hash_search(MtmXid2State, &x->xid, HASH_ENTER, NULL);
@@ -1390,6 +1390,7 @@ bool MtmRefreshClusterStatus(bool nowait)
 				}
 			} else if (BIT_CHECK(disabled, ts->gtid.node-1)) { // coordinator of transaction is on disabled node
 				if (ts->status != TRANSACTION_STATUS_ABORTED) {
+					MTM_LOG1("1) Rollback active transaction %d:%d:%d", ts->gtid.node, ts->gtid.xid, ts->xid);
 					MtmAbortTransaction(ts);
 					FinishPreparedTransaction(ts->gid, false);
 				}
@@ -1460,6 +1461,7 @@ void MtmOnNodeDisconnect(int nodeId)
 					}
 				} else if (ts->gtid.node == nodeId) { //coordinator of transaction is on disabled node
 					if (ts->status != TRANSACTION_STATUS_ABORTED) {
+						MTM_LOG1("2) Rollback active transaction %d:%d", ts->gtid.node, ts->gtid.xid);
 						MtmAbortTransaction(ts);
 						FinishPreparedTransaction(ts->gid, false);
 					}

contrib/mmts/pglogical_apply.c

@@ -567,8 +567,9 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			MTM_LOG3("%d: PGLOGICAL_ABORT_PREPARED commit: gid=%s", MyProcPid, gid);
+			MTM_LOG1("%d: PGLOGICAL_ABORT_PREPARED commit: gid=%s", MyProcPid, gid);
 			if (MtmGetGlobalTransactionStatus(gid) != TRANSACTION_STATUS_ABORTED) { 
+				MTM_LOG1("%d: PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", MyProcPid, gid);
 				StartTransactionCommand();
 				MtmSetCurrentTransactionGID(gid);
 				FinishPreparedTransaction(gid, false);

contrib/mmts/pglogical_proto.c

@@ -160,7 +160,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 	}
     pq_sendbyte(out, 'C');		/* sending COMMIT */
 
-	MTM_LOG2("PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx", flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());
+	MTM_LOG1("PGLOGICAL_SEND commit: event=%d, gid=%s, commit_lsn=%lx, txn->end_lsn=%lx, xlog=%lx", flags, txn->gid, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr());
 
     /* send the flags field */
     pq_sendbyte(out, flags);

contrib/mmts/tests2/lib/bank_client.py

@@ -65,106 +65,68 @@ def print_error(self, arg, comment=''):
         if self.show_errors:
             print('Node', self.node_id, 'got error', arg, comment)
 
-    def check_total(self):
-        conn, cur = self.connect()
-        i = 0
+    def exec_tx(self, name, tx_block):
+        conn = psycopg2.connect(self.connstr)
+        cur = conn.cursor()
 
         while self.run.value:
-            i += 1
-
-            event_id = self.history.register_start('total')
+            event_id = self.history.register_start(name)
 
-            if not conn.closed:
+            if conn.closed:
+                self.history.register_finish(event_id, 'ReConnect')
                 try :
                     conn = psycopg2.connect(self.connstr)
                     cur = conn.cursor()
                 except :
-                    self.history.register_finish(event_id, 'CantConnect')
-                    next
-
-            amount = 1
-            from_uid = random.randrange(1, self.accounts + 1)
-            to_uid = random.randrange(1, self.accounts + 1)
+                    continue
+                else :
+                    continue 
 
             try:
-                cur.execute('select sum(amount) from bank_test')
-                res = cur.fetchone()
-                if res[0] != 0:
-                    print("Isolation error, total = %d" % (res[0],))
-                    raise BaseException
-            except BaseException:
-                raise BaseException
+                tx_block(conn, cur)
             except psycopg2.InterfaceError:
                 self.history.register_finish(event_id, 'InterfaceError')
+            except psycopg2.Error:
+                self.history.register_finish(event_id, 'PsycopgError')
             except :
+                print(sys.exc_info())
                 self.history.register_finish(event_id, 'OtherError')
             else :
-                self.history.register_finish(event_id, 'commit')
+                self.history.register_finish(event_id, 'Commit')
 
         cur.close()
         conn.close()
 
-    def transfer_money(self):
-        conn, cur = self.connect()
-
-        i = 0
+    def check_total(self):
 
-        while self.run.value:
-            i += 1
+        def tx(conn, cur):
+            cur.execute('select sum(amount) from bank_test')
+            res = cur.fetchone()
+            if res[0] != 0:
+                print("Isolation error, total = %d" % (res[0],))
+                raise BaseException
 
-            event_id = self.history.register_start('transfer')
+        self.exec_tx('total', tx)
 
-            if not conn.closed:
-                try :
-                    conn = psycopg2.connect(self.connstr)
-                    cur = conn.cursor()
-                except :
-                    self.history.register_finish(event_id, 'CantConnect')
-                    next
+    def transfer_money(self):
 
+        def tx(conn, cur):
             amount = 1
-            from_uid = random.randrange(1, self.accounts + 1)
-            to_uid = random.randrange(1, self.accounts + 1)
-
-            try:
-                cur.execute('''update bank_test
-                    set amount = amount - %s
-                    where uid = %s''',
-                    (amount, from_uid))
-                cur.execute('''update bank_test
-                    set amount = amount + %s
-                    where uid = %s''',
-                    (amount, to_uid))
-                conn.commit()
-
-            except psycopg2.InterfaceError:
-                self.history.register_finish(event_id, 'InterfaceError')
-            except :
-                self.history.register_finish(event_id, 'OtherError')
-            else :
-                self.history.register_finish(event_id, 'commit')
-
-        cur.close()
-        conn.close()
-
-    def connect(self, reconnect=False):
-        
-        while True:
-            try:
-                conn = psycopg2.connect(self.connstr)
-                cur = conn.cursor()
-                return conn, cur
-            except:
-                self.print_error(sys.exc_info(),'2')
-                if not reconnect:
-                    raise
-                if not self.run.value:
-                    raise
-
-    # def watchdog(self):
-    #    while self.run.value:
-    #        time.sleep(1)
-    #        print('watchdog: ', self.history.aggregate())
+            from_uid = random.randrange(1, self.accounts - 10)
+            to_uid = from_uid + 1 #random.randrange(1, self.accounts + 1)
+
+            conn.commit()
+            cur.execute('''update bank_test
+                set amount = amount - %s
+                where uid = %s''',
+                (amount, from_uid))
+            cur.execute('''update bank_test
+                set amount = amount + %s
+                where uid = %s''',
+                (amount, to_uid))
+            conn.commit()
+
+        self.exec_tx('transfer', tx)
 
     def start(self):
         self.transfer_process = Process(target=self.transfer_money, args=())
@@ -173,9 +135,6 @@ def start(self):
         self.total_process = Process(target=self.check_total, args=())
         self.total_process.start()
 
-        #self.total_process = Process(target=self.watchdog, args=())
-        #self.total_process.start()
-
         return
 
     def stop(self):

contrib/mmts/tests2/lib/event_history.py

@@ -1,6 +1,7 @@
 import time
 import datetime
 import uuid
+import copy
 from multiprocessing import Queue
 
 
@@ -12,11 +13,10 @@ def __init__(self):
         self.running_events = {}
         self.last_aggregation = datetime.datetime.now()
         self.agg_template = {
-            'commit': 0,
-            'rollback': 0,
             'max_latency': 0.0,
             'running': 0,
-            'running_latency': 0.0
+            'running_latency': 0.0,
+            'finish': {}
         }
 
     def register_start(self, name):
@@ -43,18 +43,21 @@ def load_queue(self):
                 self.running_events[event['event_id']] = event
             else:
                 # finish mark
-                if event['event_id'] in self.running_events:
-                    start_ev = self.running_events[event['event_id']]
-                    self.events.append({
-                        'name': start_ev['name'],
-                        'started_at': start_ev['time'],
-                        'finished_at': event['time'],
-                        'status': event['status']
-                    })
-                    self.running_events.pop(event['event_id'], None)
-                else:
+                if event['event_id'] not in self.running_events:
                     # found finish event without corresponding start
+                    print("ololololo!")
                     raise
+
+                start_ev = self.running_events[event['event_id']]
+                self.events.append({
+                    'name': start_ev['name'],
+                    'started_at': start_ev['time'],
+                    'finished_at': event['time'],
+                    'status': event['status']
+                })
+                self.running_events.pop(event['event_id'], None)
+
+        #print(self.events)
         return
 
     def aggregate(self):
@@ -63,32 +66,38 @@ def aggregate(self):
         agg = {}
         for ev in self.events:
             if ev['finished_at'] < self.last_aggregation:
+                #print("cont")
                 continue
 
             if ev['name'] not in agg:
-                agg[ev['name']] = self.agg_template.copy()
+                agg[ev['name']] = copy.deepcopy(self.agg_template)
+                #print('-=-=-', agg)
 
             named_agg = agg[ev['name']]
             latency = (ev['finished_at'] - ev['started_at']).total_seconds()
-            named_agg[ev['status']] += 1
+
+            if ev['status'] not in named_agg['finish']:
+                named_agg['finish'][ev['status']] = 1
+            else:
+                named_agg['finish'][ev['status']] += 1
+
             if named_agg['max_latency'] < latency:
                 named_agg['max_latency'] = latency
 
         for value in self.running_events.itervalues():
+
             if value['name'] not in agg:
-                agg[value['name']] = self.agg_template.copy()
+                agg[value['name']] = copy.deepcopy(self.agg_template)
 
             named_agg = agg[value['name']]
             latency = (datetime.datetime.now() - value['time']).total_seconds()
-            if 'started' in named_agg:
-                named_agg['running'] += 1
-                if latency > named_agg['running_latency']:
-                    named_agg['running_latency'] = latency
-            else:
-                named_agg['running'] = 1
+
+            named_agg['running'] += 1
+            if named_agg['running_latency'] < latency:
                 named_agg['running_latency'] = latency
 
         self.last_aggregation = datetime.datetime.now()
+        #print("aggregeted!")
         return agg
 
     def aggregate_by(self, period):

contrib/mmts/tests2/test_recovery.py

@@ -3,7 +3,6 @@
 import subprocess
 from lib.bank_client import *
 
-
 class RecoveryTest(unittest.TestCase):
     def setUp(self):
         #subprocess.check_call(['blockade','up'])
@@ -24,41 +23,36 @@ def test_0_normal_operation(self):
         print('### normalOpsTest ###')
         print('Waiting 5s to check operability')
         time.sleep(5)
-        
+
         for client in self.clients:
             agg = client.history.aggregate()
             print(agg)
-            self.assertTrue(agg['transfer']['commit'] > 0)
+            self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
 
     def test_1_node_disconnect(self):
         print('### disconnectTest ###')
 
         subprocess.check_call(['blockade','partition','node3'])
         print('Node3 disconnected')
 
-        print('Waiting 20s to discover failure')
+        print('Waiting 15s to discover failure')
 
-        while True:
+        for i in range(5):
             time.sleep(3)
             for client in self.clients:
                 agg = client.history.aggregate()
                 print(agg)
+            print(" ")
+
+        subprocess.check_call(['blockade','join'])
 
-#        print('Waiting 3s to check operability')
-#        time.sleep(3)
-#        for client in self.clients:
-#            agg = client.history.aggregate()
-#            print(agg)
-#
-#        subprocess.check_call(['blockade','join'])
-#        print('Node3 connected back')
-#
-#        print('Waiting 12s for catch-up')
-#        time.sleep(12)
-#
-#        for client in self.clients:
-#            agg = client.history.aggregate()
-#            print(agg)
+        print('Waiting 15s to join node')
+        for i in range(1000):
+            time.sleep(3)
+            for client in self.clients:
+                agg = client.history.aggregate()
+                print(agg)
+            print(" ")
 
 
 if __name__ == '__main__':