1 files changed, 128 insertions, 108 deletions

diff --git a/parse/sched.py b/parse/sched.py
index 652378b..0ab16ce 100644
--- a/parse/sched.py
+++ b/parse/sched.py
@@ -9,53 +9,76 @@ from common import recordtype,log_once
 from point import Measurement
 from ctypes import *
 
+from heapq import *
+
 class TimeTracker:
     '''Store stats for durations of time demarcated by sched_trace records.'''
-    def __init__(self, allow_negative = False):
-        self.begin = self.avg = self.max = self.num = self.next_job = 0
+    def __init__(self, is_valid_duration = lambda x: True, delay_buffer_size = 1, max_pending = 100):
+        self.validator = is_valid_duration
+        self.avg = self.max = self.num = 0
 
-        # Count of times the job in start_time matched that in store_time
         self.matches = 0
-        # And the times it didn't
-        self.disjoints = 0
-
-        self.allow_negative = allow_negative
-
-        # Measurements are recorded in store_ time using the previous matching
-        # record which was passed to store_time. This way, the last record for
-        # any task is always skipped
-        self.last_record = None
-
-    def store_time(self, next_record):
-        '''End duration of time.'''
-        dur = (self.last_record.when - self.begin) if self.last_record else -1
-
-        if self.next_job == next_record.job:
-            self.last_record = next_record
 
-            if self.last_record:
-                self.matches += 1
-
-            if dur > 0 or self.allow_negative:
-                self.max  = max(self.max, dur)
+        self.max_pending = max_pending
+        self.discarded = 0
+
+        self.delay_buffer_size = delay_buffer_size
+        self.start_delay_buffer = []
+        self.end_delay_buffer = []
+        self.start_records = {}
+        self.end_records = {}
+
+    def disjoints(self):
+        unmatched = len(self.start_records) + len(self.end_records)
+        return self.discarded + unmatched
+
+    def process_completed(self):
+        completed = self.start_records.viewkeys() & self.end_records.viewkeys()
+        self.matches += len(completed)
+        for c in completed:
+            s, stime = self.start_records[c]
+            e, etime = self.end_records[c]
+            del self.start_records[c]
+            del self.end_records[c]
+
+            dur = etime - stime
+            if self.validator(dur):
+                self.max = max(self.max, dur)
                 self.avg *= float(self.num / (self.num + 1))
                 self.num += 1
                 self.avg += dur / float(self.num)
 
-                self.begin = 0
-                self.next_job   = 0
-        else:
-            self.disjoints += 1
+        # Give up on some jobs if they've been hanging around too long.
+        # While not strictly needed, it helps improve performance and
+        # it is unlikey to cause too much trouble.
+        if(len(self.start_records) > self.max_pending):
+            to_discard = len(self.start_records) - self.max_pending
+            for i in range(to_discard):
+                # pop off the oldest jobs
+                del self.start_records[self.start_records.iterkeys().next()]
+            self.discarded += to_discard
+        if(len(self.end_records) > self.max_pending):
+            to_discard = len(self.end_records) - self.max_pending
+            for i in range(to_discard):
+                # pop off the oldest jobs
+                del self.end_records[self.end_records.iterkeys().next()]
+            self.discarded += to_discard
+
+    def end_time(self, record, time):
+        '''End duration of time.'''
+        if len(self.end_delay_buffer) == self.delay_buffer_size:
+           to_queue = self.end_delay_buffer.pop(0)
+           self.end_records[to_queue[0].job] = to_queue
+        self.end_delay_buffer.append((record, time))
+        self.process_completed()
 
-    def start_time(self, record, time = None):
+    def start_time(self, record, time):
         '''Start duration of time.'''
-        if self.last_record:
-            if not time:
-                self.begin = self.last_record.when
-            else:
-                self.begin = time
-
-        self.next_job = record.job
+        if len(self.start_delay_buffer) == self.delay_buffer_size:
+            to_queue = self.start_delay_buffer.pop(0)
+            self.start_records[to_queue[0].job] = to_queue
+        self.start_delay_buffer.append((record, time))
+        self.process_completed()
 
 # Data stored for each task
 TaskParams  = namedtuple('TaskParams',  ['wcet', 'period', 'cpu'])
@@ -152,7 +175,12 @@ def make_iterator(fname):
 
 def read_data(task_dict, graph_dict, fnames):
     '''Read records from @fnames and store per-pid stats in @task_dict.'''
