Pause Celery Task

I'm trying to pause a celery task temporary based on user button click. What I've done is: When a user clicks a button; I release an AJAX request that updates my celery task state

Solution 1:

Faced with the same question and no good answers I came up with solution you might like and it is not dependent on the message queue you are using (aka Redis or RabbitMQ). The key for me was that the update_state method in the celery.app.task.Task class takes task_id as an optional parameter. In my case I am running long running file copy and checksum tasks through multiple worker nodes and sometimes the user wants to pause one running task to reduce performance requirements on the storage to allow other tasks to finish first. I am also running a stateless Flask REST API to initiate the backend tasks and retrieve status of running tasks so I needed a way to have an API call come in to pause and resume the tasks.

Here is my test function which can receive a "message" to pause itself by monitoring it's own state:

celery.task(bind=True)
def long_test(self, i):
    print('long test starting with delay of ' + str(i) + 'seconds on each loop')
    print('task_id =' + str(self.request.id))
    self.update_state(state='PROCESSING')
    count = 0
    while True:
        task = celery.AsyncResult(self.request.id)
        while task.state == 'PAUSING' or task.state == 'PAUSED':
            if task.state == 'PAUSING':
                self.update_state(state='PAUSED')
            time.sleep(i)
        if task.state == 'RESUME':
            self.update_state(state='PROCESSING')
        print('long test loop ' + str(count) + ' ' + str(task.state))
        count += 1
        time.sleep(i)

Then, in order to pause or resume I can do the following:

>>> from project.celeryworker.tasks import long_test
>>> from project import create_app, make_celery
>>> flaskapp = create_app()
>>> celery = make_celery(flaskapp)
>>> from celery.app.task import Task
>>> long_test.apply_async(kwargs={'i': 5})
<AsyncResult: bf19d50f-cf04-47f0-a069-6545fb253887>
>>> Task.update_state(self=celery, task_id='bf19d50f-cf04-47f0-a069-6545fb253887', state='PAUSING')
>>> celery.AsyncResult('bf19d50f-cf04-47f0-a069-6545fb253887').state
'PAUSED'
>>> Task.update_state(self=celery, task_id='bf19d50f-cf04-47f0-a069-6545fb253887', state='RESUME')
>>> celery.AsyncResult('bf19d50f-cf04-47f0-a069-6545fb253887').state
'PROCESSING'
>>> Task.update_state(self=celery, task_id='bf19d50f-cf04-47f0-a069-6545fb253887', state='PAUSING')
>>> celery.AsyncResult('bf19d50f-cf04-47f0-a069-6545fb253887').state
'PAUSED'