Fix race in dp ID generation in dp_ptested

Race allowed a single design point to get multiple
dp ids in dp_ptested.

1 files changed, 160 insertions, 106 deletions

diff --git a/rtss14/database.py b/rtss14/database.py
index c0b099a..e9fe99b 100755
--- a/rtss14/database.py
+++ b/rtss14/database.py
@@ -43,6 +43,8 @@ def init_distr_mapper(conn):
             break
         except db.OperationalError, e:
             print e
+            if c is not None:
+                print c._last_executed
             elapsed = time.time() - start
             if elapsed > timeout:
                 raise
@@ -78,6 +80,8 @@ def init_dp_col_names(conn):
             break
         except db.OperationalError, e:
             print e
+            if c is not None:
+                print c._last_executed
             elapsed = time.time() - start
             if elapsed > timeout:
                 raise
@@ -192,6 +196,8 @@ def begin_sync(conn, c):
             break
         except db.OperationalError, e:
             print e
+            if c is not None:
+                print c._last_executed
             elapsed = time.time() - start
             if elapsed > transaction_timeout:
                 raise
@@ -205,6 +211,9 @@ def end_sync(conn):
     conn.autocommit(True)
     return success
 
+def rollback(conn):
+    conn.rollback()
+    conn.autocommit(True)
 
 def connect_db(db_name):
     global distr_mapper
@@ -249,74 +258,49 @@ def clear_tables(db_name):
         c.close()
         ################
 
-def __get_dp_id(c, dp, add_if_missing = False):
+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)
-    query = 'SELECT id FROM dp_ptested WHERE %s' % ' AND '.join(map(lambda x: '%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys()))
-    c.execute(query, tuple(seeking.values()))
-    row = c.fetchone()
-    if row:
-        return row['id']
-    elif add_if_missing == False:
-        return None
-    # add the design point
-    col_names = copy.deepcopy(dp_col_names)
-    col_names.pop(0) # remove the id field
-    c.execute('INSERT INTO dp_ptested VALUES(NULL,%s)' % ','.join(['%s']*len(col_names)),
-        tuple(map(lambda x: seeking[x], col_names)))
-    # get the assigned id
-    c.execute(query, tuple(seeking.values()))
-    row = c.fetchone()
-    return row['id']
-
-#def __already_pending(dp, conn):
-#    seeking = dp_to_db(dp)
-#    c = conn.cursor()
-#    c.execute('SELECT COUNT(*) FROM dp_pending WHERE taken=0 AND %s' % ' AND '.join(map(lambda x: '%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys())),
-#            tuple(seeking.values()))
-#    row = c.fetchone()
-#   processed = bool(row[0])
-#    c.close()
-#    return processed
-
-#def already_pending(dp, conn = None, db_name = None):
-#    if conn:
-#        return __already_pending(dp, conn)
-#    else:
-#        conn = connect_db(db_name)
-#        with conn:
-#           return __already_pending(dp, conn)
-
-#def __already_processed(dp, conn):
-#    seeking = dp_to_db(dp)
-#    seeking.pop('ts_util', None)
-#    c = conn.cursor()
-#    c.execute('SELECT COUNT(*) FROM sched_results as R JOIN dp_ptested as K on R.dp=K.id '
-#              'WHERE R.ts_util=%%s AND %s' % ' AND '.join(map(lambda x: 'K.%s=%s' % (x, dp_col_type_strs[x]), seeking.iterkeys())),
-#            ((dp.sys_util,) + tuple(seeking.values())))
-#    row = c.fetchone()
-#    processed = bool(row[0])
-#    c.close()
-#    return processed
 
-#def already_processed(dp, conn = None, db_name = None):
-#    if conn:
-#        return __already_processed(dp, conn)
-#    else:
-#        conn = connect_db(db_name)
-#        with conn:
-#            return __already_processed(dp, conn)
+    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()')
+        dp_id = c.fetchone().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()) )
+        dp_id = c.fetchone().values()[0]
+
+    return dp_id
 