-    buff = []
+    q = []
+
+    # Number of trace records to buffer from each stream/file.
+    # A sorted window is maintained in order to deal with
+    # events that were recorded out-of-order.
+    window_size = 500
 
     def get_time(record):
         return record.when if hasattr(record, 'when') else 0
@@ -164,19 +192,18 @@ def read_data(task_dict, graph_dict, fnames):
         except StopIteration:
             return
 
-        i = 0
-        for (i, (brecord, _)) in enumerate(buff):
-            if get_time(brecord) > get_time(arecord):
-                break
-        buff.insert(i, (arecord, itera))
+        sort_key = (get_time(arecord), arecord.job, arecord.pid)
+        heappush(q, (sort_key, arecord, itera))
 
     for fname in fnames:
         itera = make_iterator(fname)
-        add_record(itera)
-
-    while buff:
-        record, itera = buff.pop(0)
+        for w in range(window_size):
+            add_record(itera)
 
+    sys_released = False
+    while q:
+        sort_key, record, itera = heappop(q)
+        # fetch another record
         add_record(itera)
         record.process(task_dict)
         record.process_pgm(task_dict, graph_dict)
@@ -205,7 +232,7 @@ class ParamRecord(SchedRecord):
         task_dict[self.pid].params = params
 
 class ReleaseRecord(SchedRecord):
-    # renames the 'release' field to 'when'
+    # renames the 'release' field to 'when' to enable sorting
     FIELDS = [('when', c_uint64), ('deadline', c_uint64)]
 
     def process(self, task_dict):
@@ -215,18 +242,15 @@ class ReleaseRecord(SchedRecord):
         data.misses.start_time(self, self.deadline)
         data.lateness.start_time(self, self.deadline)
 
-        print '    release %d: r=%d d=%d' % (self.pid, self.when, self.deadline)
-
     def process_pgm(self, task_dict, graph_dict):
         data = task_dict[self.pid]
         data.pgm_response.start_time(self, self.when)
         data.pgm_misses.start_time(self, self.deadline)
         data.pgm_lateness.start_time(self, self.deadline)
 
-        return
         ntype = task_dict[self.pid].pgm_params.node_type
         if ntype == PGM_SRC or ntype == PGM_SRC_SINK:
-            gid = task_dict[self.pid].pgm_params.graph_pid
+            gid = task_dict[self.pid].pgm_params.gid
             gdata = graph_dict[gid]
             gdata.jobs += 1
             gdata.response.start_time(self, self.when)
@@ -236,34 +260,34 @@ class CompletionRecord(SchedRecord):
 
     def process(self, task_dict):
         data = task_dict[self.pid]
-        data.response.store_time(self)
-        data.misses.store_time(self)
-        data.lateness.store_time(self)
+        data.response.end_time(self, self.when)
+        data.misses.end_time(self, self.when)
+        data.lateness.end_time(self, self.when)
 
     def process_pgm(self, task_dict, graph_dict):
         data = task_dict[self.pid]
-        data.pgm_response.store_time(self)
-        data.pgm_misses.store_time(self)
-        data.pgm_lateness.store_time(self)
+        data.pgm_response.end_time(self, self.when)
+        data.pgm_misses.end_time(self, self.when)
+        data.pgm_lateness.end_time(self, self.when)
 
-        return
-        ntype = data.pgm_params.node_type
-        if ntype == PGM_SINK or ntype == PGM_SRC_SINK:
-            gid = data.pgm_params.graph_pid
-            gdata = graph_dict[gid]
-            gdata.response.store_time(self)
+        if data.pgm_params:
+            ntype = data.pgm_params.node_type
+            if ntype == PGM_SINK or ntype == PGM_SRC_SINK:
+                gid = data.pgm_params.gid
+                gdata = graph_dict[gid]
+                gdata.response.end_time(self, self.when)
 
 class BlockRecord(SchedRecord):
     FIELDS = [('when', c_uint64)]
 
     def process(self, task_dict):
-        task_dict[self.pid].blocks.start_time(self)
+        task_dict[self.pid].blocks.start_time(self, self.when)
 
 class ResumeRecord(SchedRecord):
     FIELDS = [('when', c_uint64)]
 
     def process(self, task_dict):
-        task_dict[self.pid].blocks.store_time(self)
+        task_dict[self.pid].blocks.end_time(self, self.when)
 
 class SysReleaseRecord(SchedRecord):
     FIELDS = [('when', c_uint64), ('release', c_uint64)]
