Handle interrupted experiments (by renaming) and failed experiments (cleanup).

2 files changed, 42 insertions, 20 deletions

diff --git a/experiment/experiment.py b/experiment/experiment.py
index 29e6bd7..b21397d 100644
--- a/experiment/experiment.py
+++ b/experiment/experiment.py
@@ -9,26 +9,27 @@ 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 ExperimentInterrupted(ExperimentException):
     """Raised when an experiment appears to be interrupted (partial results)."""
     def __str__(self):
         return "Experiment was interrupted in progress: %d" % self.name
 
-    
+
 class ExperimentFailed(ExperimentException):
     def __str__(self):
         return "Experiment failed during execution: %d" % self.name
 
-    
+
 class Experiment(object):
     """Execute one task-set and save the results. Experiments have unique IDs."""
+    INTERRUPTED_DIR = ".interrupted"
 
     def __init__(self, name, scheduler, working_dir, finished_dir, proc_entries, executables):
         """Run an experiment, optionally wrapped in tracing."""
@@ -45,28 +46,44 @@ class Experiment(object):
 
         self.tracers = []
         if SchedTracer.enabled():
+            self.log("Enabling sched_trace")
             self.tracers.append( SchedTracer(working_dir) )
         if LinuxTracer.enabled():
+            self.log("Enabling trace-cmd / ftrace / kernelshark")
             self.tracers.append( LinuxTracer(working_dir) )
         if LogTracer.enabled():
+            self.log("Enabling logging")
             self.tracers.append( LogTracer(working_dir) )
         if PerfTracer.enabled():
+            self.log("Tracking CPU performance counters")
             self.tracers.append( PerfTracer(working_dir) )
 
         # Overhead trace must be handled seperately, see __run_tasks
         if OverheadTracer.enabled():
+            self.log("Enabling overhead tracing")
             self.overhead_trace = OverheadTracer(working_dir)
         else:
             self.overhead_trace = None
 
     def __make_dirs(self):
+        interrupted = None
+
         if os.path.exists(self.finished_dir):
             raise ExperimentDone(self.name)
+
         if os.path.exists(self.working_dir):
-            raise ExperimentInterrupted(self.name)
-        
+            self.log("Found interrupted experiment, saving in %s" %
+                     Experiment.INTERRUPTED_DIR)
+            interrupted = "%s/%s" % (os.path.split(self.working_dir)[0],
+                                     Experiment.INTERRUPTED_DIR)
+            os.rename(self.working_dir, interrupted)
+
         os.mkdir(self.working_dir)
 
+        if interrupted:
+            os.rename(interrupted, "%s/%s" % (self.working_dir,
+                                              os.path.split(interrupted)[1]))
+
     def __assign_executable_cwds(self):
         def assign_cwd(executable):
             executable.cwd = self.working_dir
@@ -90,13 +107,16 @@ class Experiment(object):
             self.log("Starting overhead trace")
             self.overhead_trace.start_tracing()
 
+        self.log("Releasing %d tasks" % len(self.executables))
         released = litmus_util.release_tasks()
 
         ret = True
         if released != len(self.executables):
+            # Some tasks failed to release, kill all tasks and fail
+            # Need to re-release non-released tasks before we can kill them though
             self.log("Failed to release %d tasks! Re-releasing and killing".format(
                 len(self.experiments) - released))
-            
+
             time.sleep(10)
             litmus_util.release_tasks()
 
@@ -124,8 +144,10 @@ class Experiment(object):
 
     def run_exp(self):
         self.setup()
-        self.__run_tasks()
-        self.teardown()
+        try:
+            self.__run_tasks()
+        finally:
+            self.teardown()
 
     def setup(self):
         self.log("Switching to %s" % self.scheduler)
@@ -133,7 +155,7 @@ class Experiment(object):
 
         self.log("Writing %d proc entries" % len(self.proc_entries))
         map(methodcaller('write_proc'), self.proc_entries)
-        
+
         self.log("Starting %d tracers" % len(self.tracers))
         map(methodcaller('start_tracing'), self.tracers)
 
diff --git a/run_exps.py b/run_exps.py
index c589f51..bc15b98 100755
--- a/run_exps.py
+++ b/run_exps.py
@@ -55,8 +55,8 @@ def convert_data(data):
         else:
             prog = match.group("TYPE") or "rtspin"
             spin = (prog, match.group("ARGS"))
-            spins.append(spin)            
-        
+            spins.append(spin)
+
     return {'proc' : procs, 'spin' : spins}
 
 
@@ -80,7 +80,7 @@ def load_experiment(sched_file, scheduler, duration, param_file, out_base):
         raise IOError("Cannot find schedule file: %s" % sched_file)
 
     dirname = os.path.split(sched_file)[0]
-    
+
     if not scheduler or not duration:
         param_file = param_file or \
           "%s/%s" % (dirname, conf.DEFAULTS['params_file'])
@@ -89,7 +89,7 @@ def load_experiment(sched_file, scheduler, duration, param_file, out_base):
             params = load_params(param_file)
             scheduler = scheduler or params[conf.PARAMS['sched']]
             duration  = duration  or params[conf.PARAMS['dur']]
-            
+
         duration = duration or conf.DEFAULTS['duration']
 
         if not scheduler:
@@ -133,10 +133,10 @@ def run_exp(name, schedule, scheduler, duration, work_dir, out_dir):
     for entry_conf in schedule['proc']:
         path = entry_conf[0]
         data = entry_conf[1]
-        
+
         if not os.path.exists(path):
             raise IOError("Invalid proc path %s: %s" % (path, name))
-        
+
         proc_entries += [ProcEntry(path, data)]
 
     # Parse spinners
@@ -158,14 +158,14 @@ def run_exp(name, schedule, scheduler, duration, work_dir, out_dir):
 
         if not lu.is_executable(real_spin):
             raise OSError("Cannot run spin %s: %s" % (real_spin, name))
-        
+
         executables += [Executable(real_spin, real_args)]
 
     exp = Experiment(name, scheduler, work_dir, out_dir,
                      proc_entries, executables)
     exp.run_exp()
 
-    
+
 def main():
     opts, args = parse_args()
 
@@ -181,12 +181,12 @@ def main():
 
         if not os.path.exists(path):
             raise IOError("Invalid experiment: %s" % path)
-        
+
         if os.path.isdir(exp):
             path = "%s%s" % (path, opts.sched_file)
 
         load_experiment(path, scheduler, duration, param_file, out_base)
 
-        
+
 if __name__ == '__main__':
     main()