opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix preservation of resubmits for resubmitted jobs.

Browse Source 
Create an abstraction for the explicit handling of resubmits that is done during job recovery at startup.

xref https://github.com/galaxyproject/galaxy/commit/5539a083e21b601bfd2f87fa3417f23bf295a690
This commit is contained in:
John Chilton
2016-12-14 13:41:29 -05:00
parent 7d52b28ab2
commit 949d122489
1 changed files with 18 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/handler.py
+18 -14
View File
@@ -149,22 +149,27 @@ class JobHandlerQueue( object ):
self.queue.put( ( job.id, job.tool_id ) )
else:
# Already dispatched and running
job_wrapper = self.job_wrapper( job )
# Use the persisted destination as its params may differ from
# what's in the job_conf xml
job_destination = JobDestination(id=job.destination_id, runner=job.job_runner_name, params=job.destination_params)
# resubmits are not persisted (it's a good thing) so they
# should be added back to the in-memory destination on startup
try:
config_job_destination = self.app.job_config.get_destination( job.destination_id )
job_destination.resubmit = config_job_destination.resubmit
except KeyError:
log.warning( '(%s) Recovered destination id (%s) does not exist in job config (but this may be normal in the case of a dynamically generated destination)', job.id, job.destination_id )
job_wrapper.job_runner_mapper.cached_job_destination = job_destination
job_wrapper = self.__recover_job_wrapper( job )
self.dispatcher.recover( job, job_wrapper )
if self.sa_session.dirty:
self.sa_session.flush()
def __recover_job_wrapper(self, job):
# Already dispatched and running
job_wrapper = self.job_wrapper( job )
# Use the persisted destination as its params may differ from
# what's in the job_conf xml
job_destination = JobDestination(id=job.destination_id, runner=job.job_runner_name, params=job.destination_params)
# resubmits are not persisted (it's a good thing) so they
# should be added back to the in-memory destination on startup
try:
config_job_destination = self.app.job_config.get_destination( job.destination_id )
job_destination.resubmit = config_job_destination.resubmit
except KeyError:
log.debug( '(%s) Recovered destination id (%s) does not exist in job config (but this may be normal in the case of a dynamically generated destination)', job.id, job.destination_id )
job_wrapper.job_runner_mapper.cached_job_destination = job_destination
return job_wrapper
def __monitor( self ):
"""
Continually iterate the waiting jobs, checking is each is ready to
@@ -260,8 +265,7 @@ class JobHandlerQueue( object ):
for job in resubmit_jobs:
log.debug( '(%s) Job was resubmitted and is being dispatched immediately', job.id )
# Reassemble resubmit job destination from persisted value
jw = self.job_wrapper( job )
jw.job_runner_mapper.cached_job_destination = JobDestination( id=job.destination_id, runner=job.job_runner_name, params=job.destination_params )
jw = self.__recover_job_wrapper( job )
self.increase_running_job_count(job.user_id, jw.job_destination.id)
self.dispatcher.put( jw )
# Iterate over new and waiting jobs and look for any that are