Use repeatable read isolation level

Author: knizhnik

contrib/mmts/tests2/docker-entrypoint.sh

@@ -73,7 +73,9 @@ if [ "$1" = 'postgres' ]; then
 			max_replication_slots = 10
 			max_wal_senders = 10
 			shared_preload_libraries = 'raftable,multimaster'
-            default_transaction_isolation = 'repeatable read'
+            		default_transaction_isolation = 'repeatable read'
+			log_checkpoints = on
+			log_autovacuum_min_duration = 0
 
 			raftable.id = $NODE_ID
 			raftable.peers = '$RAFT_PEERS'

contrib/mmts/tests2/lib/bank_client.py

@@ -5,6 +5,8 @@
 import time
 import sys
 from event_history import *
+import select
+import signal
 
 class ClientCollection(object):
     def __init__(self, connstrs):
@@ -38,8 +40,6 @@ def print_agg(self):
 
         columns = ['running', 'running_latency', 'max_latency', 'finish']
 
-        #rows = [ k+str(i+1) for k in agg.keys() for i, agg in enumerate(aggs)]
-
         print("\t\t", end="")
         for col in columns:
             print(col, end="\t")
@@ -61,17 +61,29 @@ def print_agg(self):
 
         print("")
 
+    def set_acc_to_tx(self, max_acc):
+        for client in self._clients:
+            client.set_acc_to_tx(max_acc)
+
 
 class BankClient(object):
 
-    def __init__(self, connstr, node_id):
+    def __init__(self, connstr, node_id, accounts = 10000):
         self.connstr = connstr
         self.node_id = node_id
         self.run = Value('b', True)
         self._history = EventHistory()
-        self.accounts = 10000
+        self.accounts = accounts
+        self.accounts_to_tx = accounts
         self.show_errors = True
 
+        #x = self
+        #def on_sigint(sig, frame):
+        #    x.stop()
+        #
+        #signal.signal(signal.SIGINT, on_sigint)
+
+
     def initialize(self):
         conn = psycopg2.connect(self.connstr)
         cur = conn.cursor()
@@ -88,6 +100,22 @@ def initialize(self):
         cur.close()
         conn.close()
 
+    def aconn(self):
+        return psycopg2.connect(self.connstr, async=1)
+
+    @classmethod
+    def wait(cls, conn):
+        while 1:
+            state = conn.poll()
+            if state == psycopg2.extensions.POLL_OK:
+                break
+            elif state == psycopg2.extensions.POLL_WRITE:
+                select.select([], [conn.fileno()], [])
+            elif state == psycopg2.extensions.POLL_READ:
+                select.select([conn.fileno()], [], [])
+            else:
+                raise psycopg2.OperationalError("poll() returned %s" % state)
+
     @property
     def history(self):
         return self._history
@@ -105,25 +133,16 @@ def exec_tx(self, name, tx_block):
 
             if conn.closed:
                 self.history.register_finish(event_id, 'ReConnect')
-                try :
-                    conn = psycopg2.connect(self.connstr)
-                    cur = conn.cursor()
-                except :
-                    continue
-                else :
-                    continue 
+                conn = psycopg2.connect(self.connstr)
+                cur = conn.cursor()
 
             try:
-                tx_block(conn, cur)
+                tx_block(conn, cur)                
+                self.history.register_finish(event_id, 'Commit')
             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')
 
         cur.close()
         conn.close()
@@ -135,17 +154,20 @@ def tx(conn, cur):
             res = cur.fetchone()
             conn.commit()
             if res[0] != 0:
-                print("Isolation error, total = %d" % (res[0],))
+                print("Isolation error, total = %d, node = %d" % (res[0],self.node_id))
                 raise BaseException
 
         self.exec_tx('total', tx)
 
+    def set_acc_to_tx(self, max_acc):
+        self.accounts_to_tx = max_acc
+
     def transfer_money(self):
 
         def tx(conn, cur):
             amount = 1
-            from_uid = random.randrange(1, self.accounts - 10)
-            to_uid = from_uid + 1 #random.randrange(1, self.accounts + 1)
+            from_uid = random.randrange(1, self.accounts_to_tx - 1)
+            to_uid = random.randrange(1, self.accounts_to_tx - 1)
 
             conn.commit()
             cur.execute('''update bank_test
@@ -161,7 +183,10 @@ def tx(conn, cur):
         self.exec_tx('transfer', tx)
 
     def start(self):
-        self.transfer_process = Process(target=self.transfer_money, args=())
+        print('Starting client');
+        self.run.value = True
+
+        self.transfer_process = Process(target=self.transfer_money, name="txor", args=())
         self.transfer_process.start()
 
         self.total_process = Process(target=self.check_total, args=())
@@ -170,7 +195,7 @@ def start(self):
         return
 
     def stop(self):
-        print('Stopping!');
+        print('Stopping client');
         self.run.value = False
         self.total_process.terminate()
         self.transfer_process.terminate()

contrib/mmts/tests2/test_recovery.py

@@ -4,43 +4,80 @@
 from lib.bank_client import *
 
 class RecoveryTest(unittest.TestCase):
-    def setUp(self):
+    @classmethod
+    def setUpClass(self):
         #subprocess.check_call(['blockade','up'])
         self.clients = ClientCollection([
             "dbname=postgres host=127.0.0.1 user=postgres",
             "dbname=postgres host=127.0.0.1 user=postgres port=5433",
             "dbname=postgres host=127.0.0.1 user=postgres port=5434"
         ])
-        self.clients.start()
 
-    def tearDown(self):
+    @classmethod
+    def tearDownClass(self):
         print('tearDown')
+        #subprocess.check_call(['blockade','join'])
+
+        # in case of error
         self.clients.stop()
-        self.clients[0].cleanup()
-        subprocess.check_call(['blockade','join'])
+        #self.clients[0].cleanup()
+
+
+    def test_0_normal_operation(self):
+        print('### normalOpsTest ###')
+
+        self.clients.set_acc_to_tx(10000)
+        self.clients.start()
+
+        for i in range(5):
+            time.sleep(3)
+            for client in self.clients:
+                agg = client.history.aggregate()
+                print(agg)
+                self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
+            print("\n")
+
+        self.clients.stop()
+
+    def test_1_distributed_deadlock(self):
+        print('### DDD test ###')
+
+        self.clients.set_acc_to_tx(10)
+        self.clients.start()
+
+        for i in range(5):
+            time.sleep(3)
+            for client in self.clients:
+                agg = client.history.aggregate()
+                print(agg)
+                self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
+            print("\n")
 
-#    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']['finish']['Commit'] > 0)
-
-    def test_1_node_disconnect(self):
+        self.clients.stop()
+
+    def test_2_node_disconnect(self):
         print('### disconnectTest ###')
 
+        self.clients.set_acc_to_tx(10000)
+        self.clients.start()
+
         subprocess.check_call(['blockade','partition','node3'])
         print('Node3 disconnected')
 
-        for i in range(1000):
+        # give cluster some time to discover problem
+        time.sleep(3)
+
+        for i in range(5):
             time.sleep(3)
-            self.clients.print_agg()
+            for client in self.clients:
+                agg = client.history.aggregate()
+                print(agg)
+                self.assertTrue(agg['transfer']['finish']['Commit'] > 0)
+            print("\n")
+
+        subprocess.check_call(['blockade','join'])
+        self.clients.stop()
 
 if __name__ == '__main__':
     unittest.main()
 
-
-