#NEWLINE# def _get_work( self ) : #NEWLINE# #INDENT# if self . _stop_requesting_work : #NEWLINE# #INDENT# return ( None , 0 , 0 , 0 ) #NEWLINE# #UNINDENT# logger . debug ( 'Asking scheduler for work...' ) #NEWLINE# r = self . _scheduler . get_work ( worker = self . _id , host = self . host , assistant = self . _assistant , current_tasks = list ( self . _running_tasks . keys ( ) ) ) #NEWLINE# n_pending_tasks = r [ 'n_pending_tasks' ] #NEWLINE# task_id = r [ 'task_id' ] #NEWLINE# running_tasks = r [ 'running_tasks' ] #NEWLINE# n_unique_pending = r [ 'n_unique_pending' ] #NEWLINE# self . _get_work_response_history . append ( dict ( task_id = task_id , running_tasks = running_tasks ) ) #NEWLINE# if ( ( task_id is not None ) and ( task_id not in self . _scheduled_tasks ) ) : #NEWLINE# #INDENT# logger . info ( 'Did not schedule %s, will load it dynamically' , task_id ) #NEWLINE# try : #NEWLINE# #INDENT# self . _scheduled_tasks [ task_id ] = load_task ( module = r . get ( 'task_module' ) , task_name = r [ 'task_family' ] , params_str = r [ 'task_params' ] ) #NEWLINE# #UNINDENT# except TaskClassException as ex : #NEWLINE# #INDENT# msg = ( 'Cannot find task for %s' % task_id ) #NEWLINE# logger . exception ( msg ) #NEWLINE# subject = ( 'Luigi: %s' % msg ) #NEWLINE# error_message = notifications . wrap_traceback ( ex ) #NEWLINE# notifications . send_error_email ( subject , error_message ) #NEWLINE# self . _add_task ( worker = self . _id , task_id = task_id , status = FAILED , runnable = False , assistant = self . _assistant ) #NEWLINE# task_id = None #NEWLINE# self . run_succeeded = False #NEWLINE# #UNINDENT# #UNINDENT# return ( task_id , running_tasks , n_pending_tasks , n_unique_pending )