Instantly share code, notes, and snippets.

Embed
What would you like to do?
# place this file under djapp/management/commands/
from djapp.tasks import *
from datetime import datetime
from django.conf import settings
from optparse import make_option
from django.core.management.base import BaseCommand, CommandError
class Command(BaseCommand):
help = 'Basic load test for Celery backends. Ctrl+C to abort'
def handle(self, *args, **options):
i = 0
start = datetime.now()
try:
while 1:
debug_task.delay()
i += 1
# print i
except KeyboardInterrupt:
stop = datetime.now()
delta = stop - start
conn_per_sec = float(i / (delta.days*24*3600 + delta.seconds))
print
print settings.BROKER_URL
print i, "tasks created for", delta, "seconds"
print "Tasks created per sec:", conn_per_sec
# place this file under djapp/
from __future__ import absolute_import
import os
from celery import Celery
from django.conf import settings
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'djapp.settings')
app = Celery('djapp')
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
@atodorov

This comment has been minimized.

Owner

atodorov commented Nov 5, 2014

celery_load_test.py is a management command (for Django 1.6.X) which creates delayed tasks using Celery and measures how many of them can be created per second. I use it to test performance of various transport backends.

tasks.py defines the sample task that is used for profiling.

To use just run the command and terminate with Ctrl+C:

$ ./manage.py celery_load_test
^C
memory://
6754 tasks created for 0:00:02.179551 seconds
Tasks created per sec: 3377.0
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment