Store sched results to SQL db instead of csv files

1 files changed, 212 insertions, 261 deletions

diff --git a/ecrts14/ecrts14.py b/ecrts14/ecrts14.py
index 2a81399..85c2ff6 100755
--- a/ecrts14/ecrts14.py
+++ b/ecrts14/ecrts14.py
@@ -9,8 +9,10 @@ import os
 import math
 import time
 
-#import logging, multiprocessing
+import sqlite3 as lite
+import json
 
+import copy
 from collections import defaultdict
 from csv import DictWriter
 from itertools import product
@@ -18,7 +20,6 @@ from math import ceil
 from multiprocessing import Pool, cpu_count
 from numpy import arange
 from pprint import pprint
-from copy import deepcopy
 import traceback
 
 from schedcat.model.tasks import SporadicTask, TaskSystem
@@ -40,8 +41,11 @@ import tests
 import topology
 from machines import machines
 
-import gc
-import resource
+#import gc
+#import resource
+import traceback
+
+import database as db
 
 NAMED_PERIODS_US = {
     # Named period distributions used in several UNC papers, in microseconds
@@ -50,8 +54,66 @@ NAMED_PERIODS_US = {
     'uni-long'      : tasks.uniform_int(50*1000, 250*1000),
 }
 
-MIN_SAMPLES = 500
-MAX_SAMPLES = 10000
+#based off of a 24-core system
+# fewer graphs = harder partitioning
+NAMED_NUM_GRAPHS = {
+    'uni-many'   : graph.uniform(24, 24*4),
+    'uni-medium' : graph.uniform(12, 48),
+    'uni-few'    : graph.uniform(1,12),
+
+    'bimo-many'   : graph.binomial(24, 24*4),
+    'bimo-medium' : graph.binomial(12, 48),
+    'bimo-few'    : graph.binomial(1,12),
+}
+
+NAMED_SHAPES = {
+    'uniform'  : graph.uniform(),
+    'binomial' : graph.binomial(),
+#    'geometric': graph.geometric(),
+}
+
+NAMED_HEIGHT_FACTORS = {
+    'uni-short'  : [1.0/3.0, 1.0/2.0],
+    'uni-medium' : [1.0/2.0, 3.0/4.0],
+    'uni-tall'   : [3.0/4.0, 1.0],
+}
+
+NAMED_FAN = {
+    'none'        : graph.uniform(1,1),
+    'uniform_3'   : graph.uniform(1,3),
+    'geometric_3' : graph.geometric(1,3),
+}
+
+NAMED_EDGE_HOP = {
+    'none'       : graph.uniform(1,1),
+    'uniform_3'  : graph.uniform(1,3),
+    'uniform_deep'  : graph.uniform(1,100),
+    'geometric_3': graph.geometric(1,3),
+}
+
+NAMED_EDGE_WSS = {
+    'uni-light'  : tasks.uniform(1, 64),
+    'uni-medium' : tasks.uniform(256, 1024),
+    'uni-heavy'  : tasks.uniform(2*1024, 8*1024),
+
+    'bimo-light'   : tasks.multimodal([(tasks.uniform_int(64,256), 8), (tasks.uniform_int(2*1024, 8*1024), 1)]),
+    'bimo-medium'  : tasks.multimodal([(tasks.uniform_int(64,256), 6), (tasks.uniform_int(2*1024, 8*1024), 3)]),
+    'bimo-heavy'   : tasks.multimodal([(tasks.uniform_int(64,256), 4), (tasks.uniform_int(2*1024, 8*1024), 5)]),
+}
+
+
+TESTS = [
+    (0, "CacheAgnostic", tests.test_partition_no_cache),
+    (1, "MaximizeParallelism", tests.test_partition_parallel),
+    (2, "CacheAware", tests.test_partition_cache_aware),
+    (3, "CacheAwareEdges", tests.test_partition_cache_aware_edges),
+#    ("MaximizeParallelismCacheAware", tests.test_partition_parallel2)
+    (4, "CacheAwareBFSEdges", tests.test_partition_cache_aware_bfs),
+    (5, "CacheAwareDFSEdges", tests.test_partition_cache_aware_dfs)
+]
+
+MIN_SAMPLES = 200
+MAX_SAMPLES = 500
 #MIN_SAMPLES = 1000
 #MAX_SAMPLES = 10000
 MAX_CI      = 0.05
@@ -159,8 +221,8 @@ def complete(results, n):
     elif n > MAX_SAMPLES:
         return True
     else:
-        for name, _ in TESTS:
-            if proportion_ci(results[name], n, CONFIDENCE) > MAX_CI:
+        for method, _, _ in TESTS:
+            if proportion_ci(results[method], n, CONFIDENCE) > MAX_CI:
                 return False
         return True
 
@@ -186,30 +248,43 @@ def get_overheads(dp, system = None):
     cluster_size = dp.processors/dp.nr_clusters
     max_dist = dp.system.distance(0, cluster_size-1)
     lvl = dp.system.levels[max_dist]
+    max_wss = dp.system.max_wss()
     ovh_file = 'overheads/ovh_host=%s_sched=%s_lvl=%s_type=%s.csv' % (dp.host, dp.sched, lvl, dp.ovh_type)
     ovh = Overheads.from_file(ovh_file)
     ovh.shared_cache = dp.system.schedcat_distance(0, max_dist)
     ovh.cache_affinity_loss = get_cpmds(dp)
+    ovh.cache_affinity_loss.set_max_wss(max_wss)
     ovh.consumer = get_consumer_overheads(dp, system)
     ovh.producer = get_producer_overheads(dp)
     return ovh
 
-def process_dp(dp):
-    
+def process_dp(_dp):
+
+    dp = copy.deepcopy(_dp)
+
     # kludge in parameters that pickle doesn't like...
     dp.system = topology.Topology(machines[dp.host])
     
-    dp.num_graphs = graph.binomial(1, 24)
-    dp.depth_factor = [1.0/3.0, 2.0/3.0]
-    dp.node_placement = graph.binomial()
-    dp.fan_out = graph.geometric(1, 3)
-    dp.fan_in_cap = 3
-    dp.edge_distance = graph.geometric(1, 3)
+    # convert names to distributions
+    dp.num_graphs = NAMED_NUM_GRAPHS[dp.num_graphs]
+    dp.depth_factor = NAMED_HEIGHT_FACTORS[dp.depth_factor]
+    dp.node_placement = NAMED_SHAPES[dp.node_placement]
+    dp.fan_out = NAMED_FAN[dp.fan_out]
+    dp.edge_distance = NAMED_EDGE_HOP[dp.edge_distance]
+    dp.wss = NAMED_EDGE_WSS[dp.wss]
+
+    # slam in unchaging values
     dp.nr_source = graph.uniform(1,1)
     dp.nr_sink = graph.uniform(1,1)
-    dp.wss = tasks.multimodal([(tasks.uniform_int(128,1024), 6), (tasks.uniform_int(2048, 8*1024), 3)])
+    dp.uncached = False
+    dp.huge_pages = False
+    dp.sched = 'edf'
+    dp.walk = 'seq'
     
     results = defaultdict(float)
+
+#    avg_nr_graphs = defaultdict(float)
+#    avg_graph_k   = defaultdict(float)
     avg_latencies = defaultdict(float)
     avg_ideal_ratios = defaultdict(float)
     avg_hrt_ratios = defaultdict(float)
@@ -219,276 +294,152 @@ def process_dp(dp):
 
     overheads = get_overheads(dp, dp.system)
     
-    # 512MB
-    upper_memlimit_bytes = 1024*1024*512
-    # ru_maxrss does not go back down, so we can only play this trick once...
-    if gc.isenabled() and resource.getrusage(resource.RUSAGE_SELF).ru_maxrss < upper_memlimit_bytes:
-        gc.disable()
-
     while not complete(results, n):
-        if (not gc.isenabled()) and (resource.getrusage(resource.RUSAGE_SELF).ru_maxrss > upper_memlimit_bytes):
-            gc.enable()
-
         ts, graphs, subts = create_pgm_task_set(dp)
 
-        hrt_ideal_response_times = map(graph.compute_hrt_ideal_response_time, graphs)
-        num_graphs = len(graphs)
+#        hrt_ideal_response_times = map(graph.compute_hrt_ideal_response_time, graphs)
+#        num_graphs = len(graphs)
+
         if dp.nr_clusters != 1:
-            for name, test in TESTS:
+            for method, _, test in TESTS:
                 result, processed_ts = test(ts, graphs, subts, dp, overheads)
                 if result:
-                    ideal_response_times = map(graph.compute_ideal_response_time, graphs)
-                    srt_response_times = map(graph.bound_graph_response_time, graphs)
-                    ideal_ratio = 0.0
-                    hrt_ratio = 0.0
-                    for i, h, s in zip(ideal_response_times, hrt_ideal_response_times, srt_response_times):
-                        ideal_ratio += s/i
-                        hrt_ratio += s/h
-                    
-                    ideal_ratio /= num_graphs
-                    hrt_ratio /= num_graphs
-                    avg_latency = sum(srt_response_times)/num_graphs
-                    avg_latencies[name] = update_mean(avg_latencies[name], nsched[name], avg_latency)
-                    avg_ideal_ratios[name] = update_mean(avg_ideal_ratios[name], nsched[name], ideal_ratio)
-                    avg_hrt_ratios[name] = update_mean(avg_hrt_ratios[name], nsched[name], hrt_ratio)
-                    nsched[name] += 1
-                results[name] = update_mean(results[name], n, result)
+#                    ideal_response_times = map(graph.compute_ideal_response_time, graphs)
+#                    srt_response_times = map(graph.bound_graph_response_time, graphs)
+#                    ideal_ratio = 0.0
+#                    hrt_ratio = 0.0
+#                    for i, h, s in zip(ideal_response_times, hrt_ideal_response_times, srt_response_times):
+#                        ideal_ratio += s/i
+#                        hrt_ratio += s/h
+#
+#                    ideal_ratio /= num_graphs
+#                    hrt_ratio /= num_graphs
+#                    avg_latency = sum(srt_response_times)/num_graphs
+#                    avg_latencies[method] = update_mean(avg_latencies[method], nsched[method], avg_latency)
+#                    avg_ideal_ratios[method] = update_mean(avg_ideal_ratios[method], nsched[method], ideal_ratio)
+#                    avg_hrt_ratios[method] = update_mean(avg_hrt_ratios[method], nsched[method], hrt_ratio)
+                    nsched[method] += 1
+                results[method] = update_mean(results[method], n, result)
+
         # if there is no partitioning, then same results hold for all tests
         else:
-            result, processed_ts = TESTS[0][1](ts, graphs, subts, dp, overheads)
+            result, processed_ts = TESTS[0][2](ts, graphs, subts, dp, overheads)
             if result:
-                ideal_response_times = map(graph.compute_ideal_response_time, graphs)
-                srt_response_times = map(graph.bound_graph_response_time, graphs)
-                ideal_ratio = 0.0
-                hrt_ratio = 0.0
-                for i, h, s in zip(ideal_response_times, hrt_ideal_response_times, srt_response_times):
-                    ideal_ratio += s/i
-                    hrt_ratio += s/h
-                
-                ideal_ratio /= num_graphs
-                hrt_ratio /= num_graphs
-                avg_latency = sum(srt_response_times)/num_graphs
-                for name, test in TESTS:
-                    avg_latencies[name] = update_mean(avg_latencies[name], nsched[name], avg_latency)
-                    avg_ideal_ratios[name] = update_mean(avg_ideal_ratios[name], nsched[name], ideal_ratio)
-                    avg_hrt_ratios[name] = update_mean(avg_hrt_ratios[name], nsched[name], hrt_ratio)
-                    nsched[name] += 1
-
-            for name, test in TESTS:
-                results[name] = update_mean(results[name], n, result)
+                for method, name, test in TESTS:
+                    nsched[method] += 1
+            for method, name, test in TESTS:
+                results[method] = update_mean(results[method], n, result)
 
         n += 1
-#        global TOTAL_TESTED
-#        TOTAL_TESTED += 1
-#        print TOTAL_TESTED
-
-    for name, test in TESTS:
-        if nsched[name] == 0:
-            avg_latencies[name] = -1.0
-            avg_ideal_ratios[name] = -1.0
-            avg_hrt_ratios[name] = -1.0
-
-    del dp.system
-    del dp.num_graphs
-    del dp.depth_factor
-    del dp.node_placement
-    del dp.fan_out
-    del dp.fan_in_cap
-    del dp.edge_distance
-    del dp.nr_source
-    del dp.nr_sink
-    del dp.wss
-
-#    return dict(dp.items() + results.items())
-#    return dict(dp.items() + results.items()), dict(dp.items() + avg_latencies.items()), dict(dp.items() + avg_ideal_ratios.items()), dict(dp.items() + avg_hrt_ratios.items())
-    return dp, results, avg_latencies, avg_ideal_ratios, avg_hrt_ratios
+
+#    for method, _, _ in TESTS:
+#        if nsched[method] == 0:
+#            avg_latencies[method] = -1.0
+#            avg_ideal_ratios[method] = -1.0
+#            avg_hrt_ratios[method] = -1.0
+
+    return results, avg_latencies, avg_ideal_ratios, avg_hrt_ratios
+
+def process_design_points(args):
+    try:
+        (worker_id, db_name) = args
+        nr_processed = 0
+        while True:
+            dp = db.get_design_point(db_name)
+            if dp and not db.already_processed(dp, db_name = db_name):
+                (sched, avg_lat, avg_idl_rat, avg_hrt_rat) = process_dp(dp)
+                db.store_sched_results(db_name, dp, sched)
+                nr_processed += 1
+            else:
+                break
+    except lite.OperationalError:
+        print "FUCK!"
+        print traceback.format_exc()
+    return nr_processed
 
 def valid(dp):
     return True
 
-TESTS = [
-    ("CacheAgnostic", tests.test_partition_no_cache),
-    ("MaximizeParallelism", tests.test_partition_parallel),
-    ("CacheAware", tests.test_partition_cache_aware),
-    ("CacheAwareEdges", tests.test_partition_cache_aware_edges),
-#    ("MaximizeParallelismCacheAware", tests.test_partition_parallel2)
-    ("CacheAwareBFSEdges", tests.test_partition_cache_aware_bfs),
-    ("CacheAwareDFSEdges", tests.test_partition_cache_aware_dfs)
-]
-
-def myrange(start, end, inc):
-    return arange(start, end+inc, inc)
-
-
 def main():
     random.seed(12345)
 
     parser = argparse.ArgumentParser()
-    parser.add_argument('-o', "--outfile", type = str,
-                        default = "",
-                        help = "store results to <filename>.csv")
     parser.add_argument('-p', "--pretend", action='store_true',
                         help = "Only print design point, do not execute")
     parser.add_argument('-m', "--processors", default=1, type = int,
                         help="Number of processors to execute on")
-#    parser.add_argument('-s', "--model", type = str,
-#                        default = "",
-#                        help = "Overhead model of the system")
+    parser.add_argument('-d', "--database", type = str,
+                        default = "",
+                        help = "Database for holding experiment data")
+    parser.add_argument('--initonly', action='store_true',
+                        help = "Only store design points to database")
+    parser.add_argument('--worker', action='store_true',
+                        help = "Only process design points from database")
+    parser.add_argument('--resume', action='store_true',
+                        help = "Preserve existing database entries")
     args = parser.parse_args()
 
-    exp = storage()
-    exp.host = ['ludwig']
-
-    cpus = 24.0
-    exp.processors = [cpus]
-#    exp.nr_clusters = [24]
-    exp.nr_clusters = [1]
-    exp.sched = ['edf']
-#    exp.nr_clusters = [1, 4, 12, 24]    
-    exp.task_util = ['uni-medium']
-    exp.period = ['uni-long']
-    exp.sys_util = myrange(1, cpus, 0.1)
-    #
-#    exp.num_graphs = [graph.binomial(1, 24)]
-#    exp.depth_factor = [[1.0/3.0, 2.0/3.0]]
-#    exp.node_placement = [graph.binomial()]
-#    exp.fan_out = [graph.geometric(1, 3)]
-#    exp.fan_in_cap = [3]
-#    exp.edge_distance = [graph.geometric(1, 3)]
-#    exp.nr_source = [graph.uniform(1,1)]
-#    exp.nr_sink = [graph.uniform(1,1)]
-#    
-#    exp.wss = [ tasks.multimodal([(tasks.uniform_int(128,1024), 6), (tasks.uniform_int(2048, 8*1024), 3)]) ]
-    exp.wcycle = [ 0 ]
-    exp.walk = ['seq']
-    exp.huge_pages = [False]
-    exp.uncached = [False]
-    exp.polluters = [False]
-    exp.ovh_type = ['max']
-#    exp.ovh_type = ['max', 'median', 'mean']
-
-    design_points = [dp for dp in DesignPointGenerator(exp, is_valid = valid)]
-    design_points.reverse()
-
-    # hopefully this makes the % done more indicative of progress.
-#    random.shuffle(design_points)
-
-    print "Total design points: ", len(design_points)
-
-    if not args.pretend:
-#        if args.outfile == "":
-#            sched_out = DictWriter(sys.stdout, exp.keys()+[t[0] for t in TESTS])
-#            lat_out = DictWriter(open(os.devnull, 'w'), exp.keys()+[t[0] for t in TESTS])
-#            ir_out = DictWriter(open(os.devnull, 'w'), exp.keys()+[t[0] for t in TESTS])
-#            hr_out = DictWriter(open(os.devnull, 'w'), exp.keys()+[t[0] for t in TESTS])
-#        else:
-#            sched_out = DictWriter(open(args.outfile+'_sched.csv', 'w'), exp.keys()+[t[0] for t in TESTS])
-#            lat_out = DictWriter(open(args.outfile+'_latency.csv', 'w'), exp.keys()+[t[0] for t in TESTS])
-#            ir_out = DictWriter(open(args.outfile+'_idealratio.csv', 'w'), exp.keys()+[t[0] for t in TESTS])
-#            hr_out = DictWriter(open(args.outfile+'_hrtratio.csv', 'w'), exp.keys()+[t[0] for t in TESTS])
-#        
-#        sched_out.writeheader()
-#        lat_out.writeheader()
-#        ir_out.writeheader()
-#        hr_out.writeheader()
-
-        if args.outfile == "":
-            sched_out = sys.stdout
-            lat_out = open(os.devnull, 'w')
-            ir_out = open(os.devnull, 'w')
-            hr_out = open(os.devnull, 'w')
-        else:
-            sched_out = open(args.outfile+'_sched.csv', 'w')
-            lat_out = open(args.outfile+'_latency.csv', 'w')
-            ir_out = open(args.outfile+'_idealratio.csv', 'w')
-            hr_out = open(args.outfile+'_hrtratio.csv', 'w')
-
-        hdr = 'processors,nr_clusters,task_util,period,wcycle,polluters,ovh_type,sys_util'
-        for t in TESTS:
-            hdr += ','+t[0]
-        hdr += '\n'
-        
-        sched_out.write(hdr)
-        lat_out.write(hdr)
-        ir_out.write(hdr)
-        hr_out.write(hdr)
-
-        if args.processors > 1:
-            pool = Pool(processes = args.processors)
-#            logger = multiprocessing.log_to_stderr()
-#            logger.setLevel(multiprocessing.SUBDEBUG)
-            try:
-                for i, row in enumerate(pool.imap_unordered(process_dp, design_points)):
-                    if not row:
-                        continue
-                    if sched_out != sys.stdout:
-                        sys.stderr.write('\rdone {0:%}'.format(i/len(design_points)))
-
-                    dp, sched, latency, iratio, hrtratio = row
-                    
-                    keys = '%d,%d,%s,%s,%d,%d,%s,%f' % (dp.processors, dp.nr_clusters, dp.task_util, dp.period, dp.wcycle, dp.polluters, dp.ovh_type, dp.sys_util)
-                    
-                    values = ''
-                    for t in TESTS:
-                        values += ',%f' % sched[t[0]]
-                    sched_out.write('%s%s\n' % (keys, values))
-                    
-                    values = ''
-                    for t in TESTS:
-                        values += ',%f' % latency[t[0]]
-                    lat_out.write('%s%s\n' % (keys, values))
-                    
-                    values = ''
-                    for t in TESTS:
-                        values += ',%f' % iratio[t[0]]
-                    ir_out.write('%s%s\n' % (keys, values))
-                    
-                    values = ''
-                    for t in TESTS:
-                        values += ',%f' % hrtratio[t[0]]
-                    hr_out.write('%s%s\n' % (keys, values))
-
-                pool.close()
-
-            except Exception as e:
-                pool.terminate()
-                print e
-                raise
-        else:
-
-            for i, row in enumerate(map(process_dp, design_points)):
-                if not row:
-                    continue
-                if sched_out != sys.stdout:
-                    sys.stderr.write('\rdone {0:%}'.format(i/len(design_points)))
-
-                dp, sched, latency, iratio, hrtratio = row
-                
-                entry_stem = '%d,%d,%s,%s,%d,%d,%s,%f' % (dp.processors, dp.nr_clusters, dp.task_util, dp.period, dp.wcycle, dp.polluters, dp.ovh_type, dp.sys_util)
-                
-                sched_out.write(entry_stem)
-                for t in TESTS:
-                    sched_out.write(',%f' % sched[t[0]])
-                sched_out.write('\n')
-                
-                lat_out.write(entry_stem)
-                for t in TESTS:
-                    lat_out.write(',%f' % latency[t[0]])
-                lat_out.write('\n')
-                
-                ir_out.write(entry_stem)
-                for t in TESTS:
-                    ir_out.write(',%f' % iratio[t[0]])
-                ir_out.write('\n')
-                
-                hr_out.write(entry_stem)
-                for t in TESTS:
-                    hr_out.write(',%f' % hrtratio[t[0]])
-                hr_out.write('\n')
-
-#        global TOTAL_TESTED
-#        print 'total tasksets:', TOTAL_TESTED
+    if args.database == "":
+        print "Database name required."
+        exit(-1)
+
+    if not args.worker:
+        cpus = 24.0
+        exp = storage()
+
+        # system parameters
+        exp.processors = [int(cpus)]
+#        exp.nr_clusters = [1, 4, 12, 24]
+        exp.nr_clusters = [1, 4, 12]
+        exp.host = ['ludwig']
+        exp.polluters = [False]
+        exp.ovh_type = ['max']
+
+        # task parameters
+        step_size = 0.2
+        exp.sys_util =  [float(v) for v in arange(6.0, cpus+step_size, step_size)]
+        exp.task_util = ['uni-medium']
+        exp.period =    ['uni-long']
+        exp.wcycle =    [0]
+
+        # graph parameters
+        exp.num_graphs =     ['bimo-few', 'bimo-medium']
+#        exp.depth_factor =   ['uni-medium']
+        exp.depth_factor =   ['uni-short', 'uni-tall']
+        exp.node_placement = ['binomial']
+        exp.fan_out =        ['geometric_3']
+        exp.edge_distance =  ['geometric_3']
+        exp.wss =            ['bimo-medium']
+        exp.fan_in_cap =     [3]
+
+        design_points = [dp for dp in DesignPointGenerator(exp, is_valid = valid)]
+
+        nr_dp = len(design_points)
+        if not args.pretend:
+            db.create_tables(args.database, dummy_dp = design_points[0], clean = not args.resume)
+            num_stored = db.store_design_points(args.database, design_points, clean = not args.resume)
+        print "Loaded %d of %d design points. (%d already completed)" % (num_stored, nr_dp, nr_dp - num_stored)
+
+    if args.pretend or args.initonly:
+        exit(0)
+
+    total_nr_processed = 0
+    if args.processors > 1:
+        pool = Pool(processes = args.processors)
+        args = zip(range(args.processors), [args.database]*args.processors)
+        try:
+            for i,nr_processed in enumerate(pool.map(process_design_points, args)):
+                print 'worker %d: processed %d design points.' % (i,nr_processed)
+                total_nr_processed += nr_processed
+            pool.close()
+        except Exception as e:
+            pool.terminate()
+            print e
+            raise
+    else:
+       total_nr_processed = process_design_points((0, args.database))
 
+    print 'Processed %d design points!' % total_nr_processed
 
 if __name__ == '__main__':
     main()