celery bug 1118

__init__.py

bug.py

from test_app.tasks import stuff, Bubu
import time
import os

while 1:
    try:
        results = [stuff.apply_async(
            args=(Bubu(os.urandom(200000)),),
            options={
                'queue_name': None,
                'exchange': 'C.dq',
                'routing_key': file('/etc/hostname').read().strip(),
            }
        ) for i in range(50)]
        print results
        time.sleep(2)
        print [res.revoke(terminate=True) for res in results]
        print [res.get(propagate=False) for res in results]
    except Exception:
        import traceback
        traceback.print_exc()
        print 'wait 2 secs ...'
        time.sleep(2)

celeryconfig.py

BROKER_URL = "amqp://aae:aae@localhost:5672/aae"
CELERY_IMPORTS = "test_app.tasks",
CELERY_TASK_SERIALIZER = "pickle"
CELERYD_MAX_TASKS_PER_CHILD = 1
CELERY_WORKER_DIRECT = True
BROKER_POOL_LIMIT = None
CELERY_RESULT_BACKEND = "mongodb"
CELERY_MONGODB_BACKEND_SETTINGS = {}

INSTRUCTIONS.rst

In first terminal:

virtualenv ve
ve/bin/pip install -U -I https://github.com/celery/celery/archive/3.0.zip pymongo
MP_LOG=1 ve/bin/celeryd --events --purge --loglevel=DEBUG --concurrency 5

In second terminal:

ve/bin/python bug.py

How to reproduce:

• run celeryd
• start bug.py
• in 3rd terminal restart rabbitmq-server - if you're lucky enough the node will fail after 2-3 restarts

tasks.py

import time
from celery import task

class Bubu(object):
    def __init__(self, data):
        self.data = data


@task
def stuff(whatev):
    time.sleep(10)