Optimize dp fetch from database

3 files changed, 115 insertions, 164 deletions

diff --git a/rtss14/createtables.sql b/rtss14/createtables.sql
index a072aa4..a76a579 100644
--- a/rtss14/createtables.sql
+++ b/rtss14/createtables.sql
@@ -41,6 +41,9 @@ INSERT INTO distrs(name) VALUES('c-const-heavy');
 CREATE TABLE dp_pending(
         id   INTEGER NOT NULL AUTO_INCREMENT,
 
+        -- flag design point as taken 
+        taken            INTEGER,
+
         -- task set util cap
         ts_util          REAL,
 
@@ -76,7 +79,9 @@ CREATE TABLE dp_pending(
         ncopy_engines    INTEGER,
         chunk_size       INTEGER,
 
-        PRIMARY KEY (id)
+        PRIMARY KEY (id),
+        INDEX (id, taken),
+        INDEX (ts_util)
         -- every permutation marks a unique configuration
         -- PRIMARY KEY(ts_util,
         --      ncpu, cpu_cluster_size, ngpu, gpu_cluster_size, is_release_master,
@@ -146,7 +151,8 @@ CREATE TABLE sched_results(
         avg_bandwidth REAL,
 
         FOREIGN KEY(dp) REFERENCES dp_ptested(id),
-        PRIMARY KEY(dp, ts_util)
+        PRIMARY KEY(dp, ts_util),
+        INDEX (ts_util)
 ) ENGINE=InnoDB;
 
 CREATE TABLE scaled_sched_results(
diff --git a/rtss14/database.py b/rtss14/database.py
index cd5535d..c0b099a 100755
--- a/rtss14/database.py
+++ b/rtss14/database.py
@@ -70,7 +70,7 @@ def init_dp_col_names(conn):
             dp_col_type_strs = {}
             for field in cols:
                 name = field[0].encode('ascii', 'ignore')
-                if(name == 'id'):
+                if(name == 'id' or name == 'taken'):
                     continue
                 dp_col_names.append(name)
                 form = '%s'
@@ -249,40 +249,6 @@ def clear_tables(db_name):
         c.close()
         ################
 
-#def get_results(conn, dp, fields = '*', table = 'sched_results', extra = None):
-#    where = ['%s=?' % k for k in dp.iterkeys()]
-#
-#    ################
-#    c = conn.cursor(db.cursors.DictCursor)
-#    begin_sync(conn)
-#    if extra:
-#        query = "SELECT %s FROM %s WHERE %s %s" % (", ".join(fields), table, " AND ".join(where), extra)
-#    else:
-#        query = "SELECT %s FROM %s WHERE %s" % (", ".join(fields), table, " AND ".join(where))
-#
-#    c.execute(query, tuple(dp.values()))
-#    data = c.fetchall()
-#
-#    end_sync(conn)
-#    ##############
-#
-#    return data
-
-#def count_results(conn, dp, table = 'sched_results'):
-#    c = conn.cursor(db.cursors.DictCursor)
-#    where = ['%s=?' % k for k in dp.iterkeys()]
-#
-#    query = "SELECT COUNT(*) FROM %s WHERE %s" % (table, " AND ".join(where))
-#
-#    ################
-#    begin_sync(conn)
-#    c.execute(query, tuple(dp.values()))
-#    data = c.fetchone()
-#    end_sync(conn)
-#    ##############
-#
-#    return data[0]
-
 def __get_dp_id(c, dp, add_if_missing = False):
     global dp_col_names
     global dp_col_type_strs
@@ -305,71 +271,85 @@ def __get_dp_id(c, dp, add_if_missing = False):
     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 %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)
+#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)
 
 def store_design_points(db_name, dps, clean):
     global dp_col_names
     global dp_col_type_strs
 
+    dps = [dp_to_db(dp) for dp in dps]
+    nstored = 0
+
     conn = connect_db(db_name)
-    npending = 0
     with conn:
         c = conn.cursor()
-
-        # convert to dicts with db names
-        if not clean:
-            dps = [dp_to_db(dp) for dp in dps if not (already_processed(dp, conn = conn) or already_pending(dp, conn = conn))]
+        if clean:
+            c.executemany('INSERT INTO dp_pending VALUES(NULL,0,%s)' % ','.join(['%s']*len(dp_col_names)),
+                [tuple(map(lambda x: d[x], dp_col_names)) for d in dps])
+            nstored = len(dps)
         else:
-            dps = [dp_to_db(dp) for dp in dps]
+            col_names = copy.deepcopy(dp_col_names)
+            col_names.pop(0)  # remove 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])
+            # 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)))
+            # 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)))
+            nstored = c.rowcount
+            c.execute('DROP TEMPORARY TABLE tmp_dp')
+        c.close()
+    return nstored
 
-        ################
-        # convert dicts into correctly-ordered list and insert
-        c.executemany('INSERT INTO dp_pending VALUES(NULL,%s)' % ','.join(['%s']*len(dp_col_names)),
-            [tuple(map(lambda x: d[x], dp_col_names)) for d in dps])
-        # complete
-        c.execute('SELECT COUNT(*) FROM dp_pending')
+def num_pending_design_points(db_name):
+    conn = connect_db(db_name)
+    with conn:
+        c = conn.cursor()
+        c.execute('SELECT COUNT(*) FROM dp_pending WHERE taken=0')
         npending = c.fetchone()[0]
-        ##############
-
         c.close()
-    return len(dps), npending
-
+    return npending
 
 def get_design_points(db_name, ndp = 1):
     global dp_col_names
@@ -381,44 +361,31 @@ def get_design_points(db_name, ndp = 1):
     failcount = 0
     while True:
         try:
-            retry = True
-            while retry is True:
-                retry = False
-
-                if conn is not None:
-                    conn.close()
-                    del conn
-                    conn = None
-
-                conn = connect_db(db_name)
-                with conn:
-                    ################
-                    c = conn.cursor(db.cursors.DictCursor)
-                    begin_sync(conn, c)
-#                    c.execute('LOCK TABLES dp_pending WRITE')
-                    c.execute('SELECT %s FROM dp_pending LIMIT %%s FOR UPDATE' % (','.join(dp_col_names)), (ndp,))
-                    dps = [db_to_dp(d) for d in c.fetchall()]
-                    nfetched = len(dps) if dps else 0
-                    if nfetched > 0:
-                        temp = [dp_to_db(d) for d in dps]
-                        for i in xrange(nfetched):
-                            c.execute('DELETE FROM dp_pending '
-                                      'WHERE %s' % ' AND '.join(map(lambda x: '%s=%s'%(x, dp_col_type_strs[x]), dp_col_names)),
-                                      tuple(map(lambda x: temp[i][x], dp_col_names)))
-                            if c.rowcount == 1:
-                                fetched.append(dps[i])
-                            elif c.rowcount == 0:
-                                print 'raced for design point. dropping.'
-                            else:
-                                print 'deleted too many rows... dropping.'
-                        if len(fetched) == 0:
-                            retry = True
-#                    c.execute('UNLOCK TABLES')
-                    end_sync(conn)
-                    c.close()
-                    ##############
-            # success!
-            break
+            if conn is not None:
+                conn.close()
+                del conn
+                conn = None
+
+            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()
+                break
+                ##############
         except db.OperationalError, e:
             print e
             # make sure the db connection is closed
@@ -518,24 +485,16 @@ def store_sched_results(db_name, data, ndp = 0):
 
                 if ndp > 0:
                     begin_sync(conn, c)
-#                    c.execute('LOCK TABLES dp_pending WRITE')
-                    c.execute('SELECT %s FROM dp_pending LIMIT %%s FOR UPDATE' % (','.join(dp_col_names)), (ndp,))
-                    dps = [db_to_dp(d) for d in c.fetchall()]
-                    nfetched = len(dps) if dps else 0
+                    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:
-                        temp = [dp_to_db(d) for d in dps]
-                        for i in xrange(nfetched):
-                            c.execute('DELETE FROM dp_pending '
-                                      'WHERE %s' % ' AND '.join(map(lambda x: '%s=%s'%(x, dp_col_type_strs[x]), dp_col_names)),
-                                      tuple(map(lambda x: temp[i][x], dp_col_names)))
-                            if c.rowcount == 1:
-                                fetched.append(dps[i])
-                            elif c.rowcount == 0:
-                                print 'store_sched_results: raced for design point. dropping.'
-                            else:
-                                print 'store_sched_results: deleted too many rows... dropping.'
-#                    c.execute('UNLOCK TABLES')
+                        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()
             # success!
             break
diff --git a/rtss14/rtss14.py b/rtss14/rtss14.py
index f049951..ee3564a 100755
--- a/rtss14/rtss14.py
+++ b/rtss14/rtss14.py
@@ -785,7 +785,6 @@ def get_dp_space(cpus):
     exp.ncpus = [12]
     exp.nclusters = [1, 2, 12]
     exp.ngpus = [8]
-#    exp.ngclusters = [1, 2, 4, 8]
     exp.ngclusters = [1, 2, 4, 8]
     exp.release_master = [False]
     exp.polluters = [False, True]
@@ -794,7 +793,6 @@ def get_dp_space(cpus):
 
     # gpusync config variables
     exp.rho = [0, 1, 2, 3]
-#    exp.rho = [1, 3]
     exp.dgl = [True, False]
     exp.p2p = [True, False]
     exp.ncopy_engines = [2]
@@ -802,8 +800,8 @@ def get_dp_space(cpus):
 
     # task parameters
 #    step_size = 0.1
-#    step_size = 0.5
-    step_size = 0.25
+    step_size = 0.5
+#    step_size = 0.25
     start_pt = step_size
 #    start_pt = 10.0
     exp.sys_util =  [float(v) for v in arange(start_pt, cpus+step_size, step_size)]
@@ -891,11 +889,10 @@ def main():
 
         # load design points incrementally as to not overload memory
         gen = DesignPointGenerator(exp, is_valid = valid)
-        dp_chunk_sz = 10000  # number of design points to load at a time
+        dp_chunk_sz = 50000  # number of design points to load at a time
         ndp = 0
         if not args.pretend:
             nstored = 0
-            npending = 0
             while True:
                 dp_chunk = []
                 for _ in range(dp_chunk_sz):
@@ -907,26 +904,15 @@ def main():
                     break
                 ndp += len(dp_chunk)
                 random.shuffle(dp_chunk)
-                tmpnstored, npending = db.store_design_points(args.database, dp_chunk, clean = not args.resume)
+                tmpnstored = db.store_design_points(args.database, dp_chunk, clean = not args.resume)
                 nstored += tmpnstored
-            print "Loaded %d of %d design points. (%d were completed or pending, %d now pending)" % (nstored, ndp, ndp - nstored, npending)
+            npending = db.num_pending_design_points(args.database)
+            print "Loaded %d of %d design points. (%d were completed, %d now pending)" % (nstored, ndp, ndp - npending, npending)
         else:
             for _ in gen:
                 ndp += 1
             print "%d design points planned" % ndp
 
-#        design_points = [dp for dp in DesignPointGenerator(exp, is_valid = valid)]
-#        ndp = len(design_points)
-#        if not args.pretend:
-#            random.shuffle(design_points)
-#            if not args.resume:
-#                db.clear_tables(args.database)
-#            nstored, npending = db.store_design_points(args.database, design_points, clean = not args.resume)
-#            print "Loaded %d of %d design points. (%d were completed or pending, %d now pending)" % (
-#                nstored, ndp, ndp - nstored, npending)
-#        else:
-#            print "%d design points planned" % ndp
-
     if args.pretend or args.initonly:
         exit(0)