#NEWLINE# def setStageFinished( self , index , clientURI , save_state = True , checking_pipeline_status = False ) : #NEWLINE# #INDENT# 'given an index, sets corresponding stage to finished and adds successors to the runnable set' #NEWLINE# s = self . stages [ index ] #NEWLINE# if s . isFinished ( ) : #NEWLINE# #INDENT# raise ValueError ( ( 'Already finished stage %d' % index ) ) #NEWLINE# #UNINDENT# if checking_pipeline_status : #NEWLINE# #INDENT# logger . log ( SUBDEBUG , ( 'Already finished stage ' + str ( i ) ) ) #NEWLINE# s . status = 'finished' #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# logger . info ( ( ( ( 'Finished Stage ' + str ( index ) ) + ': ' ) + str ( self . stages [ index ] ) ) ) #NEWLINE# self . removeFromRunning ( index , clientURI , new_status = 'finished' ) #NEWLINE# for f in s . finished_hooks : #NEWLINE# #INDENT# f ( ) #NEWLINE# #UNINDENT# #UNINDENT# self . num_finished_stages += 1 #NEWLINE# self . finished_stages_fh . write ( ( '%d,%s\n' % ( index , self . stages [ index ] . getHash ( ) ) ) ) #NEWLINE# self . finished_stages_fh . flush ( ) #NEWLINE# for i in self . G . successors ( index ) : #NEWLINE# #INDENT# self . unfinished_pred_counts [ i ] -= 1 #NEWLINE# if self . checkIfRunnable ( i ) : #NEWLINE# #INDENT# self . enqueue ( i )