break infinite loop in celery

Question

I used celery.chord(...) to create a group of tasks and a method which gets called after all tasks in the group are done.

I use the amqp result backend (but I want to switch to memcached).

My worker prints this line over and over again, every second. I don't know how to break this infinite loop. I have access to the rabbitMQ web interface, but I can't find something with the ID "32ba5fe4-...".

[2013-03-22 14:18:26,896: INFO/MainProcess] Task celery.chord_unlock[32ba5fe4-918c-480f-8a78-a310c11d0c3a] retry: Retry in 1s
[2013-03-22 14:18:26,897: INFO/MainProcess] Got task from broker: celery.chord_unlock[32ba5fe4-918c-480f-8a78-a310c11d0c3a] eta:[2013-03-22 13:18:27.895123+00:00]

This is a testing environment. No data can get lost.

I use Celery 3.0.16

Answer 1

It should not be an infinite loop.

The celery.chord_unlock task checks if the chord subtasks have finished to call the merge callback task. If not it schedules itself to check again in a second. The moment your chord tasks are completed you will no longer see those messages in the log.

EDITED: you can revoke the chord_unlock task to stop the loop

celery.control.revoke('32ba5fe4-918c-480f-8a78-a310c11d0c3a')

Answer 2

For a sanity check, I set the max_retries via a signal at worker startup:

from celery.signals import worker_init

@worker_init.connect
def limit_chord_unlock_tasks(worker, **kwargs):
    """
    Set max_retries for chord.unlock tasks to avoid infinitely looping
    tasks. (see celery/celery#1700 or celery/celery#2725)
    """
    task = worker.app.tasks['celery.chord_unlock']
    if task.max_retries is None:
        retries = getattr(worker.app.conf, 'CHORD_UNLOCK_MAX_RETRIES', None)
        task.max_retries = retries

And then add a CHORD_UNLOCK_MAX_RETRIES variable to my Celery configuration.