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

Make sure dependent jobs are being unpaused

Browse Source 
Fixes https://github.com/galaxyproject/galaxy/issues/6926
This commit is contained in:
mvdbeek
2019-04-11 19:35:20 +02:00
parent fc0b9d9572
commit 0b8dec5af0
3 changed files with 49 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/model/__init__.py
+19 -2
View File
@@ -994,8 +994,11 @@ class Job(JobLike, UsesCreateAndUpdateTime, Dictifiable, RepresentById):
if self.state == self.states.PAUSED:
self.set_state(self.states.NEW)
object_session(self).add(self)
for dataset in self.output_datasets:
dataset.info = None
jobs_to_resume = set()
for jtod in self.output_datasets:
jobs_to_resume.update(jtod.dataset.unpause_dependent_jobs(jobs_to_resume))
for job in jobs_to_resume:
job.resume(flush=False)
if flush:
object_session(self).flush()
@@ -3102,6 +3105,20 @@ class HistoryDatasetAssociation(DatasetInstance, HasTags, Dictifiable, UsesAnnot
rval['metadata_' + name] = val
return rval
def unpause_dependent_jobs(self, jobs=None):
if self.state == self.states.PAUSED:
self.state = self.states.NEW
self.info = None
jobs_to_unpause = jobs or set()
for jtida in self.dependent_jobs:
if jtida.job not in jobs_to_unpause:
jobs_to_unpause.add(jtida.job)
for jtoda in jtida.job.output_datasets:
jobs_to_unpause.update(
jtoda.dataset.unpause_dependent_jobs(jobs=jobs_to_unpause)
)
return jobs_to_unpause
@property
def history_content_type(self):
return "dataset"
lib/galaxy/tools/actions/__init__.py
+1 -6
View File
@@ -616,6 +616,7 @@ class DefaultToolAction(object):
self.__remap_parameters(job_to_remap, jtid, jtod, out_data)
trans.sa_session.add(job_to_remap)
trans.sa_session.add(jtid)
job_to_remap.resume()
jtod.dataset.visible = False
trans.sa_session.add(jtod)
for jtodc in old_job.output_dataset_collection_instances:
@@ -639,12 +640,6 @@ class DefaultToolAction(object):
return remapped_hdas
def __remap_parameters(self, job_to_remap, jtid, jtod, out_data):
if job_to_remap.state == job_to_remap.states.PAUSED:
job_to_remap.state = job_to_remap.states.NEW
for hda in [dep_jtod.dataset for dep_jtod in job_to_remap.output_datasets]:
if hda.state == hda.states.PAUSED:
hda.state = hda.states.NEW
hda.info = None
input_values = dict([(p.name, json.loads(p.value)) for p in job_to_remap.parameters])
old_dataset_id = jtod.dataset_id
new_dataset_id = out_data[jtod.name].id
test/api/test_workflows.py
+29 -9
View File
@@ -738,6 +738,11 @@ steps:
cond_param_inner: true
input1:
$link: 0#out_file1
cat:
tool_id: cat1
in:
input1: identifier#output1
queries_0|input2: identifier#output1
""")
with self.dataset_populator.test_history() as history_id:
invocation_id = self.__invoke_workflow(history_id, workflow_id)
@@ -753,11 +758,14 @@ steps:
inputs=inputs,
history_id=history_id,
assert_ok=True)
unpaused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True, assert_ok=False)
assert unpaused_dataset['state'] == 'ok'
unpaused_dataset_1 = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True, assert_ok=False)
assert unpaused_dataset_1['state'] == 'ok'
self.dataset_populator.wait_for_history(history_id, assert_ok=False)
unpaused_dataset_2 = self.dataset_populator.get_history_dataset_details(history_id, hid=6, wait=True, assert_ok=False)
assert unpaused_dataset_2['state'] == 'ok'
@skip_without_tool("job_properties")
@skip_without_tool("identifier_collection")
@skip_without_tool("collection_creates_list")
def test_workflow_resume_from_failed_step_with_hdca_input(self):
workflow_id = self._upload_yaml_workflow("""
class: GalaxyWorkflow
@@ -767,18 +775,26 @@ steps:
state:
thebool: true
failbool: true
list_in_list_out:
tool_id: collection_creates_list
in:
input1: job_props#list_output
identifier:
tool_id: identifier_collection
in:
input1: job_props/list_output
input1: list_in_list_out#list_output
""")
with self.dataset_populator.test_history() as history_id:
invocation_id = self.__invoke_workflow(history_id, workflow_id)
self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id, assert_ok=False)
failed_dataset_one = self.dataset_populator.get_history_dataset_details(history_id, hid=1, wait=True, assert_ok=False)
assert failed_dataset_one['state'] == 'error', failed_dataset_one
paused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True, assert_ok=False)
assert paused_dataset['state'] == 'paused', paused_dataset
paused_colletion = self.dataset_populator.get_history_collection_details(history_id, hid=7, wait=True, assert_ok=False)
first_paused_element = paused_colletion['elements'][0]['object']
assert first_paused_element['state'] == 'paused', first_paused_element
dependent_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=8, wait=True,
assert_ok=False)
assert dependent_dataset['state'] == 'paused'
inputs = {"thebool": "false",
"failbool": "false",
"rerun_remap_job_id": failed_dataset_one['creating_job']}
@@ -786,9 +802,13 @@ steps:
inputs=inputs,
history_id=history_id,
assert_ok=True)
unpaused_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=5, wait=True,
assert_ok=False)
assert unpaused_dataset['state'] == 'ok'
paused_colletion = self.dataset_populator.get_history_collection_details(history_id, hid=7, wait=True, assert_ok=False)
first_paused_element = paused_colletion['elements'][0]['object']
assert first_paused_element['state'] == 'ok'
self.dataset_populator.wait_for_history(history_id, assert_ok=False)
dependent_dataset = self.dataset_populator.get_history_dataset_details(history_id, hid=8, wait=True, assert_ok=False)
assert dependent_dataset['name'].startswith('identifier_collection')
assert dependent_dataset['state'] == 'ok'
@skip_without_tool("fail_identifier")
@skip_without_tool("identifier_collection")