Inheritance diagram for Cluster:

Collaboration diagram for Cluster:

Public Member Functions
	__init__ (self)
	The default constructor.

	adjust_path (self, str path)
	This method can be used if path names are different on submission and execution hosts.

	available (self)
	The cluster should always be available to accept new jobs.

	execute (self, Script job, options="", dry=False, tag="current")
	Takes a Script object and a string with options and runs it on the cluster, either with ROOT or with basf2, depending on the file type.

	is_job_finished (self, Script job)
	Checks whether the '.done'-file has been created for a job.

str	createDoneFileName (self, Script job)
	Generate the file name used for the done output.

	prepareSubmission (self, Script job, options, tag)
	Setup output folders and create the wrapping shell script.

	checkDoneFile (self, job)
	Checks whether the '.done'-file has been created for a job.

	terminate (self, Script job)
	Terminate running job.

Static Public Member Functions
	is_supported ()

	name ()

	description ()

Public Attributes
tuple	native_spec
	The command to submit a job.

int	requirement_vmem = 4
	required vmem by the job in GB, required on DESY NAF, otherwise jobs get killed due to memory consumption

int	requirement_storage = 50
	the storage IO in GB which can be performed by each job.

str	queuename = "short.q"
	Queue best suitable for execution at DESY NAF.

	path = os.getcwd()
	The default constructor.

	logger = logging.getLogger("validate_basf2")
	Contains a reference to the logger-object from validate_basf2 Set up the logging functionality for the 'cluster execution'-Class, so we can log to validate_basf2.py's log what is going on in .execute and .is_finished.

	tools = self.adjust_path(os.environ["BELLE2_TOOLS"])
	Path to the basf2 tools and central/local release.

str	b2setup = "b2setup"
	The command for b2setup (and b2code-option)

	clusterlog = open(clusterlog_dir + "clusterlog.log", "w+")
	The file object to which all cluster messages will be written.

Detailed Description

A class that provides the controls for running jobs on a (remote)
Sun Grid Engine cluster. It provides two methods:
- is_job_finished(job): Returns True or False, depending on whether the job
    has finished execution
- execute(job): Takes a job and executes it by sending it to the cluster

Definition at line 19 of file clustercontroldrmaa.py.

Constructor & Destructor Documentation

◆ init()

__init__ ( self )

The default constructor.

Holds the current working directory, which is also the location of the shellscripts that are being sent to the cluster.
Initializes a logger which writes to validate_basf2.py's log.
Finds the revision of basf2 that will be set up on the cluster.

Definition at line 62 of file clustercontroldrmaa.py.

    def __init__(self):
        """!
        The default constructor.
        - Holds the current working directory, which is also the location of
          the shellscripts that are being sent to the cluster.
        - Initializes a logger which writes to validate_basf2.py's log.
        - Finds the revision of basf2 that will be set up on the cluster.
        """
 
        
        self.native_spec = (
            "-l h_vmem={requirement_vmem}G,h_fsize={"
            "requirement_storage}G "
            "-q {queuename} -V"
        )
 
        
        self.requirement_vmem = 4
 
        
        self.requirement_storage = 50
 
        
        self.queuename = "short.q"
 
        # call the base constructor, which will setup the batch cluster
        # common stuff
        super().__init__()
 

Member Function Documentation

◆ adjust_path()

adjust_path	(		self,
		str	path )

This method can be used if path names are different on submission and execution hosts.

Parameters

path	The past that needs to be adjusted

Returns: : The adjusted path

Reimplemented from ClusterBase.

Definition at line 96 of file clustercontroldrmaa.py.

    def adjust_path(self, path: str):
        """!
        This method can be used if path names are different on submission
        and execution hosts.
        @param path: The past that needs to be adjusted
        @return: The adjusted path
        """
 
        return path
 

◆ available()

available ( self )

The cluster should always be available to accept new jobs.

Returns: : Will always return True if the function can be called

