Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[24.0] More efficient change_state queries, maybe fix deadlock #17632

Merged
merged 4 commits into from
Mar 7, 2024
Merged

[24.0] More efficient change_state queries, maybe fix deadlock #17632

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5a3941e
More efficient change_state queries, maybe fix deadlock
mvdbeek Mar 7, 2024
3aa957a
Capture job in Job.add_output_dataset
mvdbeek Mar 7, 2024
4a57547
Fix job manager unit tests
mvdbeek Mar 7, 2024
b8df73d
Only set dataset.job if not set
mvdbeek Mar 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 14 additions & 28 deletions lib/galaxy/model/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1633,6 +1633,10 @@ def add_input_dataset(self, name, dataset=None, dataset_id=None):

def add_output_dataset(self, name, dataset):
joda = JobToOutputDatasetAssociation(name, dataset)
if dataset.dataset.job is None:
# Only set job if dataset doesn't already have associated job.
# database operation tools that make copies should not modify the job here.
dataset.dataset.job = self
add_object_to_object_session(self, joda)
self.output_datasets.append(joda)

Expand Down Expand Up @@ -1996,24 +2000,8 @@ def update_output_states(self, supports_skip_locked):
SET
state = :state,
update_time = :update_time
WHERE id IN (
SELECT hda.dataset_id FROM history_dataset_association hda
INNER JOIN job_to_output_dataset jtod
ON jtod.dataset_id = hda.id AND jtod.job_id = :job_id
);
"""
),
text(
"""
UPDATE dataset
SET
state = :state,
update_time = :update_time
WHERE id IN (
SELECT ldda.dataset_id FROM library_dataset_dataset_association ldda
INNER JOIN job_to_output_library_dataset jtold
ON jtold.ldda_id = ldda.id AND jtold.job_id = :job_id
);
WHERE
dataset.job_id = :job_id
"""
),
text(
Expand All @@ -2022,11 +2010,10 @@ def update_output_states(self, supports_skip_locked):
SET
info = :info,
update_time = :update_time
WHERE id IN (
SELECT jtod.dataset_id
FROM job_to_output_dataset jtod
WHERE jtod.job_id = :job_id
);
FROM dataset
WHERE
history_dataset_association.dataset_id = dataset.id
AND dataset.job_id = :job_id;
"""
),
text(
Expand All @@ -2035,11 +2022,10 @@ def update_output_states(self, supports_skip_locked):
SET
info = :info,
update_time = :update_time
WHERE id IN (
SELECT jtold.ldda_id
FROM job_to_output_library_dataset jtold
WHERE jtold.job_id = :job_id
);
FROM dataset
WHERE
library_dataset_dataset_association.dataset_id = dataset.id
AND dataset.job_id = :job_id;
"""
),
]
Expand Down
5 changes: 0 additions & 5 deletions lib/galaxy/tools/execute.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,11 +175,6 @@ def execute_single_job(execution_slice, completed_job, skip=False):
history = execution_slice.history or history
jobs_executed += 1

if job_datasets:
for job, datasets in job_datasets.items():
for dataset_instance in datasets:
dataset_instance.dataset.job = job

if execution_slice:
history.add_pending_items()
# Make sure collections, implicit jobs etc are flushed even if there are no precreated output datasets
Expand Down
12 changes: 6 additions & 6 deletions test/unit/app/managers/test_JobConnectionsManager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,10 @@ def job_connections_manager(sa_session) -> JobConnectionsManager:

# =============================================================================
def setup_connected_dataset(sa_session: galaxy_scoped_session):
center_hda = HistoryDatasetAssociation(sa_session=sa_session)
input_hda = HistoryDatasetAssociation(sa_session=sa_session)
center_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
input_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
input_hdca = HistoryDatasetCollectionAssociation()
output_hda = HistoryDatasetAssociation(sa_session=sa_session)
output_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
output_hdca = HistoryDatasetCollectionAssociation()
input_job = Job()
output_job = Job()
Expand Down Expand Up @@ -56,10 +56,10 @@ def setup_connected_dataset(sa_session: galaxy_scoped_session):

def setup_connected_dataset_collection(sa_session: galaxy_scoped_session):
center_hdca = HistoryDatasetCollectionAssociation()
input_hda1 = HistoryDatasetAssociation(sa_session=sa_session)
input_hda2 = HistoryDatasetAssociation(sa_session=sa_session)
input_hda1 = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
input_hda2 = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
input_hdca = HistoryDatasetCollectionAssociation()
output_hda = HistoryDatasetAssociation(sa_session=sa_session)
output_hda = HistoryDatasetAssociation(sa_session=sa_session, create_dataset=True)
output_hdca = HistoryDatasetCollectionAssociation()
input_job = Job()
output_job = Job()
Expand Down
Loading