Allow rescheduling of single workflow steps.

Implement maximum_workflow_jobs_per_scheduling_iteration to test this and mitigate memory issues associated with scheduling large workflow steps (maybe).

Fixes #3883.

config/galaxy.ini.sample

@@ -1131,6 +1131,14 @@ use_interactive = True
 # invocation to schedule indefinitely. The default corresponds to 1 month.
 #maximum_workflow_invocation_duration = 2678400
 
+# Specify a maximum number of jobs that any given workflow scheduling iteration can create.
+# Set this to a positive integer to prevent large collection jobs in a workflow from
+# preventing other jobs from executing. This may also mitigate memory issues associated with
+# scheduling workflows at the expense of increased total DB traffic because model objects
+# are expunged from the SQL alchemy session between workflow invocation scheduling iterations.
+# Set to -1 to disable any such maximum (the default).
+#maximum_workflow_jobs_per_scheduling_iteration = -1
+
 # Force serial scheduling of workflows within the context of a particular history
 #history_local_serial_workflow_scheduling=False
 

lib/galaxy/config.py

@@ -388,6 +388,7 @@ class Configuration(object):
         self.history_local_serial_workflow_scheduling = string_as_bool(kwargs.get('history_local_serial_workflow_scheduling', 'False'))
         self.parallelize_workflow_scheduling_within_histories = string_as_bool(kwargs.get('parallelize_workflow_scheduling_within_histories', 'False'))
         self.maximum_workflow_invocation_duration = int(kwargs.get("maximum_workflow_invocation_duration", 2678400))
+        self.maximum_workflow_jobs_per_scheduling_iteration = int(kwargs.get("maximum_workflow_jobs_per_scheduling_iteration", -1))
 
         self.cache_user_job_count = string_as_bool(kwargs.get('cache_user_job_count', False))
         self.pbs_application_server = kwargs.get('pbs_application_server', "")

lib/galaxy/tools/execute.py

@@ -23,8 +23,8 @@ EXECUTION_SUCCESS_MESSAGE = "Tool [%s] created job [%s] %s"
 
 class PartialJobExecution(Exception):
 
-    def __init__(self):
-        pass
+    def __init__(self, execution_tracker):
+        self.execution_tracker = execution_tracker
 
 
 MappingParameters = collections.namedtuple("MappingParameters", ["param_template", "param_combinations"])
@@ -121,7 +121,7 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle
         q.join()
 
     if has_remaining_jobs:
-        raise PartialJobExecution()
+        raise PartialJobExecution(execution_tracker)
     else:
         execution_tracker.finalize_dataset_collections(trans)
 

lib/galaxy/workflow/modules.py

@@ -875,6 +875,7 @@ class ToolModule(WorkflowModule):
         complete = False
         try:
             mapping_params = MappingParameters(tool_state.inputs, param_combinations)
+            max_num_jobs = progress.maximum_jobs_to_schedule_or_none
             execution_tracker = execute(
                 trans=self.trans,
                 tool=tool,
@@ -883,23 +884,24 @@ class ToolModule(WorkflowModule):
                 collection_info=collection_info,
                 workflow_invocation_uuid=invocation.uuid.hex,
                 invocation_step=invocation_step,
+                max_num_jobs=max_num_jobs,
                 job_callback=lambda job: self._handle_post_job_actions(step, job, invocation.replacement_dict),
             )
             complete = True
-        except PartialJobExecution:
-            pass
+        except PartialJobExecution as pje:
+            execution_tracker = pje.execution_tracker
 
         except ToolInputsNotReadyException:
             delayed_why = "tool [%s] inputs are not ready, this special tool requires inputs to be ready" % tool.id
             raise DelayedWorkflowEvaluation(why=delayed_why)
 
