python.trfUtils.ParallelJobProcessor Class Reference

ParallelJobProcessor: a multiple-process processor of Job objects. More...

Inheritance diagram for python.trfUtils.ParallelJobProcessor:

Collaboration diagram for python.trfUtils.ParallelJobProcessor:

Public Member Functions
def	__init__ (self, jobSubmission=None, numberOfProcesses=multiprocessing.cpu_count())
	initialisation method that accepts submissions and starts pool More...
def	__str__ (self)
	return an object self-description string More...
def	printout (self)
	print in a human-readable way the items of the object self More...
def	submit (self, jobSubmission=None)
	submit a Job object or a JobGroup object for processing More...
def	getResults (self)
	get results of JobGroup object submission More...
def	statusReport (self)
	return a status report string More...

Public Attributes
	jobSubmission
	numberOfProcesses
	className
	status
	countOfJobs
	countOfRemainingJobs
	pool
	listOfNamesOfRemainingJobs

Private Member Functions
def	_abort (self)
	abort parallel job processor More...
def	_terminate (self)
	terminate parallel job processor More...

Detailed Description

ParallelJobProcessor: a multiple-process processor of Job objects.

Parameters

jobSubmission	Job object or JobGroup object for submission
numberOfProcesses	the number of processes in the process pool

Definition at line 867 of file trfUtils.py.

Constructor & Destructor Documentation

__init__()

def python.trfUtils.ParallelJobProcessor.__init__	(		self,
			jobSubmission = None,
			numberOfProcesses = multiprocessing.cpu_count()
	)

initialisation method that accepts submissions and starts pool

This method is the initialisation method of the parallel job processor. It accepts input JobGroup object submissions and prepares a pool of workers.

Definition at line 873 of file trfUtils.py.

    def __init__(
        self,
        jobSubmission = None,
        numberOfProcesses = multiprocessing.cpu_count(),  # noqa: B008 (cpu_count is constant)
        ):
        self.jobSubmission = jobSubmission
        self.numberOfProcesses = numberOfProcesses
        self.className = self.__class__.__name__
        self.status = "starting"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status)
        )
        self.countOfJobs = None
        self.countOfRemainingJobs = 0
        self.pool = multiprocessing.Pool(
            self.numberOfProcesses,
            initialise_processes
        )
        msg.debug("{notifier}: pool of {numberOfProcesses} {units} created".format(
            notifier = self.className,
            numberOfProcesses = str(self.numberOfProcesses),
            units = units(quantity = self.numberOfProcesses,
            unitSingular = "process", unitPlural = "processes")
        ))
        self.status = "ready"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
 

Member Function Documentation

__str__()

def python.trfUtils.ParallelJobProcessor.__str__

(

self

)

return an object self-description string

This method returns an object description string consisting of a listing of the items of the object self.

Returns

object description string

Definition at line 908 of file trfUtils.py.

    def __str__(self):
        descriptionString = ""
        for key, value in sorted(vars(self).items()):
            descriptionString += str("{key}:{value} ".format(
                key = key,
                value = value
            ))
        return descriptionString
 

_abort()

def python.trfUtils.ParallelJobProcessor._abort (   self )

private

abort parallel job processor

This method aborts the parallel job processor. It is used typically when an exception is raised.

Definition at line 1214 of file trfUtils.py.

    def _abort(self):
        self.status = "aborting"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
        self._terminate()
 

_terminate()

def python.trfUtils.ParallelJobProcessor._terminate (   self )

private

terminate parallel job processor

This method terminates the parallel job processor. It terminates the subprocesses of the parallel job processor. It is used typically when terminating the parallel job processor on successful completion of job processing and when aborting the parallel job processor.

Definition at line 1227 of file trfUtils.py.

    def _terminate(self):
        self.status = "terminating"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
        msg.debug("{notifier}: terminating pool of {numberOfProcesses} {units}".format(
            notifier = self.className,
            numberOfProcesses = str(self.numberOfProcesses),
            units = units(
                quantity = self.numberOfProcesses,
                unitSingular = "process",
                unitPlural = "processes"
            )
        ))
        self.pool.terminate()
        self.pool.join()
        self.status = "finished"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
        msg.debug(self.statusReport())
 

