#NEWLINE# def dispatch( self ) : #NEWLINE# #INDENT# '\n        Starts a loop with the batch system to run the Toil workflow\n        ' #NEWLINE# assert ( len ( self . batchSystem . getIssuedBatchJobIDs ( ) ) == 0 ) #NEWLINE# logger . info ( 'Checked batch system has no running jobs and no updated jobs' ) #NEWLINE# logger . info ( 'Found %s jobs to start and %i jobs with successors to run' , len ( self . toilState . updatedJobs ) , len ( self . toilState . successorCounts ) ) #NEWLINE# timeSinceJobsLastRescued = time . time ( ) #NEWLINE# totalFailedJobs = 0 #NEWLINE# logger . info ( 'Starting the main loop' ) #NEWLINE# while True : #NEWLINE# #INDENT# if ( len ( self . toilState . updatedJobs ) > 0 ) : #NEWLINE# #INDENT# logger . debug ( 'Built the jobs list, currently have %i jobs to update and %i jobs issued' , len ( self . toilState . updatedJobs ) , self . getNumberOfJobsIssued ( ) ) #NEWLINE# for ( jobWrapper , resultStatus ) in self . toilState . updatedJobs : #NEWLINE# #INDENT# if ( ( jobWrapper . command != None ) or ( resultStatus != 0 ) ) : #NEWLINE# #INDENT# if ( jobWrapper . remainingRetryCount > 0 ) : #NEWLINE# #INDENT# iJ = IssuedJob ( jobWrapper . jobStoreID , jobWrapper . memory , jobWrapper . cores , jobWrapper . disk , jobWrapper . preemptable ) #NEWLINE# self . issueJob ( iJ ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# totalFailedJobs += 1 #NEWLINE# logger . warn ( 'Job: %s is completely failed' , jobWrapper . jobStoreID ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( len ( jobWrapper . stack ) > 0 ) : #NEWLINE# #INDENT# assert ( len ( jobWrapper . stack [ ( - 1 ) ] ) > 0 ) #NEWLINE# logger . debug ( 'Job: %s has %i successors to schedule' , jobWrapper . jobStoreID , len ( jobWrapper . stack [ ( - 1 ) ] ) ) #NEWLINE# assert ( jobWrapper not in self . toilState . successorCounts ) #NEWLINE# self . toilState . successorCounts [ jobWrapper ] = len ( jobWrapper . stack [ ( - 1 ) ] ) #NEWLINE# successors = [ ] #NEWLINE# for ( successorJobStoreID , memory , cores , disk , preemptable , predecessorID ) in jobWrapper . stack . pop ( ) : #NEWLINE# #INDENT# if ( successorJobStoreID not in self . toilState . successorJobStoreIDToPredecessorJobs ) : #NEWLINE# #INDENT# self . toilState . successorJobStoreIDToPredecessorJobs [ successorJobStoreID ] = [ ] #NEWLINE# #UNINDENT# self . toilState . successorJobStoreIDToPredecessorJobs [ successorJobStoreID ] . append ( jobWrapper ) #NEWLINE# if ( predecessorID != None ) : #NEWLINE# #INDENT# job2 = self . jobStore . load ( successorJobStoreID ) #NEWLINE# job2 . predecessorsFinished . add ( predecessorID ) #NEWLINE# self . jobStore . update ( job2 ) #NEWLINE# assert ( len ( job2 . predecessorsFinished ) >= 1 ) #NEWLINE# assert ( len ( job2 . predecessorsFinished ) <= job2 . predecessorNumber ) #NEWLINE# if ( len ( job2 . predecessorsFinished ) < job2 . predecessorNumber ) : #NEWLINE# #INDENT# continue #NEWLINE# #UNINDENT# #UNINDENT# successors . append ( IssuedJob ( successorJobStoreID , memory , cores , disk , preemptable ) ) #NEWLINE# #UNINDENT# map ( self . issueJob , successors ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( jobWrapper . remainingRetryCount > 0 ) : #NEWLINE# #INDENT# iJ = IssuedJob ( jobWrapper . jobStoreID , self . config . defaultMemory , self . config . defaultCores , self . config . defaultDisk , True ) #NEWLINE# self . issueJob ( iJ ) #NEWLINE# logger . debug ( 'Job: %s is empty, we are scheduling to clean it up' , jobWrapper . jobStoreID ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# totalFailedJobs += 1 #NEWLINE# logger . warn ( 'Job: %s is empty but completely failed - something is very wrong' , jobWrapper . jobStoreID ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# self . toilState . updatedJobs = set ( ) #NEWLINE# #UNINDENT# if ( self . getNumberOfJobsIssued ( ) == 0 ) : #NEWLINE# #INDENT# logger . info ( 'Only failed jobs and their dependents (%i total) are remaining, so exiting.' , totalFailedJobs ) #NEWLINE# return totalFailedJobs #NEWLINE# #UNINDENT# if ( self . processAnyUpdatedJob ( 10 ) == None ) : #NEWLINE# #INDENT# if ( ( time . time ( ) - timeSinceJobsLastRescued ) >= self . config . rescueJobsFrequency ) : #NEWLINE# #INDENT# self . reissueOverLongJobs ( ) #NEWLINE# logger . info ( 'Reissued any over long jobs' ) #NEWLINE# hasNoMissingJobs = self . reissueMissingJobs ( ) #NEWLINE# if hasNoMissingJobs : #NEWLINE# #INDENT# timeSinceJobsLastRescued = time . time ( ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# timeSinceJobsLastRescued += 60 #NEWLINE# #UNINDENT# logger . info ( 'Rescued any (long) missing jobs' )