mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Merge branch 'release_18.09' into release_19.01
This commit is contained in:
@@ -1361,19 +1361,6 @@ galaxy:
|
||||
# Enable the new container interface for Interactive Environments
|
||||
#enable_beta_containers_interface: false
|
||||
|
||||
# Set the following to a number of threads greater than 1 to spawn a
|
||||
# Python task queue for dealing with large tool submissions (either
|
||||
# through the tool form or as part of an individual workflow step
|
||||
# across large collection). This affects workflow scheduling and web
|
||||
# processes, not job handlers. This is a beta option and should not be
|
||||
# used in production.
|
||||
#tool_submission_burst_threads: 1
|
||||
|
||||
# If tool_submission_burst_threads is set to a number greater than 1,
|
||||
# this is the number of jobs to schedule at which the task queue will
|
||||
# be created.
|
||||
#tool_submission_burst_at: 10
|
||||
|
||||
# Enable beta workflow modules that should not yet be considered part
|
||||
# of Galaxy's stable API.
|
||||
#enable_beta_workflow_modules: false
|
||||
|
||||
@@ -2802,33 +2802,6 @@
|
||||
:Type: bool
|
||||
|
||||
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
``tool_submission_burst_threads``
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
:Description:
|
||||
Set the following to a number of threads greater than 1 to spawn a
|
||||
Python task queue for dealing with large tool submissions (either
|
||||
through the tool form or as part of an individual workflow step
|
||||
across large collection). This affects workflow scheduling and web
|
||||
processes, not job handlers. This is a beta option and should not
|
||||
be used in production.
|
||||
:Default: ``1``
|
||||
:Type: int
|
||||
|
||||
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
``tool_submission_burst_at``
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
:Description:
|
||||
If tool_submission_burst_threads is set to a number greater than
|
||||
1, this is the number of jobs to schedule at which the task queue
|
||||
will be created.
|
||||
:Default: ``10``
|
||||
:Type: int
|
||||
|
||||
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
``enable_beta_workflow_modules``
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
|
||||
@@ -368,8 +368,6 @@ class Configuration(object):
|
||||
# Tasked job runner.
|
||||
self.use_tasked_jobs = string_as_bool(kwargs.get('use_tasked_jobs', False))
|
||||
self.local_task_queue_workers = int(kwargs.get("local_task_queue_workers", 2))
|
||||
self.tool_submission_burst_threads = int(kwargs.get('tool_submission_burst_threads', '1'))
|
||||
self.tool_submission_burst_at = int(kwargs.get('tool_submission_burst_at', '10'))
|
||||
|
||||
# Enable new interface for API installations from TS.
|
||||
# Admin menu will list both if enabled.
|
||||
|
||||
@@ -1125,7 +1125,7 @@ class WorkflowContentsManager(UsesAnnotations):
|
||||
if not module.label and module.type in ['data_input', 'data_collection_input']:
|
||||
new_state = safe_loads(state)
|
||||
default_label = new_state.get('name')
|
||||
if util.unicodify(default_label).lower() not in ['input dataset', 'input dataset collection']:
|
||||
if default_label and util.unicodify(default_label).lower() not in ['input dataset', 'input dataset collection']:
|
||||
step.label = module.label = default_label
|
||||
|
||||
|
||||
|
||||
@@ -5,11 +5,9 @@ collections from matched collections.
|
||||
"""
|
||||
import collections
|
||||
import logging
|
||||
from threading import Thread
|
||||
|
||||
import six
|
||||
import six.moves
|
||||
from six.moves.queue import Queue
|
||||
|
||||
from galaxy import model
|
||||
from galaxy.dataset_collections.structure import get_structure, tool_output_to_structure
|
||||
@@ -46,7 +44,6 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle
|
||||
execution_tracker = ToolExecutionTracker(trans, tool, mapping_params, collection_info)
|
||||
else:
|
||||
execution_tracker = WorkflowStepExecutionTracker(trans, tool, mapping_params, collection_info, invocation_step, job_callback=job_callback)
|
||||
app = trans.app
|
||||
execution_cache = ToolExecutionCache(trans)
|
||||
|
||||
def execute_single_job(execution_slice, completed_job):
|
||||
@@ -86,46 +83,17 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle
|
||||
)
|
||||
|
||||
execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template)
|
||||
config = app.config
|
||||
burst_at = getattr(config, 'tool_submission_burst_at', 10)
|
||||
burst_threads = getattr(config, 'tool_submission_burst_threads', 1)
|
||||
|
||||
job_count = len(execution_tracker.param_combinations)
|
||||
|
||||
jobs_executed = 0
|
||||
has_remaining_jobs = False
|
||||
|
||||
if (job_count < burst_at or burst_threads < 2):
|
||||
for i, execution_slice in enumerate(execution_tracker.new_execution_slices()):
|
||||
if max_num_jobs and jobs_executed >= max_num_jobs:
|
||||
has_remaining_jobs = True
|
||||
break
|
||||
else:
|
||||
execute_single_job(execution_slice, completed_jobs[i])
|
||||
else:
|
||||
# TODO: re-record success...
|
||||
q = Queue()
|
||||
|
||||
def worker():
|
||||
while True:
|
||||
params = q.get()
|
||||
execute_single_job(params)
|
||||
q.task_done()
|
||||
|
||||
for i in range(burst_threads):
|
||||
t = Thread(target=worker)
|
||||
t.daemon = True
|
||||
t.start()
|
||||
|
||||
for i, execution_slice in enumerate(execution_tracker.new_execution_slices()):
|
||||
if max_num_jobs and jobs_executed >= max_num_jobs:
|
||||
has_remaining_jobs = True
|
||||
break
|
||||
else:
|
||||
q.put(execution_slice, completed_jobs[i])
|
||||
jobs_executed += 1
|
||||
|
||||
q.join()
|
||||
for i, execution_slice in enumerate(execution_tracker.new_execution_slices()):
|
||||
if max_num_jobs and jobs_executed >= max_num_jobs:
|
||||
has_remaining_jobs = True
|
||||
break
|
||||
else:
|
||||
execute_single_job(execution_slice, completed_jobs[i])
|
||||
|
||||
if has_remaining_jobs:
|
||||
raise PartialJobExecution(execution_tracker)
|
||||
|
||||
@@ -286,8 +286,8 @@ OPTION_ACTIONS = {
|
||||
'trust_ipython_notebook_conversion': _RenameAction("trust_jupyter_notebook_conversion"),
|
||||
'enable_beta_tool_command_isolation': _DeprecatedAndDroppedAction(),
|
||||
'single_user': _ProductionUnsafe(True),
|
||||
'tool_submission_burst_threads': _ProductionPerformance(),
|
||||
'tool_submission_burst_at': _ProductionPerformance(),
|
||||
'tool_submission_burst_threads': _DeprecatedAndDroppedAction(),
|
||||
'tool_submission_burst_at': _DeprecatedAndDroppedAction(),
|
||||
'toolform_upgrade': _DeprecatedAndDroppedAction(),
|
||||
}
|
||||
|
||||
|
||||
@@ -2083,25 +2083,6 @@ mapping:
|
||||
desc: |
|
||||
Enable the new container interface for Interactive Environments
|
||||
|
||||
tool_submission_burst_threads:
|
||||
type: int
|
||||
default: 1
|
||||
required: false
|
||||
desc: |
|
||||
Set the following to a number of threads greater than 1 to spawn
|
||||
a Python task queue for dealing with large tool submissions (either
|
||||
through the tool form or as part of an individual workflow step across
|
||||
large collection). This affects workflow scheduling and web processes,
|
||||
not job handlers. This is a beta option and should not be used in production.
|
||||
|
||||
tool_submission_burst_at:
|
||||
type: int
|
||||
default: 10
|
||||
required: false
|
||||
desc: |
|
||||
If tool_submission_burst_threads is set to a number greater than 1, this
|
||||
is the number of jobs to schedule at which the task queue will be created.
|
||||
|
||||
enable_beta_workflow_modules:
|
||||
type: bool
|
||||
default: false
|
||||
|
||||
Reference in New Issue
Block a user