getResults()

def python.trfUtils.ParallelJobProcessor.getResults

(

self

)

get results of JobGroup object submission

This method returns an ordered list of results for jobs submitted.

Returns

order list of results for jobs

Definition at line 1008 of file trfUtils.py.

    def getResults(self):
        # While the number of jobs remaining is greater than zero, cycle over
        # all jobs in the JobGroup object submission submission, watching for a
        # timeout of the JobGroup object submission. If a result has not been
        # retrived for a job (i.e. the Job object does not have a result data
        # attribute), then check if a result is available for the job (using the
        # method multiprocessing.pool.AsyncResult.ready()). If a result is
        # available for the job, then check if the job has run successfully
        # (using the method multiprocessing.pool.AsyncResult.successful()). If
        # the job has not been successful, raise an exception, otherwise, get
        # the result of the job and save it to the result data attribute of the
        # job.
        msg.debug("{notifier}: checking for job {units}".format(
            notifier = self.className,
            units = units(
                quantity = self.countOfRemainingJobs,
                unitSingular = "result",
                unitPlural = "results")
            )
        )
        while self.countOfRemainingJobs > 0:
            # Check for timeout of the job group. If the current timestamp is
            # greater than the job group timeout (derived from the sum of the
            # set of all job timeout specifications in the job group) + the job
            # group submission timestamp, then raise an excepton, otherwise
            # cycle over all jobs.
            # Allow time for jobs to complete.
            time.sleep(0.25)
            if self.jobSubmission.timeoutStatus():
                msg.error("{notifier}: job group '{name}' timed out".format(
                    notifier = self.className,
                    name = self.jobSubmission.name
                ))
                self._abort()
                exceptionMessage = "timeout of a function in list {listOfNamesOfRemainingJobs}".format(
                    listOfNamesOfRemainingJobs = self.listOfNamesOfRemainingJobs
                )
                msg.error("{notifier}: exception message: {exceptionMessage}".format(
                    notifier = self.className,
                    exceptionMessage = exceptionMessage
                ))
                raise trfExceptions.TransformTimeoutException(
                    trfExit.nameToCode('TRF_EXEC_TIMEOUT'),
                    exceptionMessage
                )
            else:
                for job in self.jobSubmission.jobs:
                    self.listOfNamesOfRemainingJobs = []
                    if not hasattr(job, 'result'):
                        # Maintain a contemporary list of the names of remaining
                        # jobs.
                        self.listOfNamesOfRemainingJobs.append(job.name)
                        # If the result of the job is ready...
                        if job.resultGetter.ready():
                            msg.debug(
                                "{notifier}: result ready for job '{name}'".format(
                                    notifier = self.className,
                                    name = job.name
                                )
                            )
                            job.successStatus = job.resultGetter.successful()
                            msg.debug(
                                "{notifier}: job '{name}' success status: {successStatus}".format(
                                    notifier = self.className,
                                    name = job.name,
                                    successStatus = job.successStatus
                                )
                            )
                            # If the job was successful, create the result data
                            # attribute of the job and save the result to it.
                            if job.successStatus:
                                job.result = job.resultGetter.get()
                                msg.debug(
                                    "{notifier}: result of job '{name}': {result}".format(
                                        notifier = self.className,
                                        name = job.name,
                                        result = job.result
                                    )
                                )
                                self.countOfRemainingJobs -= 1
                                msg.debug(
                                    "{notifier}: {countOfRemainingJobs} {units} remaining".format(
                                        notifier = self.className,
                                        countOfRemainingJobs = self.countOfRemainingJobs,
                                        units = units(
                                            quantity = self.countOfRemainingJobs,
                                            unitSingular = "job",
                                            unitPlural = "jobs"
                                        )
                                    )
                                )
                            # If the job was not successful, raise an exception
                            # and abort processing.
                            elif not job.successStatus:
                                msg.error(
                                    "{notifier}: job '{name}' failed".format(
                                        notifier = self.className,
                                        name = job.name
                                    )
                                )
                                self._abort()
                                exceptionMessage = "failure of function '{name}' with arguments {arguments}".format(
                                    name = job.workFunction.__name__,
                                    arguments = job.workFunctionKeywordArguments
                                )
                                msg.error("{notifier}: exception message: {exceptionMessage}".format(
                                    notifier = self.className,
                                    exceptionMessage = exceptionMessage
                                ))
                                raise trfExceptions.TransformExecutionException(
                                    trfExit.nameToCode('TRF_EXEC_FAIL'),
                                    exceptionMessage
                                )
        # All results having been returned, create the 'results' list data
        # attribute of the job group and append all individual job results to
        # it.
        self.jobSubmission.timeStampComplete = time.time()
        self.jobSubmission.completeStatus = True
        msg.debug("{notifier}: all {countOfJobs} {units} complete (timestamp: {timeStampComplete})".format(
            notifier = self.className,
            countOfJobs = self.countOfJobs,
            units = units(
                quantity = self.countOfJobs,
                unitSingular = "job",
                unitPlural = "jobs"
            ),
            timeStampComplete = self.jobSubmission.timeStampComplete
        ))
        self.jobSubmission.processingTime = self.jobSubmission.timeStampComplete - self.jobSubmission.timeStampSubmission
        msg.debug("{notifier}: time taken to process all {countOfJobs} {units}: {processingTime}".format(
            notifier = self.className,
            countOfJobs = self.countOfJobs,
            units = units(
                quantity = self.countOfJobs,
                unitSingular = "job",
                unitPlural = "jobs"
            ),
            processingTime = self.jobSubmission.processingTime
        ))
        for job in self.jobSubmission.jobs:
            self.jobSubmission.results.append(job.result)
            self._terminate()
        return self.jobSubmission.results
        self._terminate()
 

