Description
Covalent now supports dispatch cancellation, as a result all executor plugins can opt-in to this functionality by exposing a cancel method, periodically checking if cancellation has been requested and registering a job handle (job ID) associated with a task (in this case the function name or function ARN).
Essentially only the following methods are defined in the parent classes of the executors need to be consumed within the executor class:
await self.get_cancel_requested() in order to determine if task cancellation has been requested, at which point it is appropriate to raise a TaskCancelledError exceptionawait self.set_job_handle(handle=job_handle) should be set once the job handle is known- expose a
async def cancel(self, task_metadata: Dict, job_handle: str) method in the executor class
The below code can be used as a reference:
from covalent._shared_files.exceptions import TaskCancelledError
...
async def proceed_if_task_not_cancelled(self):
if await self.get_cancel_requested():
self._debug_log(f"Task Cancelled")
raise TaskCancelledError(f"Batch job {batch_job_name} requested to be cancelled")
async def run(self, function: Callable, args: List, kwargs: Dict, task_metadata: Dict) -> Any:
...
await self.proceed_if_task_not_cancelled()
# pickle task
...
await self.proceed_if_task_not_cancelled()
# upload pickled assets
...
await self.proceed_if_task_not_cancelled()
# invoke job/task
await self.set_job_handle(handle=job_handle)
async def cancel(self, task_metadata: Dict, job_handle: str) -> None:
"""
Cancel the batch job
Arg(s)
task_metadata: Dictionary with the task's dispatch_id and node id
job_handle: Unique job handle assigned to the task by Batch
Return(s)
None
"""
# boto client invocations to cancel the taskAcceptance Criteria
Description
Covalent now supports dispatch cancellation, as a result all executor plugins can opt-in to this functionality by exposing a
cancelmethod, periodically checking if cancellation has been requested and registering ajob handle(job ID) associated with a task (in this case the function name or function ARN).Essentially only the following methods are defined in the parent classes of the executors need to be consumed within the executor class:
await self.get_cancel_requested()in order to determine if task cancellation has been requested, at which point it is appropriate to raise aTaskCancelledErrorexceptionawait self.set_job_handle(handle=job_handle)should be set once thejob handleis knownasync def cancel(self, task_metadata: Dict, job_handle: str)method in the executor classThe below code can be used as a reference:
Acceptance Criteria
job handleonce job / task id is determinedcancelfunctionality correctly integratedcancelmethod (may not be possible with lambda so the above tasks suffice)