Definition at line 107 of file clustercontroldrmaa.py.

    def available(self):
        """!
        The cluster should always be available to accept new jobs.
        @return: Will always return True if the function can be called
        """
 
        return True
 

◆ checkDoneFile()

checkDoneFile	(		self,
			job )

inherited

Checks whether the '.done'-file has been created for a job.

If so, it returns True, else it returns False in the first part of the tuple. Also deletes the .done-File it if exists. The second entry in the tuple will be the exit code read from the done file

Definition at line 139 of file clustercontrolbase.py.

    def checkDoneFile(self, job):
        """!
        Checks whether the '.done'-file has been created for a job. If so, it
        returns True, else it returns False in the first part of the tuple.
        Also deletes the .done-File it if exists.
        The second entry in the tuple will be the exit code read from the done file
        """
 
        # If there is a file indicating the job is done, that is its name:
        donefile_path = self.createDoneFileName(job)
 
        donefile_exists = False
        # Check if such a file exists. If so, this means that the job has
        # finished.
        if os.path.isfile(donefile_path):
 
            # Read the returncode/exit_status for the job from the *.done-file
            with open(donefile_path) as f:
                try:
                    returncode = int(f.read().strip())
                except ValueError:
                    returncode = -654
 
            print(f"donefile found with return code {returncode}")
            donefile_exists = True
            os.remove(donefile_path)
        else:
            print("no donefile found")
            returncode = -555
 
        return [donefile_exists, returncode]
 

◆ createDoneFileName()

str createDoneFileName	(		self,
		Script	job )

inherited

Generate the file name used for the done output.

Definition at line 76 of file clustercontrolbase.py.

    def createDoneFileName(self, job: Script) -> str:
        """!
        Generate the file name used for the done output
        """
        return f"{self.path}/script_{job.name}.done"
 

◆ description()

description ( )

static

Returns description of this job control

Definition at line 56 of file clustercontroldrmaa.py.

    def description():
        """
        Returns description of this job control
        """
        return "Batch submission via the drmaa interface to Grid Engine"
 

◆ execute()

execute	(		self,
		Script	job,
			options = "",
			dry = False,
			tag = "current" )

Takes a Script object and a string with options and runs it on the cluster, either with ROOT or with basf2, depending on the file type.

Parameters

job	The steering file object that should be executed
options	Options that will be given to the basf2 command
dry	Whether to perform a dry run or not
tag	The folder within the results directory

Returns: : None

Definition at line 115 of file clustercontroldrmaa.py.

    def execute(self, job: Script, options="", dry=False, tag="current"):
        """!
        Takes a Script object and a string with options and runs it on the
        cluster, either with ROOT or with basf2, depending on the file type.
 
        @param job: The steering file object that should be executed
        @param options: Options that will be given to the basf2 command
        @param dry: Whether to perform a dry run or not
        @param tag: The folder within the results directory
        @return: None
        """
 
        # import here first so the whole module can also be imported on python
        # installations which have no drmaa at all
        import drmaa
 
        print(str(drmaa.Session()))
 
        with drmaa.Session() as session:
            print("got session ")
            print(str(session))
 
            shell_script_name = self.prepareSubmission(job, options, tag)
 
            # native specification with all the good settings for the batch
            # server
            native_spec_string = self.native_spec.format(
                requirement_storage=self.requirement_storage,
                requirement_vmem=self.requirement_vmem,
                queuename=self.queuename,
            )
            print(
                f"Creating job template for wrapper script {shell_script_name}"
            )
            jt = session.createJobTemplate()
            jt.remoteCommand = shell_script_name
            jt.joinFiles = True
            jt.nativeSpecification = native_spec_string
 
            if not dry:
                jobid = session.runJob(jt)
                self.logger.debug(
                    f"Script {job.name} started with job id {jobid}"
                )
                job.job_id = jobid
 
            session.deleteJobTemplate(jt)
        return
 

◆ is_job_finished()

is_job_finished	(		self,
		Script	job )

Checks whether the '.done'-file has been created for a job.