-        if invocation_step.is_new:
-            if collection_info:
-                step_outputs = dict(execution_tracker.implicit_collections)
-            else:
-                step_outputs = dict(execution_tracker.output_datasets)
-                step_outputs.update(execution_tracker.output_collections)
-            progress.set_step_outputs(invocation_step, step_outputs)
+        progress.record_executed_job_count(len(execution_tracker.successful_jobs))
+        if collection_info:
+            step_outputs = dict(execution_tracker.implicit_collections)
+        else:
+            step_outputs = dict(execution_tracker.output_datasets)
+            step_outputs.update(execution_tracker.output_collections)
+        progress.set_step_outputs(invocation_step, step_outputs, already_persisted=not invocation_step.is_new)
 
         if execution_tracker.execution_errors:
             message = "Failed to create one or more job(s) for workflow step."

lib/galaxy/workflow/run.py

@@ -140,12 +140,18 @@ class WorkflowInvoker(object):
 
         module_injector = modules.WorkflowModuleInjector(trans)
         if progress is None:
-            progress = WorkflowProgress(self.workflow_invocation, workflow_run_config.inputs, module_injector)
+            progress = WorkflowProgress(
+                self.workflow_invocation,
+                workflow_run_config.inputs,
+                module_injector,
+                jobs_per_scheduling_iteration=getattr(trans.app.config, "maximum_workflow_jobs_per_scheduling_iteration", -1),
+            )
         self.progress = progress
 
     def invoke(self):
         workflow_invocation = self.workflow_invocation
-        maximum_duration = getattr(self.trans.app.config, "maximum_workflow_invocation_duration", -1)
+        config = self.trans.app.config
+        maximum_duration = getattr(config, "maximum_workflow_invocation_duration", -1)
         if maximum_duration > 0 and workflow_invocation.seconds_since_created > maximum_duration:
             log.debug("Workflow invocation [%s] exceeded maximum number of seconds allowed for scheduling [%s], failing." % (workflow_invocation.id, maximum_duration))
             workflow_invocation.state = model.WorkflowInvocation.states.FAILED
@@ -258,16 +264,23 @@ STEP_OUTPUT_DELAYED = object()
 
 class WorkflowProgress(object):
 
-    def __init__(self, workflow_invocation, inputs_by_step_id, module_injector):
+    def __init__(self, workflow_invocation, inputs_by_step_id, module_injector, jobs_per_scheduling_iteration=-1):
         self.outputs = odict()
         self.module_injector = module_injector
         self.workflow_invocation = workflow_invocation
         self.inputs_by_step_id = inputs_by_step_id
-        self.jobs_per_scheduling_iteration = 1
+        self.jobs_per_scheduling_iteration = jobs_per_scheduling_iteration
+        self.jobs_scheduled_this_iteration = 0
 
     @property
-    def maximum_jobs_to_schedule(self):
-        return 1
+    def maximum_jobs_to_schedule_or_none(self):
+        if self.jobs_per_scheduling_iteration > 0:
+            return self.jobs_per_scheduling_iteration - self.jobs_scheduled_this_iteration
+        else:
+            return None
+
+    def record_executed_job_count(self, job_count):
+        self.jobs_scheduled_this_iteration += job_count
 
     def remaining_steps(self):
         # Previously computed and persisted step states.

test/integration/test_maximum_worklfow_invocation_duration.py

