Local Class Reference

Inheritance diagram for Local:

Classes
class	LocalResult

Public Member Functions
	__init__ (self, *, backend_args=None, max_processes=1)
	join (self)
	max_processes (self)
	max_processes (self, value)
	submit (self, job)
	get_submit_script_path (self, job)

Static Public Member Functions
	run_job (name, working_dir, output_dir, script)

Public Attributes
	pool = None
	The actual Pool object of this instance of the Backend.
	max_processes = max_processes
	The size of the multiprocessing process pool.
dict	backend_args = {**self.default_backend_args, **backend_args}
	The backend args that will be applied to jobs unless the job specifies them itself.

Static Public Attributes
str	submit_script = "submit.sh"
	Default submission script name.
str	exit_code_file = "__BACKEND_CMD_EXIT_STATUS__"
	Default exit code file name.
dict	default_backend_args = {}
	Default backend_args.

Protected Member Functions
	_ (self, job)
	_ (self, job)
	_ (self, jobs)
	_create_parent_job_result (cls, parent)
	_add_wrapper_script_setup (self, job, batch_file)
	_add_wrapper_script_teardown (self, job, batch_file)

Static Protected Member Functions
	_add_setup (job, batch_file)

Protected Attributes
	_max_processes = value
	Internal attribute of max_processes.

Detailed Description

Backend for local processes i.e. on the same machine but in a subprocess.

Note that you should call the self.join() method to close the pool and wait for any
running processes to finish before exiting the process. Once you've called join you will have to set up a new
instance of this backend to create a new pool. If you don't call `Local.join` or don't create a join yourself
somewhere, then the main python process might end before your pool is done.

Keyword Arguments:
    max_processes (int): Integer that specifies the size of the process pool that spawns the subjobs, default=1.
        It's the maximum simultaneous subjobs.
        Try not to specify a large number or a number larger than the number of cores.
        It won't crash the program but it will slow down and negatively impact performance.

Definition at line 924 of file backends.py.

Constructor & Destructor Documentation

__init__()

__init__	(		self,
		*	,
			backend_args = None,
			max_processes = 1 )

Definition at line 940 of file backends.py.

    def __init__(self, *, backend_args=None, max_processes=1):
        """
        """
        super().__init__(backend_args=backend_args)
        
        self.pool = None
        
        self.max_processes = max_processes
 

Member Function Documentation

_() [1/3]

_ ( self, job )

protected

Submission of a `SubJob` for the Local backend

Definition at line 1024 of file backends.py.

    def _(self, job):
        """
        Submission of a `SubJob` for the Local backend
        """
        # Make sure the output directory of the job is created
        job.output_dir.mkdir(parents=True, exist_ok=True)
        # Make sure the working directory of the job is created
        job.working_dir.mkdir(parents=True, exist_ok=True)
        job.copy_input_sandbox_files_to_working_dir()
        job.dump_input_data()
        # Get the path to the bash script we run
        script_path = self.get_submit_script_path(job)
        with open(script_path, mode="w") as batch_file:
            print("#!/bin/bash", file=batch_file)
            self._add_wrapper_script_setup(job, batch_file)
            self._add_setup(job, batch_file)
            print(job.full_command, file=batch_file)
            self._add_wrapper_script_teardown(job, batch_file)
        B2INFO(f"Submitting {job}")
        job.result = Local.LocalResult(job,
                                       self.pool.apply_async(self.run_job,
                                                             (job.name,
                                                              job.working_dir,
                                                              job.output_dir,
                                                              script_path)
                                                             )
                                       )
        job.status = "submitted"
        B2INFO(f"{job} submitted")
 

_() [2/3]

_ ( self, job )

protected

Submission of a `Job` for the Local backend

