From 1a444e2d84fc18d5c8334e72d3f384a19d66cdca Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 10 Oct 2023 15:36:10 -0400 Subject: [PATCH] Make invocation cancellation two-step process API / users set invocation to cancelling, scheduler then deletes outputs. This avoids race conditions where the cancelled invocation still generates new jobs. --- lib/galaxy/managers/workflows.py | 31 +---------- lib/galaxy/model/__init__.py | 65 +++++++++++++++++++--- lib/galaxy/webapps/galaxy/api/workflows.py | 4 +- lib/galaxy/workflow/run.py | 2 + lib/galaxy/workflow/scheduling_manager.py | 6 ++ lib/galaxy_test/api/test_workflows.py | 10 +++- lib/galaxy_test/base/populators.py | 19 ++++++- 7 files changed, 92 insertions(+), 45 deletions(-) diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py index fda3faf7ac6..6527839cf16 100644 --- a/lib/galaxy/managers/workflows.py +++ b/lib/galaxy/managers/workflows.py @@ -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): diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 28229cae905..fd5eaec21a3 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -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: diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py index c8f9864d172..508b2e48188 100644 --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -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 diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 322a8800b61..59d19a5c079 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -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, diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py index fc44e59b699..32fceaee7cf 100644 --- a/lib/galaxy/workflow/scheduling_manager.py +++ b/lib/galaxy/workflow/scheduling_manager.py @@ -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 diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py index 4b8a418b0c7..7ed5c5ba1a2 100644 --- a/lib/galaxy_test/api/test_workflows.py +++ b/lib/galaxy_test/api/test_workflows.py @@ -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( diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index 4010cff642d..86cdba03f8e 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -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