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

Fix double enqueuing of upload jobs

Browse Source 
The canonical place to call enqueue is in
https://github.com/galaxyproject/galaxy/blob/a5bb3e92667b96ef802b378f17788b253a5c0bc6/lib/galaxy/tools/execute.py#L127.
In the process of enqueue the db-skip-locked assignment method will set
the handler name to the default tag. If you do this more than once the
job might be picked up by compatible handlers.

Fixes https://github.com/galaxyproject/galaxy/issues/11335
This commit is contained in:
mvdbeek
2021-10-22 16:49:00 +02:00
parent 6fcdc5c44d
commit 2a335c981b
12 changed files with 21 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/actions/library.py
+1 -2
View File
@@ -126,8 +126,7 @@ class LibraryActions:
job_params['link_data_only'] = json.dumps(kwd.get('link_data_only', 'copy_files'))
job_params['uuid'] = json.dumps(kwd.get('uuid', None))
job, output = upload_common.create_job(trans, tool_params, tool, json_file_path, data_list, folder=library_bunch.folder, job_params=job_params)
trans.sa_session.add(job)
trans.sa_session.flush()
trans.app.job_manager.enqueue(job, tool=tool)
return output
def _get_server_dir_uploaded_datasets(self, trans, params, full_dir, import_dir_desc, library_bunch, response_code, message):
lib/galaxy/datatypes/data.py
+4 -3
View File
@@ -637,14 +637,15 @@ class Data(metaclass=DataMeta):
params[input_name] = original_dataset
# Run converter, job is dispatched through Queue
converted_dataset = converter.execute(trans, incoming=params, set_output_hid=visible, history=history)[1]
job, converted_datasets, *_ = converter.execute(trans, incoming=params, set_output_hid=visible, history=history)
trans.app.job_manager.enqueue(job, tool=converter)
if len(params) > 0:
trans.log_event("Converter params: %s" % (str(params)), tool_id=converter.id)
if not visible:
for value in converted_dataset.values():
for value in converted_datasets.values():
value.visible = False
if return_output:
return converted_dataset
return converted_datasets
return f"The file conversion of {converter.name} on data {original_dataset.hid} has been added to the Queue."
# We need to clear associated files before we set metadata
lib/galaxy/managers/datasets.py
+4 -2
View File
@@ -374,9 +374,10 @@ class DatasetAssociationManager(base.ModelManager,
if spec.get('default'):
setattr(data.metadata, name, spec.unwrap(spec.get('default')))
self.app.datatypes_registry.set_external_metadata_tool.tool_action.execute(
job, *_ = self.app.datatypes_registry.set_external_metadata_tool.tool_action.execute(
self.app.datatypes_registry.set_external_metadata_tool, trans, incoming={'input1': data, 'validate': validate},
overwrite=overwrite)
self.app.job_manager.enqueue(job, tool=self.app.datatypes_registry.set_external_metadata_tool)
def update_permissions(self, trans, dataset_assoc, **kwd):
action = kwd.get('action', 'set_permissions')
@@ -691,7 +692,8 @@ class DatasetAssociationDeserializer(base.ModelDeserializer, deletable.PurgableD
sa_session = self.app.model.context
sa_session.flush()
trans = context.get("trans")
self.app.datatypes_registry.set_external_metadata_tool.tool_action.execute(self.app.datatypes_registry.set_external_metadata_tool, trans, incoming={'input1': item}, overwrite=False) # overwrite is False as per existing behavior
job, *_ = self.app.datatypes_registry.set_external_metadata_tool.tool_action.execute(self.app.datatypes_registry.set_external_metadata_tool, trans, incoming={'input1': item}, overwrite=False) # overwrite is False as per existing behavior
trans.app.job_manager.enqueue(job, tool=trans.app.datatypes_registry.set_external_metadata_tool)
return item.datatype
lib/galaxy/tools/__init__.py
+2 -1
View File
@@ -2606,10 +2606,11 @@ class SetMetadataTool(Tool):
def regenerate_imported_metadata_if_needed(self, hda, history, job):
if len(hda.metadata_file_types) > 0:
self.tool_action.execute_via_app(
job, *_ = self.tool_action.execute_via_app(
self, self.app, job.session_id,
history.id, job.user, incoming={'input1': hda}, overwrite=False
)
self.app.job_manager.enqueue(job=job, tool=self)
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None, **kwds):
working_directory = app.object_store.get_filename(
lib/galaxy/tools/actions/__init__.py
-3
View File
@@ -600,9 +600,6 @@ class DefaultToolAction:
trans.sa_session.flush()
log.info(f"Flushed transaction for job {job.log_str()} {job_flush_timer}")
# Dispatch to a job handler. enqueue() is responsible for flushing the job
app.job_manager.enqueue(job, tool=tool)
trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
return job, out_data, history
def _remap_job_on_rerun(self, trans, galaxy_session, rerun_remap_job_id, current_job, out_data):
lib/galaxy/tools/actions/history_imp_exp.py
-5
View File
@@ -67,11 +67,6 @@ class ImportHistoryToolAction(ToolAction):
job.add_parameter(name, value)
job.state = start_job_state # job inputs have been configured, restore initial job state
# Queue the job for execution
trans.app.job_manager.enqueue(job, tool=tool)
trans.log_event("Added import history job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
return job, {}
lib/galaxy/tools/actions/metadata.py
-5
View File
@@ -120,11 +120,6 @@ class SetMetadataToolAction(ToolAction):
job.state = start_job_state # job inputs have been configured, restore initial job state
sa_session.flush()
# Queue the job for execution
app.job_manager.enqueue(job, tool=tool)
# FIXME: need to add event logging to app and log events there rather than trans.
# trans.log_event( "Added set external metadata job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id )
# clear e.g. converted files
dataset.datatype.before_setting_metadata(dataset)
lib/galaxy/tools/actions/upload_common.py
+1 -13
View File
@@ -395,6 +395,7 @@ def create_job(trans, params, tool, json_file_path, outputs, folder=None, histor
Create the upload job.
"""
job = trans.app.model.Job()
trans.sa_session.add(job)
job.galaxy_version = trans.app.config.version_major
galaxy_session = trans.get_galaxy_session()
if type(galaxy_session) == trans.model.GalaxySession:
@@ -410,16 +411,10 @@ def create_job(trans, params, tool, json_file_path, outputs, folder=None, histor
job.tool_id = tool.id
job.tool_version = tool.version
job.dynamic_tool = tool.dynamic_tool
job.set_state(job.states.UPLOAD)
trans.sa_session.add(job)
trans.sa_session.flush()
log.info('tool %s created job id %d' % (tool.id, job.id))
trans.log_event('created job id %d' % job.id, tool_id=tool.id)
for name, value in tool.params_to_strings(params, trans.app).items():
job.add_parameter(name, value)
job.add_parameter('paramfile', dumps(json_file_path))
object_store_id = None
for i, output_object in enumerate(outputs):
output_name = "output%i" % i
if hasattr(output_object, "collection"):
@@ -432,18 +427,11 @@ def create_job(trans, params, tool, json_file_path, outputs, folder=None, histor
else:
job.add_output_dataset(output_name, dataset)
trans.sa_session.add(output_object)
job.object_store_id = object_store_id
job.set_state(job.states.NEW)
if job_params:
for name, value in job_params.items():
job.add_parameter(name, value)
trans.sa_session.add(job)
# Queue the job for execution
trans.app.job_manager.enqueue(job, tool=tool)
trans.log_event("Added job to the job queue, id: %s" % str(job.id), tool_id=job.tool_id)
output = {}
for i, v in enumerate(outputs):
if not hasattr(output_object, "collection_type"):
lib/galaxy/webapps/base/controller.py
+4 -2
View File
@@ -436,7 +436,8 @@ class ExportsHistoryMixin:
# Run job to do export.
history_exp_tool = trans.app.toolbox.get_tool(export_tool_id)
job, _ = history_exp_tool.execute(trans, incoming=params, history=history, set_output_hid=True)
job, *_ = history_exp_tool.execute(trans, incoming=params, history=history, set_output_hid=True)
trans.app.job_manager.enqueue(job, tool=history_exp_tool)
return job
@@ -446,7 +447,8 @@ class ImportsHistoryMixin:
# Run job to do import.
history_imp_tool = trans.app.toolbox.get_tool('__IMPORT_HISTORY__')
incoming = {'__ARCHIVE_SOURCE__': archive_source, '__ARCHIVE_TYPE__': archive_type}
job, _ = history_imp_tool.execute(trans, incoming=incoming)
job, *_ = history_imp_tool.execute(trans, incoming=incoming)
trans.app.job_manager.enqueue(job, tool=history_imp_tool)
return job
lib/galaxy/webapps/galaxy/api/library_datasets.py
+1 -2
View File
@@ -519,8 +519,7 @@ class LibraryDatasetsController(BaseGalaxyAPIController, UsesVisualizationMixin,
job_params['link_data_only'] = dumps(kwd.get('link_data_only', 'copy_files'))
job_params['uuid'] = dumps(kwd.get('uuid', None))
job, output = upload_common.create_job(trans, tool_params, tool, json_file_path, data_list, folder=folder, job_params=job_params)
trans.sa_session.add(job)
trans.sa_session.flush()
trans.app.job_manager.enqueue(job, tool=tool)
job_dict = job.to_dict()
job_dict['id'] = trans.security.encode_id(job_dict['id'])
return job_dict
lib/galaxy/webapps/galaxy/controllers/async.py
+2 -1
View File
@@ -89,7 +89,8 @@ class ASync(BaseUIController):
raise Exception("Error: ToolOutput object not found")
original_history = trans.sa_session.query(trans.app.model.History).get(data.history_id)
tool.execute(trans, incoming=params, history=original_history)
job, *_ = tool.execute(trans, incoming=params, history=original_history)
trans.app.job_manager.enqueue(job, tool=tool)
else:
log.debug('async error -> %s' % STATUS)
trans.log_event('Async error -> %s' % STATUS)
lib/galaxy/webapps/galaxy/controllers/dataset.py
+2 -1
View File
@@ -377,9 +377,10 @@ class DatasetInterface(BaseUIController, UsesAnnotations, UsesItemRatings, UsesE
datatype = sniff.guess_ext(path, trans.app.datatypes_registry.sniff_order, is_binary=is_binary)
trans.app.datatypes_registry.change_datatype(data, datatype)
trans.sa_session.flush()
trans.app.datatypes_registry.set_external_metadata_tool.tool_action.execute(
job, *_ = trans.app.datatypes_registry.set_external_metadata_tool.tool_action.execute(
trans.app.datatypes_registry.set_external_metadata_tool, trans, incoming={'input1': data},
overwrite=False) # overwrite is False as per existing behavior
trans.app.job_manager.enqueue(job, tool=trans.app.datatypes_registry.set_external_metadata_tool)
message = 'Detection was finished and changed the datatype to %s.' % datatype
else:
return self.message_exception(trans, 'Changing datatype "%s" is not allowed.' % (data.extension))