@@ -278,12 +302,9 @@ class PgmParamRecord(SchedRecord):
         pass
 
     def process_pgm(self, task_dict, graph_dict):
-
         pgm_params = PgmTaskParams(self.node_type, self.graph_pid)
         task_dict[self.pid].pgm_params = pgm_params
 
-        print '%d: graph id = %d, node type = %d' % (self.pid, self.graph_pid, self.node_type)
-
         if self.node_type == PGM_SRC or self.node_type == PGM_SINK or self.node_type == PGM_SRC_SINK:
             graph_data = graph_dict[self.graph_pid]
             if not graph_data.params:
@@ -313,15 +334,12 @@ class PgmReleaseRecord(SchedRecord):
         data.pgm_misses.start_time(self, self.deadline)
         data.pgm_lateness.start_time(self, self.deadline)
 
-        print 'pgm_release %d: r=%d d=%d' % (self.pid, self.when, self.deadline)
-
-        return
-
-        ntype = data.pgm_params.node_type
-        if ntype == PGM_SRC or ntype == PGM_SRC_SINK:
-            gid = data.pgm_params.graph_pid
-            gdata = graph_dict[gid]
-            gdata.response.start_time(self, self.when)
+        if data.pgm_params:
+            ntype = data.pgm_params.node_type
+            if ntype == PGM_SRC or ntype == PGM_SRC_SINK:
+                gid = data.pgm_params.graph_pid
+                gdata = graph_dict[gid]
+                gdata.response.start_time(self, self.when)
 
 # Map records to sched_trace ids (see include/litmus/sched_trace.h
 register_record(2, ParamRecord)
@@ -339,11 +357,11 @@ def create_trace_dict(data_dir, work_dir = None):
     output_file = "%s/out-st" % work_dir
 
     task_dict = defaultdict(lambda :
-                            TaskData(None, None, 1, TimeTracker(),
-                                     TimeTracker(), TimeTracker(True), TimeTracker(),
-                                     TimeTracker(), TimeTracker(True), TimeTracker()))
+                            TaskData(None, None, 1, TimeTracker(is_valid_duration = lambda x: x > 0),
+                                     TimeTracker(), TimeTracker(), TimeTracker(is_valid_duration = lambda x: x > 0),
+                                     TimeTracker(), TimeTracker(), TimeTracker(is_valid_duration = lambda x: x > 0)))
     graph_dict = defaultdict(lambda:
-                            GraphData(None, 1, TimeTracker()))
+                            GraphData(None, 1, TimeTracker(is_valid_duration = lambda x: x > 0)))
 
     bin_names = [f for f in os.listdir(data_dir) if re.match(bin_files, f)]
     if not len(bin_names):
@@ -375,7 +393,7 @@ def extract_sched_data(result, data_dir, work_dir):
     gstat_data = defaultdict(list)
 
     # Group per-task values
-    for tdata in task_dict.itervalues():
+    for task, tdata in task_dict.iteritems():
         if not tdata.params:
             # Currently unknown where these invalid tasks come from...
             continue
@@ -387,7 +405,7 @@ def extract_sched_data(result, data_dir, work_dir):
         pgm_lateness = tdata.pgm_lateness
         pgm_miss = tdata.pgm_misses
 
-        record_loss = float(miss.disjoints)/(miss.matches + miss.disjoints)
+        record_loss = float(miss.disjoints())/(miss.matches + miss.disjoints())
         stat_data["record-loss"].append(record_loss)
 
         if record_loss > conf.MAX_RECORD_LOSS:
@@ -404,57 +422,59 @@ def extract_sched_data(result, data_dir, work_dir):
         # start with basic task information
         stat_data["miss-ratio" ].append(miss_ratio)
 
-        stat_data["max-response"].append(response.max)
-        stat_data["avg-response"].append(response.avg)
-        stat_data["max-response-prop"].append(response.max / tdata.params.period)
+        stat_data["max-response"].append(float(response.max)/NSEC_PER_MSEC)
+        stat_data["avg-response"].append(response.avg/NSEC_PER_MSEC)
+        stat_data["max-response-prop"].append(float(response.max) / tdata.params.period)
         stat_data["avg-response-prop"].append(response.avg / tdata.params.period)
 
