DSQ bench

bench.py

import os

from btasks import add

COUNT = 10000


if os.environ['QUEUE'] == 'rq':
    from rq import Queue
    from redis import Redis

    # Tell RQ what Redis connection to use
    redis_conn = Redis()
    redis_conn.flushdb()
    q = Queue('normal', connection=redis_conn)  # no args implies the default queue
    for i in xrange(COUNT):
        q.enqueue(add, i, i+1)
elif os.environ['QUEUE'] == 'dsq':
    import dsq
    manager = dsq.create_manager('127.0.0.1/0')
    manager.register('add', add)
    if __name__ == '__main__':
        manager.queue.client.flushdb()
        for i in xrange(COUNT):
            manager.push('normal', 'add', args=(i, i+1))
elif os.environ['QUEUE'] == 'celery':
    import celery
    app = celery.Celery('bench', broker='redis://127.0.0.1:6379/0')
    app.task(add)

    if __name__ == '__main__':
        from redis import Redis
        redis_conn = Redis()
        redis_conn.flushdb()
        for i in xrange(COUNT):
            app.send_task('btasks.add', (i, i+1), keep_result=False)

bench.sh

#!/bin/sh

echo === DSQ ===
echo -n Push
time env QUEUE=dsq python bench.py
echo -en '\nProcess'
time env QUEUE=dsq dsq worker -bt bench normal

echo
echo
echo === RQ ===
echo -n Push
time env QUEUE=rq python bench.py
echo -en '\nProcess'
time rq worker -qb normal

echo
echo
echo === Celery ===
echo -n Push
time env QUEUE=celery python bench.py
# Celery does not have burst mode
# echo -en '\nProcess'
# time env QUEUE=celery celery worker -A bench

btasks.py

def add(a, b):
    a + b