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
	__init__ (self, jobSubmission=None, numberOfProcesses=multiprocessing.cpu_count(), startMethod="fork")
	initialisation method that accepts submissions and starts pool
	__str__ (self)
	return an object self-description string
	printout (self)
	print in a human-readable way the items of the object self
	submit (self, jobSubmission=None)
	submit a Job object or a JobGroup object for processing
	getResults (self)
	get results of JobGroup object submission
	statusReport (self)
	return a status report string

Public Attributes
	jobSubmission = jobSubmission
	numberOfProcesses = numberOfProcesses
	startMethod = startMethod
	className = self.__class__.__name__
str	status = "starting"
	countOfJobs = None
int	countOfRemainingJobs = 0
	pool
list	listOfNamesOfRemainingJobs = []

Protected Member Functions
	_abort (self)
	abort parallel job processor
	_terminate (self)
	terminate parallel job processor

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 869 of file trfUtils.py.

Constructor & Destructor Documentation

__init__()

python.trfUtils.ParallelJobProcessor.__init__	(		self,
			jobSubmission = None,
			numberOfProcesses = multiprocessing.cpu_count(),
			startMethod = "fork" )

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 875 of file trfUtils.py.

        ):
        self.jobSubmission = jobSubmission
        self.numberOfProcesses = numberOfProcesses
        self.startMethod = startMethod
        self.className = self.__class__.__name__
        msg.debug(f"{self.className}: current process: {multiprocessing.current_process().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.Pool(
            self.numberOfProcesses,
            initialise_processes,
            (),
            None,
            multiprocessing.get_context(self.startMethod)
        )
        msg.debug("{notifier}: pool of {numberOfProcesses} {units} with start method {startMethod!r} created".format(
            notifier = self.className,
            numberOfProcesses = str(self.numberOfProcesses),
            units = units(quantity = self.numberOfProcesses,
            unitSingular = "process", unitPlural = "processes"),
            startMethod = self.startMethod,
        ))
        self.status = "ready"
        msg.debug("{notifier}: status: {status}".format(
            notifier = self.className,
            status = self.status
        ))
 

Member Function Documentation

__str__()

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 921 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()

python.trfUtils.ParallelJobProcessor._abort ( self )

protected

abort parallel job processor

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

Definition at line 1225 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()

python.trfUtils.ParallelJobProcessor._terminate ( self )

protected

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 1238 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()

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 1022 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
        # retrieved 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(5.0)
            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:
                    if not hasattr(job, 'result'):
                        # 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
                                self.listOfNamesOfRemainingJobs.remove(job.name)
                                msg.debug(
                                    "{notifier}: {countOfRemainingJobs} {units} remaining".format(
                                        notifier = self.className,
                                        countOfRemainingJobs = self.countOfRemainingJobs,
                                        units = units(
                                            quantity = self.countOfRemainingJobs,
                                            unitSingular = "job",
                                            unitPlural = "jobs"
                                        )
                                    )
                                    + f"\n          names of remaining jobs: {self.listOfNamesOfRemainingJobs}"
                                )
                            # 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
 

printout()

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 933 of file trfUtils.py.

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

statusReport()

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 1168 of file trfUtils.py.

    def statusReport(self):
        statusReport = "{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()

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 941 of file trfUtils.py.

        ):
        # 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
            # multiprocessing.pool.ApplyResult (AsyncResult alias thereof)
            # 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 = self.__class__.__name__

Definition at line 887 of file trfUtils.py.

countOfJobs

python.trfUtils.ParallelJobProcessor.countOfJobs = None

Definition at line 894 of file trfUtils.py.

countOfRemainingJobs

int python.trfUtils.ParallelJobProcessor.countOfRemainingJobs = 0

Definition at line 895 of file trfUtils.py.

jobSubmission

python.trfUtils.ParallelJobProcessor.jobSubmission = jobSubmission

Definition at line 884 of file trfUtils.py.

listOfNamesOfRemainingJobs

list python.trfUtils.ParallelJobProcessor.listOfNamesOfRemainingJobs = []

Definition at line 965 of file trfUtils.py.

numberOfProcesses

python.trfUtils.ParallelJobProcessor.numberOfProcesses = numberOfProcesses

Definition at line 885 of file trfUtils.py.

pool

python.trfUtils.ParallelJobProcessor.pool

Initial value:

=  multiprocessing.pool.Pool(
            self.numberOfProcesses,
            initialise_processes,
            (),
            None,
            multiprocessing.get_context(self.startMethod)
        )

Definition at line 897 of file trfUtils.py.

startMethod

python.trfUtils.ParallelJobProcessor.startMethod = startMethod

Definition at line 886 of file trfUtils.py.

status

str python.trfUtils.ParallelJobProcessor.status = "starting"

Definition at line 889 of file trfUtils.py.

---

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

• trfUtils.py