Improved robustness of run_exps.py execution.

Thanks to bcw and gelliott for debugging and ideas.

* Print out experiment number and total experiments when starting experiments.
* Only sleep and re-release tasks if tasks are waiting to release.
* Fail experiment with verbose messages if any tasks fail before becoming ready
to release.
* When waiting for tasks to become ready for release, reset the waiting time
whenever a new task (or task(s)) become ready.
* Start regular tracers BEFORE the plugin switch to log data from the switch.
* Check the number of running tasks AFTER trying to switch the linux scheduler.
This gives plugin deactivate code the opportunity to kill these tasks.
* If an invalid executable is specified in the schedule file, fail before
attempting to run the experiment and print out the problem.
* Propogate exceptions up from experiment failures instead of creating
ExperimentFailed exceptions.

This commit also made clock-frequency automatically ignored by parse_exps.py.
The value of this would change by +- a Mhz between experiments, ruining graphs.

6 files changed, 216 insertions, 161 deletions

diff --git a/config/config.py b/config/config.py
index 1ac468b..b631aa2 100644
--- a/config/config.py
+++ b/config/config.py
@@ -9,7 +9,7 @@ BINS = {'rtspin'    : get_executable_hint('rtspin', 'liblitmus'),
         'ftsplit'   : get_executable_hint('ft2csv', 'feather-trace-tools'),
         'ftsort'    : get_executable_hint('ftsort', 'feather-trace-tools'),
         'st_trace'  : get_executable_hint('st_trace', 'feather-trace-tools'),
-        # Option, as not everyone uses kernelshark yet
+        # Optional, as not everyone uses kernelshark yet
         'trace-cmd' : get_executable_hint('trace-cmd', 'rt-kernelshark', True),
         # Optional, as sched_trace is not a publically supported repository
         'st_show'   : get_executable_hint('st_show', 'sched_trace', True)}
diff --git a/parse_exps.py b/parse_exps.py
index d07378c..c2cbedb 100755
--- a/parse_exps.py
+++ b/parse_exps.py
@@ -140,8 +140,8 @@ def main():
     if opts.ignore:
         for param in opts.ignore.split(","):
             builder.try_remove(param)
-    # Always average multiple trials
-    builder.try_remove(PARAMS['trial'])
+    builder.try_remove(PARAMS['trial'])  # Always average multiple trials
+    builder.try_remove(PARAMS['cycles']) # Only need for feather-trace parsing
 
     col_map = builder.build()
     result_table = TupleTable(col_map)
diff --git a/run/executable/executable.py b/run/executable/executable.py
index e6f2003..a2426f1 100644
--- a/run/executable/executable.py
+++ b/run/executable/executable.py
@@ -59,6 +59,9 @@ class Executable(object):
     def interrupt(self):
         self.sp.send_signal(signal.SIGINT)
 
+    def poll(self):
+        return self.sp.poll()
+
     def terminate(self):
         '''Send the terminate signal to the binary.'''
         self.sp.terminate()
diff --git a/run/experiment.py b/run/experiment.py
index ff0e9f3..b0e46b6 100644
--- a/run/experiment.py
+++ b/run/experiment.py
@@ -1,9 +1,7 @@
-import common as com
 import os
 import time
 import run.litmus_util as lu
 import shutil as sh
-import traceback
 from operator import methodcaller
 
 class ExperimentException(Exception):
@@ -11,16 +9,11 @@ class ExperimentException(Exception):
     def __init__(self, name):
         self.name = name
 
-
 class ExperimentDone(ExperimentException):
     '''Raised when an experiment looks like it's been run already.'''
     def __str__(self):
         return "Experiment finished already: %d" % self.name
 
-class ExperimentFailed(ExperimentException):
-    def __str__(self):
-        return "Experiment failed during execution: %d" % self.name
-
 class SystemCorrupted(Exception):
     pass
 