@@ -1,48 +0,0 @@
-"""Integration tests for maximum workflow invocation duration configuration option."""
-
-import time
-
-from json import dumps
-
-from base import integration_util
-from base.populators import (
-    DatasetPopulator,
-    WorkflowPopulator,
-)
-
-
-class MaximumWorkflowInvocationDurationTestCase(integration_util.IntegrationTestCase):
-    """Start a Pulsar job."""
-
-    framework_tool_and_types = True
-
-    def setUp(self):
-        super(MaximumWorkflowInvocationDurationTestCase, self).setUp()
-        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
-        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
-
-    @classmethod
-    def handle_galaxy_config_kwds(cls, config):
-        config["maximum_workflow_invocation_duration"] = 20
-
-    def do_test(self):
-        workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause")
-        workflow_id = self.workflow_populator.create_workflow(workflow)
-        history_id = self.dataset_populator.new_history()
-        hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
-        index_map = {
-            '0': dict(src="hda", id=hda1["id"])
-        }
-        request = {}
-        request["history"] = "hist_id=%s" % history_id
-        request["inputs"] = dumps(index_map)
-        request["inputs_by"] = 'step_index'
-        url = "workflows/%s/invocations" % (workflow_id)
-        invocation_response = self._post(url, data=request)
-        invocation_url = url + "/" + invocation_response.json()["id"]
-        time.sleep(5)
-        state = self._get(invocation_url).json()["state"]
-        assert state != "failed", state
-        time.sleep(35)
-        state = self._get(invocation_url).json()["state"]
-        assert state == "failed", state

test/integration/test_workflow_scheduling_options.py

@@ -0,0 +1,92 @@
+"""Integration tests for workflow scheduling configuration option."""
+
+import time
+
+from json import dumps
+
+from base import integration_util
+from base.populators import (
+    DatasetCollectionPopulator,
+    DatasetPopulator,
+    WorkflowPopulator,
+)
+
+
+class MaximumWorkflowInvocationDurationTestCase(integration_util.IntegrationTestCase):
+
+    framework_tool_and_types = True
+
+    def setUp(self):
+        super(MaximumWorkflowInvocationDurationTestCase, self).setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["maximum_workflow_invocation_duration"] = 20
+
+    def do_test(self):
+        workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause")
+        workflow_id = self.workflow_populator.create_workflow(workflow)
+        history_id = self.dataset_populator.new_history()
+        hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
+        index_map = {
+            '0': dict(src="hda", id=hda1["id"])
+        }
+        request = {}
+        request["history"] = "hist_id=%s" % history_id
+        request["inputs"] = dumps(index_map)
+        request["inputs_by"] = 'step_index'
+        url = "workflows/%s/invocations" % (workflow_id)
+        invocation_response = self._post(url, data=request)
+        invocation_url = url + "/" + invocation_response.json()["id"]
+        time.sleep(5)
+        state = self._get(invocation_url).json()["state"]
+        assert state != "failed", state
+        time.sleep(35)
+        state = self._get(invocation_url).json()["state"]
+        assert state == "failed", state
+
+
+class MaximumWorkflowJobsPerSchedulingIterationTestCase(integration_util.IntegrationTestCase):
+
+    framework_tool_and_types = True
+
+    def setUp(self):
+        super(MaximumWorkflowJobsPerSchedulingIterationTestCase, self).setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+        self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["maximum_workflow_jobs_per_scheduling_iteration"] = 1
+
+    def do_test(self):
+        workflow_id = self.workflow_populator.upload_yaml_workflow("""
+class: GalaxyWorkflow
+steps:
+  - type: input_collection
+  - tool_id: collection_creates_pair
+    state:
+      input1:
+        $link: 0
+  - tool_id: collection_paired_test
+    state:
+      f1:
+        $link: 1#paired_output
+  - tool_id: cat_list
+    state:
+      input1:
+        $link: 2#out1
+""")
+        with self.dataset_populator.test_history() as history_id:
+            hdca1 = self.dataset_collection_populator.create_list_in_history(history_id, contents=["a\nb\nc\nd\n", "e\nf\ng\nh\n"]).json()
+            self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+            inputs = {
+                '0': {"src": "hdca", "id": hdca1["id"]},
+            }
+            invocation_id = self.workflow_populator.invoke_workflow(history_id, workflow_id, inputs)
+            self.workflow_populator.wait_for_workflow(history_id, workflow_id, invocation_id)
+            self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+            self.assertEqual("a\nc\nb\nd\ne\ng\nf\nh\n", self.dataset_populator.get_history_dataset_content(history_id, hid=0))