If so, it returns True, else it returns False. Also deletes the .done-File once it has returned True.

Parameters

job	The job of which we want to know if it finished

Returns: : (True if the job has finished, exit code). If we can't find the exit code in the '.done'-file, the returncode will be -654. If the job is not finished, the exit code is returned as 0.

Definition at line 164 of file clustercontroldrmaa.py.

    def is_job_finished(self, job: Script):
        """!
        Checks whether the '.done'-file has been created for a job. If so, it
        returns True, else it returns False.
        Also deletes the .done-File once it has returned True.
 
        @param job: The job of which we want to know if it finished
        @return: (True if the job has finished, exit code). If we can't find the
            exit code in the '.done'-file, the returncode will be -654.
            If the job is not finished, the exit code is returned as 0.
        """
 
        # import here first so the whole module can also be imported on python
        # installations which have no drmaa at all
        import drmaa
 
        if job.job_id is None:
            print(
                "Job has not been started with cluster drmaaa because "
                "job id is missing"
            )
            sys.exit(0)
 
        with drmaa.Session() as session:
 
            # some batch server will forget completed jobs right away
            try:
                status = session.jobStatus(job.job_id)
            except drmaa.errors.InvalidJobException:
                print(
                    f"Job info for jobid {job.job_id} cannot be retrieved, assuming job has terminated"
                )
 
                (donefile_exists, donefile_returncode) = self.checkDoneFile(job)
 
                # always return the job es complete even if there is no done
                #  file at this point tho job is also not longer
                # running/queued on the cluster
                return [True, donefile_returncode]
 
            # Return that the job is finished + the return code for it
            # depending when we look for the job this might never be used,
            # because the jobs disappear from qstat before we can query them
            #  ..
            if status == drmaa.JobState.DONE:
                # todo: return code
                return [True, 0]
            if status == drmaa.JobState.FAILED:
                return [True, 1]
 
            return [False, 0]

◆ is_supported()

is_supported ( )

static

Check if qsub is available

Definition at line 29 of file clustercontroldrmaa.py.

    def is_supported():
        """
        Check if qsub is available
        """
        try:
            import drmaa  # noqa
 
            return True
        except ImportError:
            print(
                "drmaa library is not installed, please use 'pip3 install "
                "drmaa'"
            )
            return False
        except RuntimeError as re:
            print("drmaa library not properly configured")
            print(str(re))
            return False
 

◆ name()

name ( )

static

Returns name of this job control

Definition at line 49 of file clustercontroldrmaa.py.

    def name():
        """
        Returns name of this job control
        """
        return "cluster-drmaa"
 

◆ prepareSubmission()

prepareSubmission	(		self,
		Script	job,
			options,
			tag )

inherited

Setup output folders and create the wrapping shell script.

Will return the full file name of the generated wrapper script.

Definition at line 82 of file clustercontrolbase.py.

    def prepareSubmission(self, job: Script, options, tag):
        """!
        Setup output folders and create the wrapping shell script. Will return
        the full file name of the generated wrapper script.
        """
 
        # Define the folder in which the results (= the ROOT files) should be
        # created. This is where the files containing plots will end up. By
        # convention, data files will be stored in the parent dir.
        # Then make sure the folder exists (create if it does not exist) and
        # change to cwd to this folder.
        output_dir = os.path.abspath(f"./results/{tag}/{job.package}")
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
 
        # Path where log file is supposed to be created
        # log_file = output_dir + '/' + os.path.basename(job.path) + '.log'
 
        # Remove any left over done files
        donefile_path = self.createDoneFileName(job)
        if os.path.isfile(donefile_path):
            os.remove(donefile_path)
 
        # Now we need to distinguish between .py and .C files:
        extension = os.path.splitext(job.path)[1]
        if extension == ".C":
            # .c files are executed with root
            command = "root -b -q " + job.path
        else:
            # .py files are executed with basf2
            # 'options' contains an option-string for basf2, e.g. '-n 100'
            command = f"basf2 {job.path} {options}"
 
        # Create a helpfile-shellscript, which contains all the commands that
        # need to be executed by the cluster.
        # First, set up the basf2 tools and perform b2setup with the correct
        # revision. The execute the command (i.e. run basf2 or ROOT on a
        # steering file). Write the return code of that into a *.done file.
        # Delete the helpfile-shellscript.
        tmp_name = self.path + "/" + "script_" + job.name + ".sh"
        with open(tmp_name, "w+") as tmp_file:
            tmp_file.write(
                "#!/bin/bash \n\n"
                + "BELLE2_NO_TOOLS_CHECK=1 \n"
                + f"source {self.tools}/b2setup \n"
                + f"cd {self.adjust_path(output_dir)} \n"
                + f"{command} \n"
                + f"echo $? > {self.path}/script_{job.name}.done \n"
                + f"rm {tmp_name} \n"
            )
 
        # Make the helpfile-shellscript executable
        st = os.stat(tmp_name)
        os.chmod(tmp_name, st.st_mode | stat.S_IEXEC)
 
        return tmp_name
 