-def store_design_points(db_name, dps, clean):
+def store_design_points(db_name, dps, clean, fix = False):
     global dp_col_names
     global dp_col_type_strs
 
+    conn = connect_db(db_name)
+
     dps = [dp_to_db(dp) for dp in dps]
     nstored = 0
 
-    conn = connect_db(db_name)
     with conn:
         c = conn.cursor()
         if clean:
@@ -324,19 +308,30 @@ def store_design_points(db_name, dps, clean):
                 [tuple(map(lambda x: d[x], dp_col_names)) for d in dps])
             nstored = len(dps)
         else:
-            col_names = copy.deepcopy(dp_col_names)
-            col_names.pop(0)  # remove ts_util
+            col_names = [n for n in dp_col_names if n != 'ts_util']
             # store entire dp set in a temporary staging table
             c.execute('CREATE TEMPORARY TABLE tmp_dp LIKE dp_pending')
-            c.executemany('INSERT INTO tmp_dp VALUES(NULL,0,%s)' % ','.join(['%s']*len(dp_col_names)),
-                [tuple(map(lambda x: d[x], dp_col_names)) for d in dps])
+            # filter out dps that are already pending and not taken.
+            # entries taken in dp_pending may still be mirrored here.
+            c.executemany('INSERT INTO tmp_dp(id,taken,%s) '
+                          'SELECT NULL,0,%s '
+                          'FROM dual '
+                          'WHERE NOT EXISTS ( '
+                              'SELECT 1 '
+                              'FROM dp_pending '
+                              'WHERE taken=0 AND %s )' % ( ','.join(dp_col_names),
+                                                           ','.join(['%s']*len(dp_col_names)),
+                                                           ' AND '.join(map(lambda x: '%s=%%s' % x, dp_col_names)) ),
+                          [tuple(map(lambda x: d[x], dp_col_names)*2) for d in dps])
             # mask out the dps that have already been processed
             c.execute('UPDATE tmp_dp as S JOIN sched_results as R on S.ts_util=R.ts_util JOIN dp_ptested as T on R.dp=T.id '
                       'SET S.taken=1 WHERE %s' % ' AND '.join(map(lambda x: 'S.%s=T.%s' % (x,x), col_names)))
-            # mask out the dps that are already pending
-            c.execute('UPDATE tmp_dp as S JOIN dp_pending as P on %s SET S.taken=1' % ' AND '.join(map(lambda x: 'S.%s=P.%s' % (x,x), dp_col_names)))
+            # reset dp_pending.taken to 0 if the dp is actually unfinished, and mask out the duplicate in tmp_dp
+            c.execute('UPDATE dp_pending as P JOIN tmp_dp as S on %s '
+                      'SET P.taken=0, S.taken=1 '
+                      'WHERE P.taken=1 AND S.taken=0' % ' AND '.join(map(lambda x: 'P.%s=S.%s' % (x,x), dp_col_names)))
             # copy any remaining dps into the pending table
-            c.execute('INSERT INTO dp_pending(taken,%s) SELECT taken,%s FROM tmp_dp WHERE tmp_dp.taken=0' % (','.join(dp_col_names), ','.join(dp_col_names)))
+            c.execute('INSERT INTO dp_pending(id,taken,%s) SELECT NULL,0,%s FROM tmp_dp WHERE tmp_dp.taken=0' % (','.join(dp_col_names), ','.join(dp_col_names)))
             nstored = c.rowcount
             c.execute('DROP TEMPORARY TABLE tmp_dp')
         c.close()
@@ -351,9 +346,83 @@ def num_pending_design_points(db_name):
         c.close()
     return npending
 
-def get_design_points(db_name, ndp = 1):
+def ___get_design_points(conn, ndp):
     global dp_col_names
