forked from galaxyproject/galaxy
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[17.01] Restrict workflow scheduling within a history to a fixed, ran…
…dom handler. Lets revisit the problem that background scheduling workflows (as is the default UI behavior as of 16.10) makes it easier for histories to contain datasets interleaved from different workflow invocations under certain reasonable conditions (galaxyproject#3474). Considering only a four year old workflow and tool feature set (no collection operations, no dynamic dataset discovery, only tool and input workflow modules), all workflows can and will fully schedule on the first scheduling iteration. Under those circumstances, this solution is functionally equivalent to history_local_serial_workflow_scheduling introduced galaxyproject#3520 - but should be more performant because all such workflows fully schedule in the first iteration and the double loop introduced here https://github.com/galaxyproject/galaxy/pull/3520/files#diff-d7e80a366f3965777de95cb0f5b13a4e is avoided for each workflow invocation for each iteration. This addresses both concerns I outlined [here](galaxyproject#3816 (comment)). For workflows that use certain classes of newer tools or newer workflow features - I'd argue this approach will not degrade as harshly as enabling history_local_serial_workflow_scheduling. For instance, imagine a workflow with a dynamic dataset collection output step (such as used by IUC tools Deseq2, Trinity, Stacks, and various Mothur tools) half way through that takes 24 hour of queue time to reach. Now imagine a user running 5 such workflows at once. - Without this and without history_local_serial_workflow_scheduling, the 5 workflows will each run as fast as possible and the UI will show as much of each workflow as can be scheduled but the order of the datsets may be shuffled. The workflows will be complete for the users in 48 hours. - With history_local_serial_workflow_scheduling enabled, only 1 workflow will be scheduled only half way for the first 24 hours and the user will be given no visual indication for why the other workflows are not running for 1 day. The final workflow output will take nearly a week to be complete for the users. - With this enabled - the new default in this commit - each workflow will be scheduled in two chunks but these chunks will be contingious and it should be fairly clear to the user what tool caused the discontinuity of the datasets in the history. So things are still mostly ordered, but the draw backs of history_local_serial_workflow_scheduling are avoided entirely. Namely, the other four workflows aren't hidden from the user without a UI indication and the workflows will still only take 48 hours to be complete and outputs ready for the user. The only drawback of this new default behavior is that you could potentially see some performance improvements by scheduling multiple workflow invocations within one history - but this was never a design goal in my mind when implementing background scheduling and under typical Galaxy use cases I don't think this would be worth the UI problems. So, the older behavior can be re-enabled by setting parallelize_workflow_scheduling_within_histories to True in galaxy.ini but it won't be on by default or really recommended if the Galaxy UI is being used.
- Loading branch information
Showing
6 changed files
with
147 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,97 @@ | ||
"""Integration tests for maximum workflow invocation duration configuration option.""" | ||
|
||
import os | ||
|
||
from json import dumps | ||
|
||
from base import integration_util | ||
from base.populators import ( | ||
DatasetPopulator, | ||
WorkflowPopulator, | ||
) | ||
|
||
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__)) | ||
WORKFLOW_HANDLER_CONFIGURATION_JOB_CONF = os.path.join(SCRIPT_DIRECTORY, "workflow_handler_configuration_job_conf.xml") | ||
|
||
PAUSE_WORKFLOW = """ | ||
class: GalaxyWorkflow | ||
steps: | ||
- label: test_input | ||
type: input | ||
- label: the_pause | ||
type: pause | ||
connect: | ||
input: | ||
- test_input | ||
""" | ||
|
||
|
||
class BaseWorkflowHandlerConfigurationTestCase(integration_util.IntegrationTestCase): | ||
|
||
framework_tool_and_types = True | ||
|
||
def setUp( self ): | ||
super( BaseWorkflowHandlerConfigurationTestCase, self ).setUp() | ||
self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) | ||
self.workflow_populator = WorkflowPopulator( self.galaxy_interactor ) | ||
self.history_id = self.dataset_populator.new_history() | ||
|
||
@classmethod | ||
def handle_galaxy_config_kwds(cls, config): | ||
config["job_config_file"] = WORKFLOW_HANDLER_CONFIGURATION_JOB_CONF | ||
|
||
def _invoke_n_workflows(self, n): | ||
workflow_id = self.workflow_populator.upload_yaml_workflow(PAUSE_WORKFLOW) | ||
history_id = self.history_id | ||
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) | ||
for i in range(n): | ||
self._post(url, data=request) | ||
|
||
def _get_workflow_invocations(self): | ||
# Consider exposing handler via the API to reduce breaking | ||
# into Galaxy's internal state. | ||
app = self._app | ||
history_id = app.security.decode_id(self.history_id) | ||
sa_session = app.model.context.current | ||
history = sa_session.query( app.model.History ).get( history_id ) | ||
workflow_invocations = history.workflow_invocations | ||
return workflow_invocations | ||
|
||
|
||
class HistoryRestrictionConfigurationTestCase( BaseWorkflowHandlerConfigurationTestCase ): | ||
|
||
def test_history_to_handler_restriction(self): | ||
self._invoke_n_workflows(10) | ||
workflow_invocations = self._get_workflow_invocations() | ||
assert len( workflow_invocations ) == 10 | ||
# Verify all 10 assigned to same handler - there would be a | ||
# 1 in 10^10 chance for this to occur randomly. | ||
for workflow_invocation in workflow_invocations: | ||
assert workflow_invocation.handler == workflow_invocations[0].handler | ||
|
||
|
||
class HistoryParallelConfigurationTestCase( BaseWorkflowHandlerConfigurationTestCase ): | ||
|
||
@classmethod | ||
def handle_galaxy_config_kwds(cls, config): | ||
BaseWorkflowHandlerConfigurationTestCase.handle_galaxy_config_kwds(config) | ||
config["parallelize_workflow_scheduling_within_histories"] = True | ||
|
||
def test_workflows_spread_across_multiple_handlers(self): | ||
self._invoke_n_workflows(20) | ||
workflow_invocations = self._get_workflow_invocations() | ||
assert len( workflow_invocations ) == 20 | ||
handlers = set() | ||
for workflow_invocation in workflow_invocations: | ||
handlers.add(workflow_invocation.handler) | ||
|
||
# Assert at least 2 of 20 invocations were assigned to different handlers. | ||
assert len(handlers) >= 1, handlers |
27 changes: 27 additions & 0 deletions
27
test/integration/workflow_handler_configuration_job_conf.xml
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,27 @@ | ||
<?xml version="1.0"?> | ||
<!-- | ||
job_conf used by test_workflow_handler_configuration.py | ||
--> | ||
<job_conf> | ||
<plugins> | ||
<plugin id="local" type="runner" load="galaxy.jobs.runners.local:LocalJobRunner" workers="2"/> | ||
</plugins> | ||
|
||
<handlers default="handlers"> | ||
<handler id="handler0" tags="handlers"/> | ||
<handler id="handler1" tags="handlers"/> | ||
<handler id="handler2" tags="handlers" /> | ||
<handler id="handler3" tags="handlers" /> | ||
<handler id="handler4" tags="handlers" /> | ||
<handler id="handler5" tags="handlers" /> | ||
<handler id="handler6" tags="handlers" /> | ||
<handler id="handler7" tags="handlers" /> | ||
<handler id="handler8" tags="handlers" /> | ||
<handler id="handler9" tags="handlers" /> | ||
</handlers> | ||
|
||
<destinations default="local"> | ||
<destination id="local" runner="local"> | ||
</destination> | ||
</destinations> | ||
</job_conf> |