diff --git a/config/galaxy.ini.sample b/config/galaxy.ini.sample index 11d457d892c..1d14bbb025a 100644 --- a/config/galaxy.ini.sample +++ b/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 diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index af2a9aa0142..72fbd663208 100644 --- a/lib/galaxy/config.py +++ b/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', "") diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index 34794d6d338..7e96ce8c914 100644 --- a/lib/galaxy/tools/execute.py +++ b/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) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index ba44feeb5e6..80cec0cbd97 100644 --- a/lib/galaxy/workflow/modules.py +++ b/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." diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 9a3c409243c..a1c76b2bf80 100644 --- a/lib/galaxy/workflow/run.py +++ b/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. diff --git a/test/integration/test_maximum_worklfow_invocation_duration.py b/test/integration/test_maximum_worklfow_invocation_duration.py deleted file mode 100644 index b06661b4448..00000000000 --- a/test/integration/test_maximum_worklfow_invocation_duration.py +++ /dev/null @@ -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 diff --git a/test/integration/test_workflow_scheduling_options.py b/test/integration/test_workflow_scheduling_options.py new file mode 100644 index 00000000000..e1df0ba3417 --- /dev/null +++ b/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))