Definition at line 1055 of file backends.py.

    def _(self, job):
        """
        Submission of a `Job` for the Local backend
        """
        # Make sure the output directory of the job is created
        job.output_dir.mkdir(parents=True, exist_ok=True)
        # Make sure the working directory of the job is created
        job.working_dir.mkdir(parents=True, exist_ok=True)
        # Check if we have any valid input files
        job.check_input_data_files()
 
        if not job.splitter:
            # Get all of the requested files for the input sandbox and copy them to the working directory
            job.copy_input_sandbox_files_to_working_dir()
            job.dump_input_data()
            # Get the path to the bash script we run
            script_path = self.get_submit_script_path(job)
            with open(script_path, mode="w") as batch_file:
                print("#!/bin/bash", file=batch_file)
                self._add_wrapper_script_setup(job, batch_file)
                self._add_setup(job, batch_file)
                print(job.full_command, file=batch_file)
                self._add_wrapper_script_teardown(job, batch_file)
            B2INFO(f"Submitting {job}")
            job.result = Local.LocalResult(job,
                                           self.pool.apply_async(self.run_job,
                                                                 (job.name,
                                                                  job.working_dir,
                                                                  job.output_dir,
                                                                  script_path)
                                                                 )
                                           )
            B2INFO(f"{job} submitted")
        else:
            # Create subjobs according to the splitter's logic
            job.splitter.create_subjobs(job)
            # Submit the subjobs
            self.submit(list(job.subjobs.values()))
            # After submitting subjobs, make a Job.result for the parent Job object, used to call ready() on
            self._create_parent_job_result(job)
 

_() [3/3]

_ ( self, jobs )

protected

Submit method of Local() that takes a list of jobs instead of just one and submits each one.

Definition at line 1097 of file backends.py.

    def _(self, jobs):
        """
        Submit method of Local() that takes a list of jobs instead of just one and submits each one.
        """
        # Submit the jobs
        for job in jobs:
            self.submit(job)
        B2INFO("All requested jobs submitted.")
 

_add_setup()

_add_setup ( job, batch_file )

staticprotectedinherited

Adds setup lines to the shell script file.

Definition at line 807 of file backends.py.

    def _add_setup(job, batch_file):
        """
        Adds setup lines to the shell script file.
        """
        for line in job.setup_cmds:
            print(line, file=batch_file)
 

_add_wrapper_script_setup()

_add_wrapper_script_setup ( self, job, batch_file )

protectedinherited

Adds lines to the submitted script that help with job monitoring/setup. Mostly here so that we can insert
`trap` statements for Ctrl-C situations.

Definition at line 814 of file backends.py.

    def _add_wrapper_script_setup(self, job, batch_file):
        """
        Adds lines to the submitted script that help with job monitoring/setup. Mostly here so that we can insert
        `trap` statements for Ctrl-C situations.
        """
        start_wrapper = f"""# ---
# trap ctrl-c and call ctrl_c()
trap '(ctrl_c 130)' SIGINT
trap '(ctrl_c 143)' SIGTERM
 
function write_exit_code() {{
  echo "Writing $1 to exit status file"
  echo "$1" > {self.exit_code_file}
  exit $1
}}
 
function ctrl_c() {{
  trap '' SIGINT SIGTERM
  echo "** Trapped Ctrl-C **"
  echo "$1" > {self.exit_code_file}
  exit $1
}}
# ---"""
        print(start_wrapper, file=batch_file)
 

_add_wrapper_script_teardown()

_add_wrapper_script_teardown ( self, job, batch_file )

protectedinherited

Adds lines to the submitted script that help with job monitoring/teardown. Mostly here so that we can insert
an exit code of the job cmd being written out to a file. Which means that we can know if the command was
successful or not even if the backend server/monitoring database purges the data about our job i.e. If PBS
removes job information too quickly we may never know if a job succeeded or failed without some kind of exit
file.

Definition at line 839 of file backends.py.

    def _add_wrapper_script_teardown(self, job, batch_file):
        """
        Adds lines to the submitted script that help with job monitoring/teardown. Mostly here so that we can insert
        an exit code of the job cmd being written out to a file. Which means that we can know if the command was
        successful or not even if the backend server/monitoring database purges the data about our job i.e. If PBS
        removes job information too quickly we may never know if a job succeeded or failed without some kind of exit
        file.
        """
        end_wrapper = """# ---
write_exit_code $?"""
        print(end_wrapper, file=batch_file)
 

_create_parent_job_result()

_create_parent_job_result ( cls, parent )

protected

We want to be able to call `ready()` on the top level `Job.result`. So this method needs to exist
so that a Job.result object actually exists. It will be mostly empty and simply updates subjob
statuses and allows the use of ready().

Reimplemented from Backend.

Definition at line 1133 of file backends.py.

    def _create_parent_job_result(cls, parent):
        parent.result = cls.LocalResult(parent, None)
 
 

get_submit_script_path()

get_submit_script_path ( self, job )

inherited

