Dump of progress.

This ugly patch updates code and
checks in overhead data for ludwig.

1 files changed, 494 insertions, 0 deletions

diff --git a/ecrts14/ecrts14.py b/ecrts14/ecrts14.py
new file mode 100755
index 0000000..2a81399
--- /dev/null
+++ b/ecrts14/ecrts14.py
@@ -0,0 +1,494 @@
+#!/usr/bin/env python
+
+from __future__ import division
+
+import argparse
+import random
+import sys
+import os
+import math
+import time
+
+#import logging, multiprocessing
+
+from collections import defaultdict
+from csv import DictWriter
+from itertools import product
+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
+from schedcat.overheads.model import Overheads, CacheDelay, ConsumerOverheads, ProducerOverheads
+
+import schedcat.model.resources as resources
+import schedcat.generator.tasks as tasks
+import schedcat.mapping.binpack as bp
+
+from schedcat.generator.tasksets import NAMED_UTILIZATIONS
+
+from schedcat.util.storage import storage
+
+from generator import DesignPointGenerator
+from schedcat.stats.stats import proportion_ci
+
+import graph
+import tests
+import topology
+from machines import machines
+
+import gc
+import resource
+
+NAMED_PERIODS_US = {
+    # Named period distributions used in several UNC papers, in microseconds
+    'uni-short'     : tasks.uniform_int( 3*1000,  33*1000),
+    'uni-moderate'  : tasks.uniform_int(10*1000, 100*1000),
+    'uni-long'      : tasks.uniform_int(50*1000, 250*1000),
+}
+
+MIN_SAMPLES = 500
+MAX_SAMPLES = 10000
+#MIN_SAMPLES = 1000
+#MAX_SAMPLES = 10000
+MAX_CI      = 0.05
+CONFIDENCE  = 0.95
+
+#TOTAL_TESTED = 0
+
+def create_pgm_task_set(dp):
+    tg = tasks.TaskGenerator(period = NAMED_PERIODS_US[dp.period],
+                             util   = NAMED_UTILIZATIONS[dp.task_util])
+    ts = tg.make_task_set(max_util = dp.sys_util, squeeze = True)
+    
+    # swap the squeeze task into random position
+    shuf = random.randint(0, len(ts)-1)
+    ts[-1], ts[shuf] = ts[shuf], ts[-1]
+    
+    nrTasks = len(ts)
+    nrGraphs = min(dp.num_graphs(), nrTasks)
+
+    shares = []
+    for i in range(nrGraphs):
+        shares.append(1.0 - random.random())
+    
+    weight = sum(shares)
+    shares = [int((s/weight)*nrTasks + 0.5) for s in shares]
+
+    # we may have gained/lost a node due to rounding
+    # add/remove node from any share with space
+    todrop = sum(shares) - nrTasks
+    if todrop > 0:
+        for i in range(todrop):
+            candidates = [i for i,y in enumerate(shares) if y > 1]
+            shares[random.choice(candidates)] -= 1
+    elif todrop < 0:
+        for i in range(-1*todrop):
+            shares[random.randint(0,len(shares)-1)] += 1
+
+    # make sure that no graph has zero nodes
+    # steal from graphs at random
+    nullGraphs = [i for i,y in enumerate(shares) if y == 0]
+    while nullGraphs:
+        stealGraphs = [i for i,y in enumerate(shares) if y > 1]
+        assert stealGraphs
+        shares[random.choice(stealGraphs)] -= 1
+        shares[nullGraphs[-1]] += 1
+        nullGraphs.pop()
+
+    assert sum(shares) == nrTasks
+
+    subtasksets = []
+    count = 0
+    for i in range(nrGraphs):
+        subts = ts[count:count+shares[i]]
+        assert len(subts) > 0
+        subtasksets.append(subts)
+        count += shares[i]
+
+    graphs = []
+    for subts in subtasksets:
+        graphsz = len(subts)
+        min_depth = max(1, int(dp.depth_factor[0] * graphsz))
+        max_depth = max(1, int(dp.depth_factor[1] * graphsz))
+        gg = graph.GraphGenerator(tasks.uniform_int(graphsz, graphsz),
+                                 tasks.uniform_int(min_depth, max_depth),
+                                 dp.node_placement,
+                                 dp.fan_out, dp.fan_in_cap,
+                                 dp.edge_distance,
+                                 dp.nr_source, dp.nr_sink,
+                                 False, False)
+        g = gg.graph()
+                                 
+        assert len(g.nodes) == graphsz
+        
+        # asign working sets to each edge
+        for e in g.edges:
+            e.wss = dp.wss()
+        
+        # bind tasks/nodes, and compute node wss
+        for i in range(graphsz):
+            g.nodes[i].task = subts[i]
+            subts[i].node = g.nodes[i]
+            subts[i].graph = g
+            # task working set is the sum of inputs
+            subts[i].wss = 0 if len(g.nodes[i].inEdges) == 0 else sum([w.wss for w in g.nodes[i].inEdges])
+
+        # single root
+        root = [r for r in g.nodes if r.isSrc][0]
+
+        # adjust every task to have same execution rate as the source. must
+        # adjust both period and execution time. utilization remains unchanged.
+        for t in subts:
+            if root.task != t:
+                tutil = t.utilization()
+                t.period = root.task.period
+                t.deadline = t.period
+                t.cost = int(t.period * tutil + 0.5)
+
+        graphs.append(g)
+
+    return ts, graphs, subtasksets
+
+def complete(results, n):
+    if n < MIN_SAMPLES:
+        return False
+    elif n > MAX_SAMPLES:
+        return True
+    else:
+        for name, _ in TESTS:
+            if proportion_ci(results[name], n, CONFIDENCE) > MAX_CI:
+                return False
+        return True
+
+def update_mean(old_mean, n, new_sample):
+    return (old_mean*n + new_sample)/(n+1)
+
+def get_consumer_overheads(dp, _system):
+    co_file = 'overheads/%s/consumer/dco_host=%s_lvl=mem_polluters=%s_walk=%s_hpages=%s_upages=%s_type=%s.csv' % (dp.host, dp.host, str(dp.polluters), dp.walk, str(dp.huge_pages), str(dp.uncached), dp.ovh_type)
+    co = ConsumerOverheads.from_file(co_file, non_decreasing=False, system=_system)
+    return co
+
+def get_producer_overheads(dp):
+    po_file = 'overheads/%s/producer/dpo_host=%s_type=%s.csv' % (dp.host, dp.host, dp.ovh_type)
+    po = ProducerOverheads.from_file(po_file, non_decreasing=False)
+    return po
+
+def get_cpmds(dp):
+    cpmd_file = 'overheads/%s/cpmd/dpmo_host=%s_lvl=mem_wcycle=%s_polluters=%s_walk=%s_hpages=%s_upages=%s_type=%s.csv' % (dp.host, dp.host, str(dp.wcycle), str(dp.polluters), dp.walk, str(dp.huge_pages), str(dp.uncached), dp.ovh_type)
+    cpmds = CacheDelay.from_file(cpmd_file, non_decreasing=False)
+    return cpmds
+
+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]
+    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.consumer = get_consumer_overheads(dp, system)
+    ovh.producer = get_producer_overheads(dp)
+    return ovh
+
+def process_dp(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)
+    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)])
+    
+    results = defaultdict(float)
+    avg_latencies = defaultdict(float)
+    avg_ideal_ratios = defaultdict(float)
+    avg_hrt_ratios = defaultdict(float)
+    nsched = defaultdict(int)
+    
+    n = 0
+
+    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)
+        if dp.nr_clusters != 1:
+            for name, 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)
+        # if there is no partitioning, then same results hold for all tests
+        else:
+            result, processed_ts = TESTS[0][1](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)
+
+        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
+
+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")
+    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 __name__ == '__main__':
+    main()