Tier.py

Go to the documentation of this file.

class Tier():
    """
    Tiers are used so that the execution of multiple tiers can be 
    processed inside a single loop. Tier-to-tier differences are small 
    and can be configured based on a small number of options.
    The options defined are:
        inspect_only  : This means we only look at the input file and
                        do not try and run an art process. Typically this
                        is used to study an input file, such as that coming
                        off of the detector. It can also be used for 
                        starting the production chain off at a higher tier, 
                        providing an appropriate input file is provided.
        parent_in_sam : If the input file is in SAM we can grab it's meta-
                        data from there.
        input_name    : Input file name.
        fcl           : FHiCL to run.
        output_name   : Output file name. Note this only works if there is 
                        only a single, not TFileService output. Exclusive 
                        with multiple_outputs.
        multiple_outputs : An array of output names for the case where a 
                           tier produces multiple output files. Exclusive 
                           with output name.
        precommand    : Use subprocess to exectute this shell command before 
                        running the tier. This is used for example to 
                        generate simulation generation FHiCLs.
        multiply_events : Multiple the number of events requested for this 
                          stage by this factor. This is used for example for
                          ND rock overlay events where a large number of 
                          these need to be overlaid onto a single beam 
                          event.
    """
    import ProductionTestTools as tools
    from Metric import Metric 
    import ParserDBUsage as db_parser
    
    def __init__(self, **kwargs):
        """
        Constructor
        """
        self.short_name         = kwargs.get("short_name",       "blank_tier")
        self.long_name          = kwargs.get("long_name",        "A blank tier")
        self.inspect_only       = kwargs.get("inspect_only",     False)
        self.parent_in_sam      = kwargs.get("parent_in_sam",    False)
        self.input_name         = kwargs.get("input_name",       False)
        self.fcl                = kwargs.get("fcl",              False)
        self.output_name        = kwargs.get("output_name",      False)
        self.precommand         = kwargs.get("precommand",       False)
        self.multiply_events    = kwargs.get("multiply_events",  1)
        self.minimum_events     = kwargs.get("minimum_events",   1)
        self.multiple_outputs   = kwargs.get("multiple_outputs", False)
        self.universal_fcl_mod  = kwargs.get("universal_fcl_mod",True)
        self.parse_timing_db    = kwargs.get("parse_timing_db",  True)
        #self.verbose            = kwargs.get("verbose",         False)
        self.metric             = self.Metric()
        self.log                = ""
        self.full_fcl           = ""
        self.fcl_path           = ""
    
    def __repr__(self):
        _repr  = "tier : Tier class\n"
        _repr += "tier :     short_name         %s\n"%(repr(self.short_name))
        _repr += "tier :     long_name          %s\n"%(repr(self.long_name))
        _repr += "tier :     inspect_only       %s\n"%(repr(self.inspect_only))
        _repr += "tier :     parent_in_sam      %s\n"%(repr(self.parent_in_sam))
        _repr += "tier :     input_name         %s\n"%(repr(self.input_name))
        _repr += "tier :     fcl                %s\n"%(repr(self.fcl))
        _repr += "tier :     output_name        %s\n"%(repr(self.output_name))
        _repr += "tier :     precommand         %s\n"%(repr(self.precommand))
        _repr += "tier :     multiply_events    %s\n"%(repr(self.multiply_events))
        _repr += "tier :     minimum_events     %s\n"%(repr(self.minimum_events))
        _repr += "tier :     multiple_outputs   %s\n"%(repr(self.multiple_outputs))
        _repr += "tier :     universal fcl mod: %s\n"%(repr(self.universal_fcl_mod))
        _repr += "tier :     parse timing db:   %s\n"%(repr(self.parse_timing_db))
        _repr += "tier :     log lines          %s"  %(len(self.log))
        return _repr

    def setChainName(self, chain_name):
        self.chain_name = chain_name

    def runTier(self, dry_run, n_events, verbose, log_file):
        if "os" not in dir():   import os
        
        # Run pre command if requested
        if self.precommand:
            print "tier :     running precommand: %s"%self.precommand
            if "subprocess" not in dir(): import subprocess
            cmd_list = self.precommand.split()
            try:
                return_code = subprocess.check_call(cmd_list)
            except subprocess.CalledProcessError:
                print "tier : Error, precommand: %s failed"%self.precommand
                self.metric.return_code = 1
                return self.metric
            if verbose: print "tier :     precommand completed with return code: %i"%return_code
        
        # Run universal modify fcl if required
        if self.universal_fcl_mod:
            print "tier :     running universal modify fcl"
            if "subprocess" not in dir(): import subprocess
            mod_fcl_name = "%s_mod.fcl"%self.fcl.split("/")[-1][:-4]
            cmd = "python modifyFHiCL.py -b %s -f %s -a -v"%(self.fcl, mod_fcl_name)
            print "tier :         cmd: %s"%cmd
            self.fcl = mod_fcl_name
            cmd_list = cmd.split()
            try:
                return_code = subprocess.check_call(cmd_list)
            except subprocess.CalledProcessError:
                print "tier : Error, universal modify fcl"
                self.metric.return_code = 1
                return self.metric
            if verbose: print "tier :     universal modify fcl completed with return code: %i"%return_code

        # Parse DB files setup
        if self.parse_timing_db:
            if os.path.exists("timing.db"):
                os.system("rm timing.db")

        # Inspect only is used to study files coming off of the detector
        if self.inspect_only:
            self.metric.output_name = "n/a"
            self.metric.run = True
            try:
                size_string, size_float = self.tools.inspectFile(self.input_name)
                self.metric.output_size = size_float
            except NameError:
                pass
            
            if self.parent_in_sam:
                input_metadata = self.tools.getSAMMetaData(self.input_name.split("/")[-1])
                n_events = input_metadata["Online.TotalEvents"]
                self.metric.input_events = n_events
                
            return self.metric
        
        # Multiple number of events if requested
        if self.multiply_events != 1:
            if verbose: print "tier :     Multiplying the number of events requested %i by %i to %i"%(n_events, self.multiply_events, n_events*self.multiply_events)
            n_events = self.multiply_events*n_events

        # Minimum number of events
        if self.minimum_events != 1:
            if n_events < self.minimum_events:
                if verbose: print "tier :     Minimum number of events %i is greater than requested number of events %i, so using minimum"%(self.minimum_events, n_events)
                n_events = self.minimum_events
        
        # For all other tiers we run an art job
        # Check the input file exists if we need one
        if self.input_name:
            if "*" in self.input_name:
                file_name = self.tools.findWildcardFile(self.input_name)
                if file_name: self.input_name = file_name
            if not (os.path.exists(self.input_name)):
                print "tier :     Input file %s not found, exiting"%self.input_name
                return self.metric
        assert ( self.fcl )
       
        self.prepareAndExecuteProcess(dry_run, n_events, verbose, log_file)
        return self.metric

    def prepareAndExecuteProcess(self, dry_run, n_events, verbose, log_file):
        cmd_list = ["nova"]
        cmd_list.append("-c")
        cmd_list.append(self.fcl)
        self.fcl_path = self.tools.findFHiCL(self.fcl)
        if self.fcl_path != "":
            self.full_fcl = self.tools.fileToString(self.fcl_path)
        if self.output_name:
            cmd_list.append("-o")
            cmd_list.append(self.output_name)
        if n_events:
            cmd_list.append("-n")
            cmd_list.append(str(n_events))
        if self.input_name: cmd_list.append(self.input_name)
        cmd = " ".join(cmd_list)
        print "tier :     cmd: %s"%cmd
        
        if not dry_run:
            print "tier :     running..."
            stdout_redirect = False
            stderr_redirect = False
            if log_file:
                stdout_redirect = open("%s.out"%log_file, "w")
                print "tier :     writing stdout/err to: %s"%stdout_redirect.name
            
            return_code = self.executeProcess(cmd_list, stdout_redirect)
            
            print "tier :     done, got return code: %i"%(return_code)
            return self.metric
        return False

    def executeProcess(self, cmd_list, stdout_redirect):
        if "MemoryMonitor" not in dir(): import MemoryMonitor
        if "subprocess" not in dir():    import subprocess
        if "resource" not in dir():      import resource
        if "time" not in dir():          import time
        if "sys" not in dir():           import sys
        if "os" not in dir():            import os
        # Try and import psutil and enable advanced monitoring
        advanced = False
        if "psutil" not in dir():
            try:
                import psutil
                advanced = True
            except ImportError:
                print "tier : psutils not found, advanced features disabled"
        self.metric.run = True
        if self.input_name and self.input_name[-4:]=="root":
            self.metric.input_events = self.tools.countEventsInRootFile(self.input_name)
        usage_start = resource.getrusage(resource.RUSAGE_CHILDREN)
        # set memory limits for the process
        MEMORY_LIMIT = 2.9*1024*1024*1024
        
        if stdout_redirect:
            stdout = subprocess.PIPE
            stderr = subprocess.STDOUT
        else:
            stdout = None
            stderr = None
        process = subprocess.Popen(cmd_list,  stdout=stdout, stderr=stderr)
        SLICE_IN_SECONDS = 1.
        memory_usage   = []
        if advanced:
            p = psutil.Process(process.pid)
            self.metric.enable_advanced_metrics()
        
        start_time = time.time()
        previous_sample_time = 0.
        lines = []
        sample_times = []
        while True:
            if stdout_redirect: line = process.stdout.readline()
            else:               line = ""
            elapsed_time = time.time()-start_time
            if (line != "") or ((elapsed_time - previous_sample_time) > SLICE_IN_SECONDS):
                memory_usage.append(MemoryMonitor.resident(process.pid))
                sample_times.append(elapsed_time)
                if advanced: self.metric.fill_advanced_metrics(p)
                if line != "":
                    lines.append("TimeMarker: %f\n"%(elapsed_time))
                    lines.append(line)
                previous_sample_time = elapsed_time
                if elapsed_time > 60: SLICE_IN_SECONDS = 10.
                if elapsed_time > 60*60: SLICE_IN_SECONDS = 60.
                
                # limit memory usage
                if memory_usage[-1] > MEMORY_LIMIT:
                    line = "Tier : Subprocess exceeded memory limit of %i (current usage: %i), so was killed\n"%\
                                 (MEMORY_LIMIT,memory_usage[-1])
                    print line
                    sys.stderr.write(line)
                    lines.append(line)
                    process.kill()
            if line == '' and process.poll() != None: break
            # if process.poll() != None: break
        
        if stdout_redirect:
            for line in lines:
                stdout_redirect.write(line)
                self.log += line
            stdout_redirect.close()
        
        return_code = process.returncode
        if return_code != 0:
            print "tier :     Exception caught while executing nova executable, return code: %i"%return_code
        
        self.metric.return_code = return_code
        self.metric.sample_times = sample_times
        usage_end = resource.getrusage(resource.RUSAGE_CHILDREN)
        
        u_cpu_time = usage_end.ru_utime - usage_start.ru_utime
        s_cpu_time = usage_end.ru_stime - usage_start.ru_stime
        max_memory = max(memory_usage)
        
        #memory_usage = convertSize(usage_end.ru_maxrss * resource.getpagesize())
        # By manual inspection of the memory usage in top I believe ru_maxrss is returning a size in kb
        #memory_usage = convertSize(usage_end.ru_maxrss * 1024 )
        #print usage_end.ru_maxrss, resource.getpagesize(), usage_end.ru_maxrss * resource.getpagesize(), memory_usage
        
        self.metric.user_cpu       = u_cpu_time
        self.metric.system_cpu     = s_cpu_time
        self.metric.peak_memory    = max_memory
        self.metric.memory         = memory_usage
        
        if "-n" in cmd_list:
            self.metric.used_events = int(cmd_list[cmd_list.index("-n")+1])
            
        if "-o" in cmd_list:
            if return_code == 0:
                self.metric.output_name = [cmd_list[cmd_list.index("-o")+1]]
                output_size, size_float = self.tools.inspectFile(self.metric.output_name[0])
                self.metric.output_size = [size_float]
                self.metric.output_events = [self.tools.countEventsInRootFile(self.metric.output_name[0])]
                self.metric.file_size_ana = [self.tools.fileSizeAna(self.metric.output_name[0])]
        # Parse DB usage from the output file
        # note this only works if we have pushed the STDOUT to a files
        # if we haven't then we fill a blank metric
        if stdout_redirect:
            self.metric.db_usage = self.db_parser.parseSTDOUT(stdout_redirect.name)
        else:
            self.metric.db_usage = self.db_parser.DBUsage()
        
        if self.multiple_outputs and (return_code == 0):
            self.metric.output_name   = []
            self.metric.output_events = []
            self.metric.output_size   = []
            self.metric.file_size_ana = []

            for output in self.multiple_outputs:
                print "tier : Checking output: %s"%output
                if "*" in output:
                    file_name = self.tools.findWildcardFile(output)
                    if file_name: output = file_name
                try:
                    output_size, size_float = self.tools.inspectFile(output)
                except NameError:
                    print "tier : Output file: %s has not been created"%output
                    return return_code
                self.metric.output_name.append(output)
                self.metric.output_size.append(size_float)
                self.metric.output_events.append(self.tools.countEventsInRootFile(output))
                self.metric.file_size_ana.append(self.tools.fileSizeAna(output))

        # Parse DB files
        if self.parse_timing_db and os.path.exists("timing.db"):
            print "tier :     Parse timing DB"
            import ViewTimingDB as view
            json_name = "timing_db_%s_%s.json"%(self.chain_name,self.short_name)
            cmd = "plot_timetracker -d timing.db -b -q -J %s"%json_name
            print "tier :         cmd: %s"%cmd
            cmd_list = cmd.split()
            try:
                db_return_code = subprocess.check_call(cmd_list)
            except subprocess.CalledProcessError:
                print "tier : Error, parse timing DB failed"
            print "tier :     parse timing DB completed with return code: %i"%db_return_code
            if db_return_code == 0:
                print "tier :         - converting to CSV"
                csv_name  = "timing_db_%s_%s.csv" %(self.chain_name,self.short_name)
                timing_db = view.ViewTimingDB(json_name)
                timing_db.write_csv(csv_name)

        return return_code
        # return 0