retry on transaction deadlock

1 files changed, 148 insertions, 82 deletions

diff --git a/rtss14/database.py b/rtss14/database.py
index 1939056..c2597f6 100755
--- a/rtss14/database.py
+++ b/rtss14/database.py
@@ -172,6 +172,7 @@ def dp_to_db_vals(dp):
         vals[key] = to_db_val(value)
     return vals
 
+
 def begin_sync(conn, c):
     global timeout
     start = time.time()
@@ -216,6 +217,7 @@ def connect_db(db_name):
         init_dp_col_names(conn)
     return conn
 
+
 def clear_tables(db_name):
     # tables to clear
     tables = ['sched_results', 'scaled_sched_results', 'dp_ptested', 'dp_pending']
@@ -230,43 +232,6 @@ def clear_tables(db_name):
         ################
         c.close()
 
-def lookup_dp_id(c, dp):
-    global dp_col_names
-    global dp_col_type_strs
-    seeking = dp_to_db(dp)
-    seeking.pop('ts_util', None)
-
-    col_names = [n for n in dp_col_names if n != 'ts_util']
-
-    # atomically add new dp to dp_ptested. fails if dp already in table
-    c.execute('INSERT INTO dp_ptested(id,%s) '
-              'SELECT NULL,%s '
-              'FROM dual '
-              'WHERE NOT EXISTS ( '
-                  'SELECT 1 '
-                  'FROM dp_ptested '
-                  'WHERE %s )' % ( ','.join(col_names),
-                                   ','.join(['%s']*(len(col_names))),
-                                   ' AND '.join(map(lambda x: '%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys())) ),
-               tuple([seeking[n] for n in col_names] + seeking.values()) )
-
-    if c.rowcount != 0:
-        # new id
-        c.execute('SELECT LAST_INSERT_ID()')
-        row = c.fetchone()
-        assert row is not None
-        dp_id = row.values()[0]
-    else:
-        # id already exists. look it up.
-        c.execute('SELECT id '
-                  'FROM dp_ptested '
-                  'WHERE %s' % (' AND '.join(map(lambda x: '%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys())) ),
-                  tuple(seeking.values()) )
-        row = c.fetchone()
-        assert row is not None
-        dp_id = row['id']
-    return dp_id
-
 
 def __repair_design_points(conn):
     global dp_col_names
@@ -288,6 +253,7 @@ def repair_design_points(db_name):
         nrepaired = __repair_design_points(conn)
     return nrepaired
 
+
 repaired = False
 def store_design_points(db_name, dps, clean):
     global dp_col_names
@@ -335,6 +301,7 @@ def store_design_points(db_name, dps, clean):
         c.close()
     return nstored
 
+
 def num_pending_design_points(db_name):
     conn = connect_db(db_name)
     with conn:
@@ -344,27 +311,49 @@ def num_pending_design_points(db_name):
         c.close()
     return npending
 
+
 def __get_design_points(conn, ndp):
     global dp_col_names
     fetched = []
     c = conn.cursor(db.cursors.DictCursor)
     ####################
-    try:
-        begin_sync(conn, c)
-        c.execute('SELECT id,%s FROM dp_pending WHERE taken=0 LIMIT %%s FOR UPDATE' % ','.join(dp_col_names), (ndp,))
-        rows = c.fetchall()
-        nfetched = c.rowcount
-        if nfetched > 0:
-            dp_ids = [r['id'] for r in rows]
-            c.execute('UPDATE dp_pending SET taken=1 WHERE id IN (%s)' % ','.join(['%s']*len(dp_ids)), tuple(dp_ids))
-        end_sync(conn)
-        fetched = [db_to_dp(r) for r in rows]
-    except db.OperationalError, e:
-        print e
-        print 'Last Query: %s' % c._last_executed
-        rollback(conn)
-        c.close()
-        raise e
+    done = False
+    while not done:
+
+        try:
+            begin_sync(conn, c)
+            c.execute('SELECT id,%s FROM dp_pending WHERE taken=0 LIMIT %%s FOR UPDATE' % ','.join(dp_col_names), (ndp,))
+            rows = c.fetchall()
+            nfetched = c.rowcount
+            if nfetched > 0:
+                dp_ids = [r['id'] for r in rows]
+                c.execute('UPDATE dp_pending SET taken=1 WHERE id IN (%s)' % ','.join(['%s']*len(dp_ids)), tuple(dp_ids))
+            end_sync(conn)
+            fetched = [db_to_dp(r) for r in rows]
+            done = True
+
+        except db.OperationalError, e:
+            errcode = e[0]
+            if errcode == 1213:
+                # deadlock - retry
+                print '(__get_design_points) Error is transaction deadlock. Will retry.'
+                print '(__get_design_points) Last query: %s' % c._last_executed
+                rollback(conn)
+            else:
+                print e
+                print '(__get_design_points) Last query: %s' % c._last_executed
+                rollback(conn)
+                c.close()
+                # rethrow
+                raise e
+
+        except BaseException, e:
+            print e
+            print '(__get_design_points) Last query: %s' % c._last_executed
+            rollback(conn)
+            c.close()
+            raise e
+
     c.close()
     return fetched
 
@@ -375,6 +364,60 @@ def get_design_points(db_name, ndp = 1):
         fetched = __get_design_points(conn, ndp)
     return fetched
 
+
+def lookup_dp_id(c, dp):
+    global dp_col_names
+    global dp_col_type_strs
+    seeking = dp_to_db(dp)
+    seeking.pop('ts_util', None)
+
+    col_names = [n for n in dp_col_names if n != 'ts_util']
+
+    dp_id = None
+    while dp_id is None:
+        try:
+            # atomically add new dp to dp_ptested. fails if dp already in table
+            c.execute('INSERT INTO dp_ptested(id,%s) '
+                      'SELECT NULL,%s '
+                      'FROM dual '
+                      'WHERE NOT EXISTS ( '
+                          'SELECT 1 '
+                          'FROM dp_ptested '
+                          'WHERE %s )' % ( ','.join(col_names),
+                                           ','.join(['%s']*(len(col_names))),
+                                           ' AND '.join(map(lambda x: '%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys())) ),
+                       tuple([seeking[n] for n in col_names] + seeking.values()) )
+
+            if c.rowcount != 0:
+                # new id
+                c.execute('SELECT LAST_INSERT_ID()')
+                row = c.fetchone()
+                assert row is not None
+                dp_id = row.values()[0]
+            else:
+                # id already exists. look it up.
+                c.execute('SELECT id '
+                          'FROM dp_ptested '
+                          'WHERE %s' % (' AND '.join(map(lambda x: '%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys())) ),
+                          tuple(seeking.values()) )
+                row = c.fetchone()
+                assert row is not None
+                dp_id = row['id']
+
+        except db.OperationalError, e:
+            errcode = e[0]
+            if errcode == 1213:
+                # deadlock - retry
+                print '(lookup_dp_id) Error is transaction deadlock. Will retry.'
+                print '(lookup_dp_id) Last query: %s' % c._last_executed
+            else:
+                print e
+                print '(lookup_dp_id) Last query: %s' % c._last_executed
+                raise e
+
+    return dp_id
+
+
 def __store_eff_sched_results(c, dp_id, stats):
     for factor,eff_curve in stats.iteritems():
         for eff_ts_util,sched in eff_curve.iteritems():
@@ -406,53 +449,76 @@ def __store_eff_sched_results(c, dp_id, stats):
                           'VALUES(%s,%s,%s,%s,%s,%s,%s,%s)',
                           (dp_id, eff_ts_util, factor, sched.ntested, sched.nsched, sched.avg_sched, sched.avg_tard, sched.avg_bandwidth))
 
+def __store_sched_result(c, dp_id, ts_util, stats):
+    c.execute('INSERT INTO sched_results VALUES(%s,%s,%s,%s,%s,%s,%s)',
+              (dp_id, ts_util,
+               stats.avg_sched,
+               stats.ntested,
+               stats.nsched,
+               stats.avg_tard,
+               stats.avg_bandwidth) )
+
+def store_sched_result(c, dp_id, result):
+    done = False
+    while not done:
+        try:
+            begin_sync(conn, c)
+            __store_sched_result(c, dp_id, result.dp.sys_util, result.sched_stats)
+            if result.eff_sched_stats is not None:
+                __store_eff_sched_results(c, dp_id, result.eff_sched_stats)
+            end_sync(conn)
+            done = True
+
+        except db.OperationalError, e:
+            errcode = e[0]
+            if errcode == 1213:
+                # deadlock - retry
+                print '(store_sched_result) Error is transaction deadlock. Will retry.'
+                print '(store_sched_result) Last query: %s' % c._last_executed
+                rollback(conn)
+            else:
+                print e
+                print '(store_sched_result) Last query: %s' % c._last_executed
+                rollback(conn)
+                raise e
+
+        except BaseException, e:
+            print e
+            print '(store_sched_result) Last query: %s' % c._last_executed
+            rollback(conn)
+            raise e
+
 def store_sched_results(db_name, data, ndp = 0):
     global dp_col_names
-    global dp_col_type_strs
-    col_names = None
     fetched = []
 
     conn = connect_db(db_name)
-
-    if col_names is None:
-        col_names = [n for n in dp_col_names if n != 'id']
-
     with conn:
         # data is a list of storage(), with fields 'dp, 'sched_stats', and 'eff_sched_stats'.
         # 'eff_sched_stats' is a dictionary keyed on scale-factor. stats in 'eff_sched_stats'
         # must be merged with prior results in the 'scaled_sched_results' table
         c = conn.cursor(db.cursors.DictCursor)
-        try:
-            bundle = [(lookup_dp_id(c, d.dp), d) for d in data]
 
+        col_names = [n for n in dp_col_names if n != 'id']
+
+        try:
             # get IDs for all design points.
-            # sort to ensure that we update scaled design points
-            # in a consistent order (no deadlocks!).
-            bundle = sorted(bundle, key=lambda x: x[0])
+            # sort to reduce (eliminate?) chance of deadlock.
+            id_and_data = sorted([(lookup_dp_id(c, d.dp), d) for d in data], key=lambda x: x[0])
+
+            for dp_id, result in id_and_data:
+                store_sched_result(c, dp_id, result)
 
-            # insert the normal schedulability data
-            begin_sync(conn, c)
-            # save results
-            c.executemany('INSERT INTO sched_results VALUES(%s,%s,%s,%s,%s,%s,%s)',
-                [(dp_id, d.dp.sys_util, d.sched_stats.avg_sched, d.sched_stats.ntested, d.sched_stats.nsched, d.sched_stats.avg_tard, d.sched_stats.avg_bandwidth)
-                    for dp_id, d in bundle])
-            # insert the effective utilization-based schedulability data
-            for dp_id, d in bundle:
-                if d.eff_sched_stats is not None:
-                    __store_eff_sched_results(c, dp_id, d.eff_sched_stats)
-            end_sync(conn)
         except db.OperationalError, e:
-            print e
-            print 'Last Query: %s' % c._last_executed
-            rollback(conn)
+            print '(store_sched_results) Caught database exception.'
             c.close()
-            # TODO: mark unsaved design points as not taken in dp_pending
-            # rethrow
             raise e
+
         except BaseException, e:
-            print 'Caught unknown exception. Rolling back transaction before exiting.'
-            rollback(conn)
+            print '(store_sched_results) Caught unknown exception.'
+            c.close()
             raise e
+
         c.close()
         if ndp > 0:
             fetched = __get_design_points(conn, ndp)