@@ -31,7 +24,6 @@ class Experiment(object):
     def __init__(self, name, scheduler, working_dir, finished_dir,
                  proc_entries, executables, tracer_types):
         '''Run an experiment, optionally wrapped in tracing.'''
-
         self.name = name
         self.scheduler = scheduler
         self.working_dir  = working_dir
@@ -40,16 +32,10 @@ class Experiment(object):
         self.executables  = executables
         self.exec_out = None
         self.exec_err = None
+        self.tracer_types = tracer_types
 
-        self.task_batch = com.num_cpus()
-
-        self.__make_dirs()
-        self.__assign_executable_cwds()
-        self.__setup_tracers(tracer_types)
-
-
-    def __setup_tracers(self, tracer_types):
-        tracers = [ t(self.working_dir) for t in tracer_types ]
+    def __setup_tracers(self):
+        tracers = [ t(self.working_dir) for t in self.tracer_types ]
 
         self.regular_tracers  = [t for t in tracers if not t.is_exact()]
         self.exact_tracers = [t for t in tracers if t.is_exact()]
@@ -84,13 +70,11 @@ class Experiment(object):
         map(assign_cwd, self.executables)
 
     def __kill_all(self):
-        # Give time for whatever failed to finish failing
-        time.sleep(2)
-
-        released = lu.release_tasks()
-        self.log("Re-released %d tasks" % released)
+        if lu.waiting_tasks():
+            released = lu.release_tasks()
+            self.log("Re-released %d tasks" % released)
 
-        time.sleep(5)
+            time.sleep(1)
 
         self.log("Killing all tasks")
         for e in self.executables:
@@ -99,7 +83,62 @@ class Experiment(object):
             except:
                 pass
 
-        time.sleep(2)
+        time.sleep(1)
+
+    def __strip_path(self, path):
+        '''Shorten path to something more readable.'''
+        file_dir = os.path.split(self.working_dir)[0]
+        if path.index(file_dir) == 0:
+            path = path[len(file_dir):]
+
+        return path.strip("/")
+
+    def __check_tasks_status(self):
+        '''Raises an exception if any tasks have failed.'''
+        msgs = []
+
+        for e in self.executables:
+            status = e.poll()
+            if status != None and status:
+                err_msg = "Task %s failed with status: %s" % (e.wait(), status)
+                msgs += [err_msg]
+
+        if msgs:
+            # Show at most 3 messages so that every task failing doesn't blow
+            # up the terminal
+            if len(msgs) > 3:
+                num_errs = len(msgs) - 3
+                msgs = msgs[0:4] + ["...%d more task errors..." % num_errs]
+
+            out_name = self.__strip_path(self.exec_out.name)
+            err_name = self.__strip_path(self.exec_err.name)
+            help = "Check dmesg, %s, and %s" % (out_name, err_name)
+
+            raise Exception("\n".join(msgs + [help]))
+
+    def __wait_for_ready(self):
+        self.log("Sleeping until tasks are ready for release...")
+
+        wait_start = time.time()
+        num_ready  = lu.waiting_tasks()
+
+        while num_ready < len(self.executables):
+            # Quit if too much time passes without a task becoming ready
+            if time.time() - wait_start > 180.0:
+                s = "waiting: %d, submitted: %d" %\
+                  (lu.waiting_tasks(), len(self.executables))
+                raise Exception("Too much time spent waiting for tasks! %s" % s)
+
+            time.sleep(1)
+
+            # Quit if any tasks fail
+            self.__check_tasks_status()
+
+            # Reset the waiting time whenever more tasks become ready
+            now_ready = lu.waiting_tasks()
+            if now_ready != num_ready:
+                wait_start = time.time()
+                num_ready  = lu.now_ready
 
     def __run_tasks(self):
         self.log("Starting %d tasks" % len(self.executables))
@@ -108,16 +147,9 @@ class Experiment(object):
             try:
                 e.execute()