rabbitmq - Stopping/Purging Periodic Tasks in Django-Celery

Question

I have managed to get periodic tasks working in django-celery by subclassing PeriodicTask. I tried to create a test task and set it running doing something useless. It works.

Now I can't stop it. I've read the documentation and I cannot find out how to remove the task from the execution queue. I have tried using celeryctl and using the shell, but registry.tasks() is empty, so I can't see how to remove it.

I have seen suggestions that I should "revoke" it, but for this I appear to need a task id, and I can't see how I would find the task id.

Thanks.

Answer 1

A task is a message, and a "periodic task" sends task messages at periodic intervals. Each of the tasks sent will have an unique id assigned to it.

revoke will only cancel a single task message. To get the id for a task you have to keep track of the id sent, but you can also specify a custom id when you send a task.

I'm not sure if you want to cancel a single task message, or if you want to stop the periodic task from sending more messages, so I'll list answers for both.

There is no built-in way to keep the id of a task sent with periodic tasks, but you could set the id for each task to the name of the periodic task, that way the id will refer to any task sent with the periodic task (usually the last one). You can specify a custom id this way,

either with the @periodic_task decorator:

@periodic_task(options={"task_id": "my_periodic_task"})
def my_periodic_task():
    pass

or with the CELERYBEAT_SCHEDULE setting:

CELERYBEAT_SCHEDULE = {name: {"task": task_name,
                              "options": {"task_id": name}}}

If you want to remove a periodic task you simply remove the @periodic_task from the codebase, or remove the entry from CELERYBEAT_SCHEDULE. If you are using the Django database scheduler you have to remove the periodic task from the Django Admin interface.

PS1: revoke doesn't stop a task that has already been started. It only cancels tasks that haven't been started yet. You can terminate a running task using revoke(task_id, terminate=True). By default this will send the TERM signal to the process, if you want to send another signal (e.g. KILL) use revoke(task_id, terminate=True, signal="KILL").

PS2: revoke is a remote control command so it is only supported by the RabbitMQ and Redis broker transports. If you want your task to support cancellation you should do so by storing a cancelled flag in a database and have the task check that flag when it starts:

from celery.task import Task

class RevokeableTask(Task):
    """Task that can be revoked.

    Example usage:

        @task(base=RevokeableTask)
        def mytask():
            pass
    """

    def __call__(self, *args, **kwargs):
        if revoke_flag_set_in_db_for(self.request.id):
            return
        super(RevokeableTask, self).__call__(*args, **kwargs)