-    global dp_col_type_strs
+    fetched = []
+    c = conn.cursor(db.cursors.DictCursor)
+    ####################
+    begin_sync(conn, c)
+    c.execute('SELECT id FROM dp_pending WHERE taken=0 LIMIT %s FOR UPDATE', (ndp,))
+    dp_ids = [d['id'] for d in c.fetchall()]
+    nfetched = len(dp_ids) if dp_ids else 0
+    if nfetched > 0:
+        c.execute('UPDATE dp_pending SET taken=1 WHERE %s' % ' OR '.join(['id=%s']*len(dp_ids)), tuple(dp_ids))
+    end_sync(conn)
+    ####################
+    if nfetched > 0:
+        c.execute('SELECT %s FROM dp_pending WHERE %s' % (','.join(dp_col_names), ' OR '.join(['id=%s']*len(dp_ids))), tuple(dp_ids))
+        fetched = [db_to_dp(d) for d in c.fetchall()]
+    c.close()
+    return fetched, nfetched
+
+
+#def ___get_design_points(conn, ndp):
+#   global dp_col_names
+#    fetched = []
+#    c = conn.cursor(db.cursors.DictCursor)
+#    # get a unique id to prevent races
+#    c.execute('INSERT INTO dp_id_gen DEFAULT VALUES')
+#    begin_sync(conn, c)
+#    # try to get some ids of up to 'ndp' design points
+#    c.execute('SELECT id FROM dp_pending WHERE taken=0 LIMIT %s FOR UPDATE', (ndp,))
+#    dp_ids = [d['id'] for d in c.fetchall()]
+#    nfetched = len(dp_ids) if dp_ids else 0
+#    nclaimed = 0
+#    if nfetched > 0:
+#        # if there was work...
+#        # take fetched rows and assign our unique id
+#        # conjunction dp ids with 'taken=0' elimates races
+#        c.execute('UPDATE dp_pending SET taken=LAST_INSERT_ID() WHERE (taken=0) AND (%s)' % ' OR '.join(['id=%s']*len(dp_ids)), tuple(dp_ids))
+#        nclaimed = c.rowcount
+#    end_sync(conn)
+#    if nclaimed > 0:
+#        # fetch our design points
+#        c.execute('SELECT %s FROM dp_pending WHERE taken=LAST_INSERT_ID()' % (','.join(dp_col_names)))
+#        fetched = [db_to_dp(d) for d in c.fetchall()]
+#        # reset taken to be '1'
+#        c.execute('UPDATE dp_pending SET taken=1 WHERE taken=LAST_INSERT_ID()')
+#    # release our id
+#    c.execute('DELETE FROM dp_id_gen WHERE id=LAST_INSERT_ID()')
+#    c.close()
+#    if nfetched > 0 and nclaimed == 0:
+#        # there was a race of some sort
+#        nfetched = = -1
+#    return fetched, nfetched
+
+def __get_design_points(conn, ndp = 1):
+    failcount = 0
+    nfetched = -1
+    fetched = []
+
+    while nfetched < 0 and failcount < 5:
+
+        # sleep a moment before we retry
+        if failcount > 0:
+            time_to_sleep = (random.random()*20 + 5)*failcount
+            time_to_sleep = min(time_to_sleep, 4*60)
+            print 'retrying ___get_design_points due to race. sleep = %.3f' % time_to_sleep
+            time.sleep(time_to_sleep)
+
+        # try to get design points
+        fetched, nfetched = ___get_design_points(conn, ndp)
+
+        if nfetched < 0:
+            failcount += 1
+
+    return fetched
+
+
+def get_design_points(db_name, ndp = 1):
     global max_fail
     fetched = []
 
@@ -368,24 +437,9 @@ def get_design_points(db_name, ndp = 1):
 
             conn = connect_db(db_name)
             with conn:
