From 89c67e79d90bbe546479af3402a0f6e2e27c68db Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 11 Jun 2014 00:19:20 -0500 Subject: [PATCH 1/4] Refactor to simplify __check_if_ready_to_run. Move checks for in memory managed jobs prior to common path through code into their own helper method. --- lib/galaxy/jobs/handler.py | 59 ++++++++++++++++++++++---------------- 1 file changed, 35 insertions(+), 24 deletions(-) diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index 4c884f8997c..b872cb74283 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -240,7 +240,7 @@ class JobHandlerQueue( object ): try: # Check the job's dependencies, requeue if they're not done. # Some of these states will only happen when using the in-memory job queue - job_state = self.__check_if_ready_to_run( job ) + job_state = self.__check_job_state( job ) if job_state == JOB_WAIT: new_waiting_jobs.append( job.id ) elif job_state == JOB_INPUT_ERROR: @@ -281,7 +281,7 @@ class JobHandlerQueue( object ): # Done with the session self.sa_session.remove() - def __check_if_ready_to_run( self, job ): + def __check_job_state( self, job ): """ Check if a job is ready to run by verifying that each of its input datasets is ready (specifically in the OK state). If any input dataset @@ -293,28 +293,12 @@ class JobHandlerQueue( object ): """ # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - if job.state == model.Job.states.DELETED: - return JOB_DELETED - elif job.state == model.Job.states.ERROR: - return JOB_ADMIN_DELETED - for dataset_assoc in job.input_datasets + job.input_library_datasets: - idata = dataset_assoc.dataset - if not idata: - continue - # don't run jobs for which the input dataset was deleted - if idata.deleted: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) - return JOB_INPUT_DELETED - # an error in the input data causes us to bail immediately - elif idata.state == idata.states.ERROR: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state == idata.states.FAILED_METADATA: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): - # need to requeue - return JOB_WAIT + in_memory_job_state = self.__check_in_memory_job_state( job ) + if in_memory_job_state: + return in_memory_job_state + # If no problems, all inputs ready to go, continue + # with same checks either way. + # Create the job wrapper so that the destination can be set if job.id not in self.job_wrappers: self.job_wrappers[job.id] = self.job_wrapper( job ) @@ -348,6 +332,33 @@ class JobHandlerQueue( object ): self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) return state + def __check_in_memory_job_state( self, job ): + if job.state == model.Job.states.DELETED: + return JOB_DELETED + elif job.state == model.Job.states.ERROR: + return JOB_ADMIN_DELETED + for dataset_assoc in job.input_datasets + job.input_library_datasets: + idata = dataset_assoc.dataset + if not idata: + continue + # don't run jobs for which the input dataset was deleted + if idata.deleted: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) + return JOB_INPUT_DELETED + # an error in the input data causes us to bail immediately + elif idata.state == idata.states.ERROR: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state == idata.states.FAILED_METADATA: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): + # need to requeue + return JOB_WAIT + + # All inputs ready to go. + return None + def __clear_job_count( self ): self.user_job_count = None self.user_job_count_per_destination = None From 48b56ffd5776bf2690d218532bc4ebfff212cdc9 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 11 Jun 2014 00:19:20 -0500 Subject: [PATCH 2/4] Refactor __check_if_ready_to_run - less code duplication... ... in the sense that it is no longer repeatedly grabbing same values from map, from wrapper, etc. --- lib/galaxy/jobs/handler.py | 35 +++++++++++++++++++++-------------- 1 file changed, 21 insertions(+), 14 deletions(-) diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index b872cb74283..6e2182fda37 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -291,33 +291,37 @@ class JobHandlerQueue( object ): job can be dispatched. Otherwise, return JOB_WAIT indicating that input datasets are still being prepared. """ - # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - in_memory_job_state = self.__check_in_memory_job_state( job ) - if in_memory_job_state: - return in_memory_job_state - # If no problems, all inputs ready to go, continue - # with same checks either way. + in_memory_not_ready_state = self.__verify_in_memory_job_inputs( job ) + if in_memory_not_ready_state: + return in_memory_not_ready_state + + # Else, if tracking in the database, job.state is guaranteed to be NEW and + # the inputs are guaranteed to be OK. # Create the job wrapper so that the destination can be set - if job.id not in self.job_wrappers: - self.job_wrappers[job.id] = self.job_wrapper( job ) + job_id = job.id + job_wrapper = self.job_wrappers.get( job_id, None ) + if not job_wrapper: + job_wrapper = self.job_wrapper( job ) + self.job_wrappers[ job_id ] = job_wrapper + # Cause the job_destination to be set and cached by the mapper try: - self.job_wrappers[job.id].job_destination + job_destination = job_wrapper.job_destination except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: log.exception( 'Failed to generate job destination' ) else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) - self.job_wrappers[job.id].fail( failure_message ) + job_wrapper.fail( failure_message ) return JOB_ERROR # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable - state = self.__check_destination_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_destination_jobs( job, job_wrapper ) if state == JOB_READY: - state = self.__check_user_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_user_jobs( job, job_wrapper ) if state == JOB_READY and self.app.config.enable_quotas: quota = self.app.quota_agent.get_quota( job.user ) if quota is not None: @@ -329,10 +333,13 @@ class JobHandlerQueue( object ): pass # No history, should not happen with an anon user if state == JOB_READY: # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) + self.increase_running_job_count(job.user_id, job_destination.id ) return state - def __check_in_memory_job_state( self, job ): + def __verify_in_memory_job_inputs( self, job ): + """ Perform the same checks that happen via SQL for in-memory managed + jobs. + """ if job.state == model.Job.states.DELETED: return JOB_DELETED elif job.state == model.Job.states.ERROR: From ff94b8fa941e3d035646d60e7eb86c57b253cfc5 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 11 Jun 2014 00:19:20 -0500 Subject: [PATCH 3/4] Break up __check_job_state - extract new __verify_job_ready method. --- lib/galaxy/jobs/handler.py | 28 +++++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index 6e2182fda37..9d5c417d74b 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -306,8 +306,25 @@ class JobHandlerQueue( object ): job_wrapper = self.job_wrapper( job ) self.job_wrappers[ job_id ] = job_wrapper - # Cause the job_destination to be set and cached by the mapper + # If state == JOB_READY, assume job_destination also set - otherwise + # in case of various error or cancelled states do not assume + # destination has been set. + state, job_destination = self.__verify_job_ready( job, job_wrapper ) + + if state == JOB_READY: + # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration + self.increase_running_job_count(job.user_id, job_destination.id ) + return state + + def __verify_job_ready( self, job, job_wrapper ): + """ Compute job destination and verify job is ready at that + destination by checking job limits and quota. If this method + return a job state of JOB_READY - it MUST also return a job + destination. + """ + job_destination = None try: + # Cause the job_destination to be set and cached by the mapper job_destination = job_wrapper.job_destination except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) @@ -316,7 +333,7 @@ class JobHandlerQueue( object ): else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) job_wrapper.fail( failure_message ) - return JOB_ERROR + return JOB_ERROR, job_destination # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable state = self.__check_destination_jobs( job, job_wrapper ) @@ -328,13 +345,10 @@ class JobHandlerQueue( object ): try: usage = self.app.quota_agent.get_usage( user=job.user, history=job.history ) if usage > quota: - return JOB_USER_OVER_QUOTA + return JOB_USER_OVER_QUOTA, job_destination except AssertionError, e: pass # No history, should not happen with an anon user - if state == JOB_READY: - # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, job_destination.id ) - return state + return state, job_destination def __verify_in_memory_job_inputs( self, job ): """ Perform the same checks that happen via SQL for in-memory managed From 03563202bde03796a1c0e78da1036d11d4e5615f Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 11 Jun 2014 00:19:20 -0500 Subject: [PATCH 4/4] Allow dynamic job destinations to throw new JobNotReadyException to indicate delayed decision. Perhaps wait on some deployer defined limit or resource to become available, etc... --- lib/galaxy/jobs/handler.py | 4 ++++ lib/galaxy/jobs/mapper.py | 7 +++++++ 2 files changed, 11 insertions(+) diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index 9d5c417d74b..6ee681c9960 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -13,6 +13,7 @@ from sqlalchemy.sql.expression import and_, or_, select, func from galaxy import model from galaxy.util.sleeper import Sleeper from galaxy.jobs import JobWrapper, TaskWrapper, JobDestination +from galaxy.jobs.mapper import JobNotReadyException log = logging.getLogger( __name__ ) @@ -326,6 +327,9 @@ class JobHandlerQueue( object ): try: # Cause the job_destination to be set and cached by the mapper job_destination = job_wrapper.job_destination + except JobNotReadyException as e: + job_state = e.job_state or JOB_WAIT + return job_state, None except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: diff --git a/lib/galaxy/jobs/mapper.py b/lib/galaxy/jobs/mapper.py index 48901eb266e..51fcf98fa32 100644 --- a/lib/galaxy/jobs/mapper.py +++ b/lib/galaxy/jobs/mapper.py @@ -16,6 +16,13 @@ class JobMappingException( Exception ): self.failure_message = failure_message +class JobNotReadyException( Exception ): + + def __init__( self, job_state=None, message=None ): + self.job_state = job_state + self.message = message + + class JobRunnerMapper( object ): """ This class is responsible to managing the mapping of jobs