Cancel Job And All Its Child Tasks

Problem

You want to cancel a job.

Solution

There are two ways to cancel tasks on a job.

Method 1: Set the cancel_on_task_failure property to True to cancel a job’s tasks automatically if one of the tasks fails. Use this option if the result of a job depends on all its child tasks completing successfully.

from agiparallel.client import *
from agiparallel.infrastructure import *
from agiparallel.constants import TaskStatus


class WaitTask(object):
    def execute(self):
        from time import sleep
        for i in range(0, 100):
            sleep(.01)


class FailureTask(object):
    def execute(self):
        from time import sleep
        sleep(.1)
        raise Exception("Failure")


if __name__ == "__main__":
    with ClusterJobScheduler("localhost") as client:
        client.connect()

        job = client.create_job()
        job.cancel_on_task_failure = True

        for i in range(0, 100):
            job.add_task(WaitTask())

        # This task will fail. Watch as the other tasks are canceled.
        job.add_task(FailureTask())

        job.submit()
        job.wait_until_done()

        for task in job.tasks:
            print(f'Task with ID: {task.unique_id} has status {TaskStatus(task.task_status).name}')

Method 2: The other option is to explicitly cancel the job using the cancel() method.

Note

By default, the client will not wait until the task is fully canceled. To wait for the job to complete, use the overload to cancel() and pass True to the listen_for_cancellation_event parameter.

from agiparallel.client import *
from agiparallel.infrastructure import *


class WaitTask(object):
    def execute(self):
        from time import sleep
        for i in range(0, 100):
            sleep(.1)


if __name__ == "__main__":
    with ClusterJobScheduler("localhost") as client:
        client.connect()
        job = client.create_job()

        for i in range(0, 100):
            job.add_task(WaitTask())

        job.submit()

        from time import sleep
        sleep(10)

        job.cancel()
        job.wait_until_done()
        

Discussion

Within a task, check the is_cancelling flag. If is_cancelling is True, try to exit out of the task as soon as possible. Each task is given 30 seconds to gracefully finish before the process is terminated. This value can be configured by setting GRACEFUL_CANCELLATION_TIMEOUT.

See Also

Reference

• cancel()

• cancel_on_task_failure