opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-15 20:06:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

Make invocation cancellation two-step process

Browse Source 
API / users set invocation to cancelling, scheduler then deletes
outputs. This avoids race conditions where the cancelled invocation
still generates new jobs.
This commit is contained in:
mvdbeek
2023-11-13 15:45:27 +01:00
parent 2df6048550
commit 1a444e2d84
7 changed files with 92 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/workflows.py
+1 -30
View File
@@ -31,7 +31,6 @@ from sqlalchemy import (
or_,
select,
true,
update,
)
from sqlalchemy.orm import (
aliased,
@@ -422,45 +421,17 @@ class WorkflowsManager(sharable.SharableModelManager, deletable.DeletableManager
target_format=target_format,
)
def cancel_invocation(self, trans, decoded_invocation_id: int):
def request_invocation_cancellation(self, trans, decoded_invocation_id: int):
workflow_invocation = self.get_invocation(trans, decoded_invocation_id)
cancelled = workflow_invocation.cancel()
if cancelled:
workflow_invocation.add_message(InvocationCancellationUserRequest(reason="user_request"))
trans.sa_session.add(workflow_invocation)
with transaction(trans.sa_session):
trans.sa_session.commit()
job_subq = (
trans.sa_session.query(model.Job.id)
.join(model.WorkflowInvocationStep)
.filter(model.WorkflowInvocationStep.workflow_invocation_id == decoded_invocation_id)
.filter(~model.Job.table.c.state.in_(model.Job.terminal_states))
.with_for_update().scalar_subquery()
)
log.debug("CANCEL STEP JOBS")
trans.sa_session.execute(update(model.Job.table).where(model.Job.id == job_subq).values({"state": model.Job.states.DELETING}))
job_collection_subq = (
trans.sa_session.query(model.Job.id)
.join(model.ImplicitCollectionJobsJobAssociation)
.join(model.ImplicitCollectionJobs)
.join(model.WorkflowInvocationStep, model.WorkflowInvocationStep.implicit_collection_jobs_id == model.ImplicitCollectionJobs.id)
.filter(model.WorkflowInvocationStep.workflow_invocation_id == decoded_invocation_id)
.filter(~model.Job.table.c.state.in_(model.Job.terminal_states))
.with_for_update().subquery()
)
log.debug("CANCEL STEP IMPLICIT JOBS")
trans.sa_session.execute(update(model.Job.table).where(model.Job.table.c.id.in_(job_collection_subq.element)).values({"state": model.Job.states.DELETING}))
with transaction(trans.sa_session):
trans.sa_session.commit()
for invocation in workflow_invocation.subworkflow_invocations:
self.cancel_invocation(trans, invocation.subworkflow_invocation_id)
return workflow_invocation
def get_invocation_step(self, trans, decoded_workflow_invocation_step_id):
lib/galaxy/model/__init__.py
+56 -9
View File
@@ -135,6 +135,7 @@ from galaxy.model.item_attrs import (
from galaxy.model.orm.now import now
from galaxy.model.orm.util import add_object_to_object_session
from galaxy.objectstore import ObjectStore
from galaxy.schema.invocation import InvocationCancellationUserRequest
from galaxy.schema.schema import (
DatasetCollectionPopulatedState,
DatasetState,
@@ -1642,12 +1643,13 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
# Nothing changed, no action needed
return False
session = object_session(self)
if session and self.id and not state in Job.terminal_states:
if session and self.id and state not in Job.terminal_states:
# generate statement that will not revert DELETING or DELETED back to anything non-terminal
rval = session.execute(update(Job.table).where(
Job.table.c.id == self.id,
~Job.table.c.state.in_((Job.states.DELETING, Job.states.DELETED))
).values(state=state))
rval = session.execute(
update(Job.table)
.where(Job.table.c.id == self.id, ~Job.table.c.state.in_((Job.states.DELETING, Job.states.DELETED)))
.values(state=state)
)
if rval.rowcount == 1:
self.state_history.append(JobStateHistory(self))
return True
@@ -8170,6 +8172,7 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
READY = "ready" # Workflow ready for another iteration of scheduling.
SCHEDULED = "scheduled" # Workflow has been scheduled.
CANCELLED = "cancelled"
CANCELLING = "cancelling" # invocation scheduler will cancel job in next iteration
FAILED = "failed"
non_terminal_states = [states.NEW, states.READY]
@@ -8215,11 +8218,54 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
return self.state in [states.NEW, states.READY]
def cancel(self):
if not self.active:
return False
else:
self.state = WorkflowInvocation.states.CANCELLED
if self.state not in [WorkflowInvocation.states.CANCELLING, WorkflowInvocation.states.CANCELLED]:
# No use cancelling workflow again, for all others we may still want to be able to cancel
# remaining tool and workflow steps
self.state = WorkflowInvocation.states.CANCELLING
return True
return False
def cancel_invocation_steps(self):
sa_session = object_session(self)
job_subq = (
sa_session.query(Job.id)
.join(WorkflowInvocationStep)
.filter(WorkflowInvocationStep.workflow_invocation_id == self.id)
.filter(~Job.table.c.state.in_(Job.terminal_states))
.with_for_update()
.scalar_subquery()
)
sa_session.execute(update(Job.table).where(Job.id == job_subq).values({"state": Job.states.DELETING}))
job_collection_subq = (
sa_session.query(Job.id)
.join(ImplicitCollectionJobsJobAssociation)
.join(ImplicitCollectionJobs)
.join(
WorkflowInvocationStep, WorkflowInvocationStep.implicit_collection_jobs_id == ImplicitCollectionJobs.id
)
.filter(WorkflowInvocationStep.workflow_invocation_id == self.id)
.filter(~Job.table.c.state.in_(Job.terminal_states))
.with_for_update()
.subquery()
)
sa_session.execute(
update(Job.table)
.where(Job.table.c.id.in_(job_collection_subq.element))
.values({"state": Job.states.DELETING})
)
for invocation in self.subworkflow_invocations:
subworkflow_invocation = invocation.subworkflow_invocation
cancelled = subworkflow_invocation.cancel()
if cancelled:
subworkflow_invocation.add_message(InvocationCancellationUserRequest(reason="user_request"))
sa_session.add(subworkflow_invocation)
sa_session.commit()
def mark_cancelled(self):
self.state = WorkflowInvocation.states.CANCELLED
def fail(self):
self.state = WorkflowInvocation.states.FAILED
@@ -8269,6 +8315,7 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
or_(
WorkflowInvocation.state == WorkflowInvocation.states.NEW,
WorkflowInvocation.state == WorkflowInvocation.states.READY,
WorkflowInvocation.state == WorkflowInvocation.states.CANCELLING,
),
]
if scheduler is not None:
lib/galaxy/webapps/galaxy/api/workflows.py
+3 -1
View File
@@ -907,7 +907,9 @@ class WorkflowsAPIController(
:raises: exceptions.MessageException, exceptions.ObjectNotFound
"""
decoded_workflow_invocation_id = self.decode_id(invocation_id)
workflow_invocation = self.workflow_manager.cancel_invocation(trans, decoded_workflow_invocation_id)
workflow_invocation = self.workflow_manager.request_invocation_cancellation(
trans, decoded_workflow_invocation_id
)
return self.__encode_invocation(workflow_invocation, **kwd)
@expose_api
lib/galaxy/workflow/run.py
+2
View File
@@ -640,6 +640,8 @@ class WorkflowProgress:
when_values=None,
) -> WorkflowInvoker:
subworkflow_invocation = self._subworkflow_invocation(step)
subworkflow_invocation.handler = self.workflow_invocation.handler
subworkflow_invocation.scheduler = self.workflow_invocation.scheduler
workflow_run_config = workflow_request_to_run_config(subworkflow_invocation, use_cached_job)
subworkflow_progress = self.subworkflow_progress(
subworkflow_invocation,
lib/galaxy/workflow/scheduling_manager.py
+6
View File
@@ -332,6 +332,12 @@ class WorkflowRequestMonitor(Monitors):
workflow_invocation = session.get(model.WorkflowInvocation, invocation_id)
try:
if workflow_invocation.state == workflow_invocation.states.CANCELLING:
workflow_invocation.cancel_invocation_steps()
workflow_invocation.state = workflow_invocation.states.CANCELLED
session.commit()
return False
if not workflow_invocation or not workflow_invocation.active:
return False
lib/galaxy_test/api/test_workflows.py
+8 -2
View File
@@ -4090,12 +4090,18 @@ test_data:
assert invocation_before_cancellation["state"] == "scheduled"
subworkflow_invocation_id = invocation_before_cancellation["steps"][2]["subworkflow_invocation_id"]
self.workflow_populator.cancel_invocation(summary.invocation_id)
self.workflow_populator.wait_for_invocation_and_jobs(
history_id=history_id,
workflow_id=summary.workflow_id,
invocation_id=summary.invocation_id,
assert_ok=False,
)
invocation_jobs = self.workflow_populator.get_invocation_jobs(summary.invocation_id)
for job in invocation_jobs:
assert job["state"] == "deleted" or job["state"] == "deleting"
assert job["state"] == "deleted"
subworkflow_invocation_jobs = self.workflow_populator.get_invocation_jobs(subworkflow_invocation_id)
for job in subworkflow_invocation_jobs:
assert job["state"] == "deleted" or job["state"] == "deleting"
assert job["state"] == "deleted"
def test_workflow_failed_output_not_found(self, history_id):
summary = self._run_workflow(
lib/galaxy_test/base/populators.py
+16 -3
View File
@@ -700,7 +700,9 @@ class BaseDatasetPopulator(BasePopulator):
def active_history_jobs(self, history_id: str) -> list:
all_history_jobs = self.history_jobs(history_id)
active_jobs = [j for j in all_history_jobs if j["state"] in ["new", "upload", "waiting", "queued", "running"]]
active_jobs = [
j for j in all_history_jobs if j["state"] in ["new", "upload", "waiting", "queued", "running", "deleting"]
]
return active_jobs
def cancel_job(self, job_id: str) -> Response:
@@ -2094,7 +2096,7 @@ class BaseWorkflowPopulator(BasePopulator):
def wait_for_invocation_and_jobs(
self, history_id: str, workflow_id: str, invocation_id: str, assert_ok: bool = True
) -> None:
state = self.wait_for_invocation(workflow_id, invocation_id)
state = self.wait_for_invocation(workflow_id, invocation_id, assert_ok=assert_ok)
if assert_ok:
assert state == "scheduled", state
time.sleep(0.5)
@@ -3117,7 +3119,18 @@ def wait_on_state(
return state
if skip_states is None:
skip_states = ["running", "queued", "new", "ready", "stop", "stopped", "setting_metadata", "waiting"]
skip_states = [
"running",
"queued",
"new",
"ready",
"stop",
"stopped",
"setting_metadata",
"waiting",
"cancelling",
"deleting",
]
if ok_states is None:
ok_states = ["ok", "scheduled", "deferred"]
# Remove ok_states from skip_states, so we can wait for a state to becoming running