Construct the Path object of the bash script file that we will submit. It will contain
the actual job command, wrapper commands, setup commands, and any batch directives

Definition at line 860 of file backends.py.

    def get_submit_script_path(self, job):
        """
        Construct the Path object of the bash script file that we will submit. It will contain
        the actual job command, wrapper commands, setup commands, and any batch directives
        """
        return Path(job.working_dir, self.submit_script)
 
 

join()

join

(

self

)

Closes and joins the Pool, letting you wait for all results currently
still processing.

Definition at line 985 of file backends.py.

    def join(self):
        """
        Closes and joins the Pool, letting you wait for all results currently
        still processing.
        """
        B2INFO("Joining Process Pool, waiting for results to finish...")
        self.pool.close()
        self.pool.join()
        B2INFO("Process Pool joined.")
 

max_processes() [1/2]

max_processes

(

self

)

Getter for max_processes

Definition at line 996 of file backends.py.

    def max_processes(self):
        """
        Getter for max_processes
        """
        return self._max_processes
 

max_processes() [2/2]

max_processes	(		self,
			value )

Setter for max_processes, we also check for a previous Pool(), wait for it to join
and then create a new one with the new value of max_processes

Definition at line 1003 of file backends.py.

    def max_processes(self, value):
        """
        Setter for max_processes, we also check for a previous Pool(), wait for it to join
        and then create a new one with the new value of max_processes
        """
        
        self._max_processes = value
        if self.pool:
            B2INFO("New max_processes requested. But a pool already exists.")
            self.join()
        B2INFO(f"Starting up new Pool with {self.max_processes} processes")
        self.pool = mp.Pool(processes=self.max_processes)
 

run_job()

run_job ( name, working_dir, output_dir, script )

static

The function that is used by multiprocessing.Pool.apply_async during process creation. This runs a
shell command in a subprocess and captures the stdout and stderr of the subprocess to files.

Definition at line 1107 of file backends.py.

    def run_job(name, working_dir, output_dir, script):
        """
        The function that is used by multiprocessing.Pool.apply_async during process creation. This runs a
        shell command in a subprocess and captures the stdout and stderr of the subprocess to files.
        """
        B2INFO(f"Starting Sub-process: {name}")
        from subprocess import Popen
        stdout_file_path = Path(working_dir, _STDOUT_FILE)
        stderr_file_path = Path(working_dir, _STDERR_FILE)
        # Create unix command to redirect stdour and stderr
        B2INFO(f"stdout/err for subprocess {name} visible at:\n\t{stdout_file_path}\n\t{stderr_file_path}")
        with open(stdout_file_path, mode="w", buffering=1) as f_out, \
                open(stderr_file_path, mode="w", buffering=1) as f_err:
            with Popen(["/bin/bash", script.as_posix()],
                       stdout=f_out,
                       stderr=f_err,
                       bufsize=1,
                       universal_newlines=True,
                       cwd=working_dir,
                       env={}) as p:
                # We block here and wait so that the return code will be set.
                p.wait()
        B2INFO(f"Subprocess {name} finished.")
        return p.returncode
 

submit()

submit	(		self,
			job )

Reimplemented from Backend.

Definition at line 1017 of file backends.py.

    def submit(self, job):
        """
        """
        raise NotImplementedError("This is an abstract submit(job) method that shouldn't have been called. "
                                  "Did you submit a (Sub)Job?")
 

Member Data Documentation

_max_processes

_max_processes = value

protected

Internal attribute of max_processes.

Definition at line 1009 of file backends.py.

backend_args

dict backend_args = {**self.default_backend_args, **backend_args}

inherited

The backend args that will be applied to jobs unless the job specifies them itself.

Definition at line 797 of file backends.py.

default_backend_args

dict default_backend_args = {}

staticinherited

Default backend_args.

Definition at line 789 of file backends.py.

exit_code_file

str exit_code_file = "__BACKEND_CMD_EXIT_STATUS__"

staticinherited

Default exit code file name.

Definition at line 787 of file backends.py.

max_processes

max_processes = max_processes

The size of the multiprocessing process pool.

Definition at line 947 of file backends.py.

pool

pool = None

The actual Pool object of this instance of the Backend.

Definition at line 945 of file backends.py.

submit_script

submit_script = "submit.sh"

staticinherited

Default submission script name.

Definition at line 785 of file backends.py.

---

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

• calibration/scripts/caf/backends.py