app.py

from logging import getLogger
logger = getLogger(__name__)

import gc
import os
os.environ['MP_LOG'] = '1'
import time
import sys
from tqdm import tqdm
from celery import Celery
from celery.canvas import group

celery = Celery(
    broker="amqp://test:test@localhost:5672/test",
    backend="mongodb://localhost/test",
)
celery.conf.update(
    CELERYD_LOG_FORMAT = "[pid=%(process)d processName=%(processName)s - %(asctime)s]: %(name)s - %(levelname)s - %(message)s",

    CELERYD_MAX_TASKS_PER_CHILD = 1,
    CELERYD_WORKER_LOST_WAIT = 60,
    CELERY_WORKER_DIRECT = True,
    CELERY_TASK_RESULT_EXPIRES = 60*60*2, # 2 hour

    CELERY_DISABLE_RATE_LIMITS = True,
    CELERY_MAX_CACHED_RESULTS = 100,
    CELERY_MESSAGE_COMPRESSION = 'gzip',

    CELERY_RESULT_SERIALIZER = 'pickle',
    CELERY_TASK_SERIALIZER = 'pickle',
    CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'msgpack', 'yaml'],

    CELERY_MONGODB_BACKEND_SETTINGS = {
        "max_pool_size": None,
        "options": {
            'j': False,
            'w': 1,
        }
    },
)

@celery.task
def stuff():
    return "ok"

if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == 'test':
        ts = time.time()
        total = int(sys.argv[2]) if len(sys.argv) > 2 else 1000
        for t in tqdm([stuff.apply_async() for i in range(total)], leave=True):
            assert t.get() == 'ok'

        print
        print "%.2f tasks/sec" % (total / (time.time() - ts))
        #import objgraph
        #import IPython
        #IPython.embed()
    else:
        celery.start()

bootstrap.sh

#!/bin/bash -eEx
set -o pipefail
shopt -s xpg_echo

rm -rf .ve
virtualenv .ve

.ve/bin/pip install -I -U tqdm "celery$@"

if ! sudo rabbitmqctl list_vhosts | grep test; then
    #sudo rabbitmqctl delete_vhost test
    sudo rabbitmqctl add_vhost test
    sudo rabbitmqctl add_user test test || echo test user already created
    sudo rabbitmqctl set_permissions -p test test ".*" ".*" ".*"
fi