-                ################
-                c = conn.cursor(db.cursors.DictCursor)
-
-                begin_sync(conn, c)
-                c.execute('SELECT id FROM dp_pending WHERE taken=0 LIMIT %s FOR UPDATE', (ndp,))
-                dp_ids = [d['id'] for d in c.fetchall()]
-                nfetched = len(dp_ids) if dp_ids else 0
-                if nfetched > 0:
-                    c.execute('UPDATE dp_pending SET taken=1 WHERE %s' % ' OR '.join(['id=%s']*len(dp_ids)), tuple(dp_ids))
-                end_sync(conn)
-
-                if nfetched > 0:
-                    c.execute('SELECT %s FROM dp_pending WHERE %s' % (','.join(dp_col_names), ' OR '.join(['id=%s']*len(dp_ids))), tuple(dp_ids))
-                    fetched = [db_to_dp(d) for d in c.fetchall()]
-
-                c.close()
+                fetched = __get_design_points(conn, ndp)
                 break
-                ##############
+
         except db.OperationalError, e:
             print e
             # make sure the db connection is closed
@@ -445,7 +499,7 @@ def __store_eff_sched_results(c, dp_id, stats):
                 c.execute('INSERT INTO scaled_sched_results '
                           '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_results(db_name, data, ndp = 0):
     global dp_col_names
     global dp_col_type_strs
@@ -456,6 +510,8 @@ def store_sched_results(db_name, data, ndp = 0):
 
     conn = None
     failcount = 0
+    saved = False
+
     while True:
         try:
             # data is a list of storage(), with fields 'dp, 'sched_stats', and 'eff_sched_stats'.
@@ -464,43 +520,41 @@ def store_sched_results(db_name, data, ndp = 0):
             conn = connect_db(db_name)
 
             if col_names is None:
-                col_names = copy.deepcopy(dp_col_names)
-                col_names.pop(0) # remove the id field
+                col_names = [n for n in dp_col_names if n != 'id']
 
             with conn:
                 c = conn.cursor(db.cursors.DictCursor)
+                if not saved:
+                    # get IDs for all design points
+                    dp_ids = [lookup_dp_id(c, d.dp) for d in data]
 
-                # get IDs for all design points
-                dp_ids = [__get_dp_id(c, d.dp, add_if_missing = True) for d in data]
-                # insert the normal sched data in one go
-                c.executemany('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)
-                        for dp_id, ts_util, stats in zip(dp_ids, [d.dp.sys_util for d in data], [d.sched_stats for d in data])])
-
-                if d.eff_sched_stats is not None:
+                    # insert the normal sched data in one go
                     begin_sync(conn, c)
-                    for dp_id, d in zip(dp_ids, data):
-                        __store_eff_sched_results(c, dp_id, d.eff_sched_stats)
+
+                    # save results
+                    c.executemany('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)
+                            for dp_id, ts_util, stats in zip(dp_ids, [d.dp.sys_util for d in data], [d.sched_stats for d in data])])
+
+                    # save scaled results
+                    if d.eff_sched_stats is not None:
+                        for dp_id, d in zip(dp_ids, data):
+                            __store_eff_sched_results(c, dp_id, d.eff_sched_stats)
+
                     end_sync(conn)
+                    saved = True
 
                 if ndp > 0:
-                    begin_sync(conn, c)
-                    c.execute('SELECT id FROM dp_pending WHERE taken=0 LIMIT %s FOR UPDATE', (ndp,))
-                    dp_ids = [d['id'] for d in c.fetchall()]
-                    nfetched = len(dp_ids) if dp_ids else 0
-                    if nfetched > 0:
-                        c.execute('UPDATE dp_pending SET taken=1 WHERE %s' % ' OR '.join(['id=%s']*len(dp_ids)), tuple(dp_ids))
-                    end_sync(conn)
+                    fetched = __get_design_points(conn, ndp)
 
-                    if nfetched > 0:
-                        c.execute('SELECT %s FROM dp_pending WHERE %s' % (','.join(dp_col_names), ' OR '.join(['id=%s']*len(dp_ids))), tuple(dp_ids))
-                        fetched = [db_to_dp(d) for d in c.fetchall()]
                 c.close()
-            # success!
             break
         except db.OperationalError, e:
             print e
+            if c is not None:
+                print c._last_executed
             if conn is not None:
+                rollback(conn)
                 conn.close()
                 del conn
                 conn = None