-        stat_data["max-tard"].append(miss.max)
-        stat_data["avg-tard"].append(avg_tard)
-        stat_data["max-tard-prop"].append(miss.max / tdata.params.period)
+        stat_data["max-tard"].append(float(miss.max)/NSEC_PER_MSEC)
+        stat_data["avg-tard"].append(avg_tard/NSEC_PER_MSEC)
+        stat_data["max-tard-prop"].append(float(miss.max) / tdata.params.period)
         stat_data["avg-tard-prop"].append(avg_tard / tdata.params.period)
 
-        stat_data["max-response"].append(lateness.max)
-        stat_data["avg-response"].append(lateness.avg)
-        stat_data["max-response-prop"].append(lateness.max / tdata.params.period)
+        stat_data["max-response"].append(float(lateness.max)/NSEC_PER_MSEC)
+        stat_data["avg-response"].append(lateness.avg/NSEC_PER_MSEC)
+        stat_data["max-response-prop"].append(float(lateness.max) / tdata.params.period)
         stat_data["avg-response-prop"].append(lateness.avg / tdata.params.period)
 
         # same data, but with PGM-adjusted release times (shifted deadlines)
         stat_data["pgm-miss-ratio" ].append(pgm_miss_ratio)
 
-        stat_data["pgm-max-response"].append(pgm_response.max)
-        stat_data["pgm-avg-response"].append(pgm_response.avg)
-        stat_data["pgm-max-response-prop"].append(pgm_response.max / tdata.params.period)
+        stat_data["pgm-max-response"].append(float(pgm_response.max)/NSEC_PER_MSEC)
+        stat_data["pgm-avg-response"].append(pgm_response.avg/NSEC_PER_MSEC)
+        stat_data["pgm-max-response-prop"].append(float(pgm_response.max) / tdata.params.period)
         stat_data["pgm-avg-response-prop"].append(pgm_response.avg / tdata.params.period)
 
-        stat_data["pgm-max-tard"].append(pgm_miss.max)
-        stat_data["pgm-avg-tard"].append(pgm_avg_tard)
-        stat_data["pgm-max-tard-prop"].append(pgm_miss.max / tdata.params.period)
+        stat_data["pgm-max-tard"].append(float(pgm_miss.max)/NSEC_PER_MSEC)
+        stat_data["pgm-avg-tard"].append(pgm_avg_tard/NSEC_PER_MSEC)
+        stat_data["pgm-max-tard-prop"].append(float(pgm_miss.max) / tdata.params.period)
         stat_data["pgm-avg-tard-prop"].append(pgm_avg_tard / tdata.params.period)
 
-        stat_data["pgm-max-response"].append(pgm_lateness.max)
-        stat_data["pgm-avg-response"].append(pgm_lateness.avg)
-        stat_data["pgm-max-response-prop"].append(pgm_lateness.max / tdata.params.period)
+        stat_data["pgm-max-response"].append(float(pgm_lateness.max)/NSEC_PER_MSEC)
+        stat_data["pgm-avg-response"].append(pgm_lateness.avg/NSEC_PER_MSEC)
+        stat_data["pgm-max-response-prop"].append(float(pgm_lateness.max) / tdata.params.period)
         stat_data["pgm-avg-response-prop"].append(pgm_lateness.avg / tdata.params.period)
 
-    for gdata in graph_dict.itervalues():
+    for gid, gdata in graph_dict.iteritems():
         if not gdata.params:
             continue
-
         response = gdata.response
-        record_loss = float(response.disjoints)/(response.matches + response.disjoints)
+        if response.matches + response.disjoints() == 0:
+            record_loss = 0
+        else:
+            record_loss = float(response.disjoints())/(response.matches + response.disjoints())
         gstat_data["graph-record-loss"].append(record_loss)
 
-        if record_los > conf.MAX_RECORD_LOSS:
+        if record_loss > conf.MAX_RECORD_LOSS:
             log_once(LOSS_MSG)
             continue
 
-        gstat_data["graph-max-response"].append(response.max)
-        gstat_data["graph-avg-response"].append(response.avg)
+        gstat_data["graph-max-response"].append(float(response.max)/NSEC_PER_MSEC)
+        gstat_data["graph-avg-response"].append(response.avg/NSEC_PER_MSEC)
 
     # Summarize value groups
     for name, data in stat_data.iteritems():
-        if not data or not sum(data):
+        if not data:
             log_once(SKIP_MSG, SKIP_MSG % name)
             continue
         result[name] = Measurement(str(name)).from_array(data)