printout()

def python.trfUtils.ParallelJobProcessor.printout

(

self

)

print in a human-readable way the items of the object self

This function prints in a human-readable way the items of the object self.

Definition at line 920 of file trfUtils.py.

    def printout(self):
        printHR(vars(self)
        )
 

statusReport()

def python.trfUtils.ParallelJobProcessor.statusReport

(

self

)

return a status report string

This method returns a status report string, detailing information on the JobGroup submission and on the job processing status.

Returns

status report string

Definition at line 1157 of file trfUtils.py.

    def statusReport(self):
        statusReport = "\n{notifier}:\n   status report:".format(
            notifier = self.className
        )
        # information on parallel job processor
        statusReport += "\n       parallel job processor configuration:"
        statusReport += "\n          status: {notifier}".format(
            notifier = str(self.status)
        )
        statusReport += "\n          number of processes: {notifier}".format(
            notifier = str(self.numberOfProcesses)
        )
        # information on job group submission
        statusReport += "\n       job group submission: '{notifier}'".format(
            notifier = self.jobSubmission.name
        )
        statusReport += "\n          total number of jobs: {notifier}".format(
            notifier = str(self.countOfJobs)
        )
        statusReport += "\n          number of incomplete jobs: {notifier}".format(
            notifier = str(self.countOfRemainingJobs)
        )
        statusReport += "\n          names of incomplete jobs: {notifier}".format(
            notifier = self.listOfNamesOfRemainingJobs
        )
        # information on jobs (if existent)
        if self.jobSubmission.jobs:
            statusReport += "\n       jobs:"
            for job in self.jobSubmission.jobs:
                statusReport += "\n          job '{name}':".format(
                    name = job.name
                )
                statusReport += "\n              workFunction: '{name}'".format(
                    name = job.workFunction.__name__
                )
                statusReport += "\n              workFunctionKeywordArguments: '{arguments}'".format(
                    arguments = job.workFunctionKeywordArguments
                )
                statusReport += "\n              workFunctionTimeout: '{timeout}'".format(
                    timeout = job.workFunctionTimeout
                )
                if hasattr(job, 'result'):
                    statusReport += "\n              result: '{result}'".format(
                        result = job.result
                    )
        # statistics of parallel job processor run
        if hasattr(self.jobSubmission, 'processingTime'):
            statusReport += "\n       statistics:"
        if hasattr(self.jobSubmission, 'processingTime'):
            statusReport += "\n          total processing time: {processingTime} s".format(
                processingTime = self.jobSubmission.processingTime
            )
        return statusReport
 