◆ terminate()

terminate	(		self,
		Script	job )

inherited

Terminate running job.

Definition at line 171 of file clustercontrolbase.py.

    def terminate(self, job: Script):
        """! Terminate running job.
        """
        self.logger.error("Script termination not supported.")
 

Member Data Documentation

◆ b2setup

str b2setup = "b2setup"

inherited

The command for b2setup (and b2code-option)

Definition at line 51 of file clustercontrolbase.py.

◆ clusterlog

clusterlog = open(clusterlog_dir + "clusterlog.log", "w+")

inherited

The file object to which all cluster messages will be written.

Definition at line 74 of file clustercontrolbase.py.

◆ logger

logger = logging.getLogger("validate_basf2")

inherited

Contains a reference to the logger-object from validate_basf2 Set up the logging functionality for the 'cluster execution'-Class, so we can log to validate_basf2.py's log what is going on in .execute and .is_finished.

Definition at line 39 of file clustercontrolbase.py.

◆ native_spec

tuple native_spec

Initial value:

=  (
            "-l h_vmem={requirement_vmem}G,h_fsize={"
            "requirement_storage}G "
            "-q {queuename} -V"
        )

The command to submit a job.

'LOGFILE' will be replaced by the actual log file name

Definition at line 73 of file clustercontroldrmaa.py.

◆ path

path = os.getcwd()

inherited

The default constructor.

The path, where the help files are being created Maybe there should be a special subfolder for them?

Definition at line 33 of file clustercontrolbase.py.

◆ queuename

str queuename = "short.q"

Queue best suitable for execution at DESY NAF.

Definition at line 89 of file clustercontroldrmaa.py.

◆ requirement_storage

int requirement_storage = 50

the storage IO in GB which can be performed by each job.

By default, this is 3GB at DESY which is to small for some validation scripts

Definition at line 86 of file clustercontroldrmaa.py.

◆ requirement_vmem

int requirement_vmem = 4

required vmem by the job in GB, required on DESY NAF, otherwise jobs get killed due to memory consumption

Definition at line 81 of file clustercontroldrmaa.py.

◆ tools

tools = self.adjust_path(os.environ["BELLE2_TOOLS"])

inherited

Path to the basf2 tools and central/local release.

Definition at line 46 of file clustercontrolbase.py.

The documentation for this class was generated from the following file:

validation/scripts/clustercontroldrmaa.py

Public Member Functions

Static Public Member Functions

Public Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ adjust_path()

◆ available()

◆ checkDoneFile()

◆ createDoneFileName()

◆ description()

◆ execute()

◆ is_job_finished()

◆ is_supported()

◆ name()

◆ prepareSubmission()

◆ terminate()

Member Data Documentation

◆ b2setup

◆ clusterlog

◆ logger

◆ native_spec

◆ path

◆ queuename

◆ requirement_storage

◆ requirement_vmem

◆ tools

◆ init()