save results in configable chunk size transactions

1 files changed, 69 insertions, 26 deletions

diff --git a/rtss14/database.py b/rtss14/database.py
index 8290128..ec540ee 100755
--- a/rtss14/database.py
+++ b/rtss14/database.py
@@ -21,6 +21,10 @@ dp_col_names = None  # list of keys to extract from sched results to write to db
 dp_col_type_strs = None
 ######################
 
+def chunker(l, n):
+    for i in xrange(0, len(l), n):
+        yield l[i:i+n]
+
 def get_job_id():
     job_idx_str = os.environ.get('LSB_JOBINDEX')
     job_pid_str = os.environ.get('LSB_JOBID')
@@ -449,23 +453,63 @@ 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(conn, c, dp_id, result):
+#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(conn, 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_result_chunk(conn, c, chunk):
     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)
+            # insert normal sched results
+            c.executemany('INSERT INTO sched_results VALUES(%s,%s,%s,%s,%s,%s,%s)',
+                [(dp_id,
+                  result.dp.sys_util,
+                  result.sched_stats.avg_sched,
+                  result.sched_stats.ntested,
+                  result.sched_stats.nsched,
+                  result.sched_stats.avg_tard,
+                  result.sched_stats.avg_bandwidth) for dp_id, result in chunk])
+            # insert scaled sched results
+            for dp_id, result in chunk:
+                if result.eff_sched_stats is not None:
+                    __store_eff_sched_results(c, dp_id, result.eff_sched_stats)
             end_sync(conn)
             done = True
 
@@ -473,23 +517,22 @@ def store_sched_result(conn, c, dp_id, result):
             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
+                print '(store_sched_result_chunk) Error is transaction deadlock. Will retry.'
+                print '(store_sched_result_chunk) Last query: %s' % c._last_executed
                 rollback(conn)
             else:
                 print e
-                print '(store_sched_result) Last query: %s' % c._last_executed
+                print '(store_sched_result_chunk) 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
+            print '(store_sched_result_chunk) Last query: %s' % c._last_executed
             rollback(conn)
             raise e
 
 def store_sched_results(db_name, data, ndp = 0):
-    global dp_col_names
     fetched = []
 
     conn = connect_db(db_name)
@@ -499,15 +542,15 @@ def store_sched_results(db_name, data, ndp = 0):
         # must be merged with prior results in the 'scaled_sched_results' table
         c = conn.cursor(db.cursors.DictCursor)
 
-        col_names = [n for n in dp_col_names if n != 'id']
-
         try:
-            # get IDs for all design points.
-            # 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(conn, c, dp_id, result)
+            # X design point results per chunk
+            chunk_size = 50
+            chunks = list(chunker(data, chunk_size))
+            for chk in chunks:
+                # get IDs for all design points
+                # sort to reduce (eliminate?) chance of deadlock
+                id_and_data = sorted([(lookup_dp_id(c, d.dp), d) for d in chk], key=lambda x: x[0])
+                store_sched_result_chunk(conn, c, id_and_data)
 
         except db.OperationalError, e:
             print '(store_sched_results) Caught database exception.'