submit()

def python.trfUtils.ParallelJobProcessor.submit	(		self,
			jobSubmission = None
	)

submit a Job object or a JobGroup object for processing

This method submits a specified Job object or JobGroup object for processing. On successful submission, it returns the value 0.

Parameters

jobSubmission

Job object or JobGroup object for submission

Definition at line 928 of file trfUtils.py.

    def submit(
        self,
        jobSubmission = None
        ):
        # If the input submission is not None, then update the jobSubmission
        # data attribute to that specified for this method.
        if jobSubmission is not None:
            self.jobSubmission = jobSubmission
        self.status = "submitted"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
        # If the input submission is a Job object, contain it in a JobGroup
        # object.
        if isinstance(self.jobSubmission, Job):
            jobGroup = JobGroup(
                jobs = [self.jobSubmission,],
            )
            self.jobSubmission = jobGroup
        # Count the number of jobs.
        self.countOfJobs = len(self.jobSubmission.jobs)
        self.countOfRemainingJobs = self.countOfJobs
        # Build a contemporary list of the names of jobs.
        self.listOfNamesOfRemainingJobs = []
        for job in self.jobSubmission.jobs:
            self.listOfNamesOfRemainingJobs.append(job.name)
        msg.debug("{notifier}: received job group submission '{name}' of {countOfJobs} {units}".format(
            notifier = self.className,
            name = self.jobSubmission.name,
            countOfJobs = self.countOfJobs,
            units = units(
                quantity = self.countOfRemainingJobs,
                unitSingular = "job",
                unitPlural = "jobs"
            )
        ))
        msg.debug(self.statusReport())
        msg.debug("{notifier}: submitting job group submission '{name}' to pool".format(
            notifier = self.className,
            name = self.jobSubmission.name
        ))
        # Cycle through all jobs in the input submission and apply each to the
        # pool.
        for job in self.jobSubmission.jobs:
            job.timeStampSubmission = time.time()
            msg.debug("{notifier}: job '{name}' submitted to pool".format(
                notifier = self.className,
                name = job.name
            ))
            # Apply the job to the pool, applying the object pool.ApplyResult
            # to the job as a data attribute.
            job.resultGetter = self.pool.apply_async(
                func = job.workFunction,
                kwds = job.workFunctionKeywordArguments
            )
        # Prepare monitoring of job group times in order to detect a job group
        # timeout by recording the time of complete submission of the job group.
        self.jobSubmission.timeStampSubmission = time.time()
        msg.debug("{notifier}: job group submission complete: {countOfJobs} {units} submitted to pool (timestamp: {timeStampSubmission})".format(
            notifier = self.className,
            countOfJobs = self.countOfJobs,
            units = units(
                quantity = self.countOfJobs,
                unitSingular = "job",
                unitPlural = "jobs"
            ),
            timeStampSubmission = self.jobSubmission.timeStampSubmission
        ))
        self.status = "processing"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
        return 0
 

Member Data Documentation

className

python.trfUtils.ParallelJobProcessor.className

Definition at line 876 of file trfUtils.py.

countOfJobs

python.trfUtils.ParallelJobProcessor.countOfJobs

Definition at line 882 of file trfUtils.py.

countOfRemainingJobs

python.trfUtils.ParallelJobProcessor.countOfRemainingJobs

Definition at line 883 of file trfUtils.py.

jobSubmission

python.trfUtils.ParallelJobProcessor.jobSubmission

Definition at line 874 of file trfUtils.py.

listOfNamesOfRemainingJobs

python.trfUtils.ParallelJobProcessor.listOfNamesOfRemainingJobs

Definition at line 949 of file trfUtils.py.

numberOfProcesses

python.trfUtils.ParallelJobProcessor.numberOfProcesses

Definition at line 875 of file trfUtils.py.

pool

python.trfUtils.ParallelJobProcessor.pool

Definition at line 884 of file trfUtils.py.

status

python.trfUtils.ParallelJobProcessor.status

Definition at line 877 of file trfUtils.py.

---

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

• trfUtils.py