Create dynamic queues with Celery

21
votes

Here's my scenario:

When a user logs in to my website, I queue up a bunch of tasks for the given user (typically each task takes 100s of msecs and there are 100s of tasks per user). These tasks are queued to the default Celery Queue and I have 100s of workers running. I use websockets to show the user real-time progress as the tasks complete on the backend. Life is good if I have just 1 or 2 users active.

Now if I a few concurrent users log-in to my site, the latter users are queued behind the initial users and their tasks starve (since all the tasks go to the same queue). My thoughts are to create a dynamic queue per user to ensure fairness. However as per Celery documentation (http://docs.celeryproject.org/en/latest/userguide/routing.html#defining-queues), seems I need to be define the queues statically.

Any suggestions on best practices for using celery for my scenario?

2
pythonqueueflaskcelery
Wouldn't having dynamic queues defeat the object of queing? 100's of tasks sounds like a lot per user. Can these not be be grouped together so a worker would pull all the users tasks of at once? That way the user would have a dedicated worker for the login process. Does all the login tasks need to run in sync? – Joe Doherty
@JoeDoherty - I don't get your point as to how dynamic queues would defeat the object of queueing. However your suggestion of batching all per-user tasks as one big task is very interesting and would definitely alleviate my issue - thanks!. From a design point of view though, I was trying to go for a big pool of workers executing small/quick tasks (versus long running tasks). I'd still be interested to see if there is an option to create queues on the fly though - I can see this being useful even if I batch the user tasks. – El Diablo
@JoeDoherty - Regarding 'Does all the login tasks need to run in sync' - the answer is 'No'. The requirement is that they all complete in a reasonable time since the user is waiting for the results of the tasks to be complete on the website. – El Diablo
I forgot the user was waiting for each to complete. And also I was thinking incorrectly about the dynamic queues that would solve your problem. You could batch them all together and then report back as each stage complete similar to this – Joe Doherty

2 Answers

5
votes

http://docs.celeryproject.org/en/latest/userguide/workers.html#queues

celery -A proj control add_consumer foo -d worker1.local

The same can be accomplished dynamically using the app.control.add_consumer() method:

app.control.add_consumer('foo', reply=True)
[{u'worker1.local': {u'ok': u"already consuming from u'foo'"}}]

app.control.add_consumer('foo', reply=True,
destination=['[email protected]'])
2
votes

You can dynamically assign a task to a Queue at runtime when invoking it refer calling.html#routing-options. This would work if you have CELERY_CREATE_MISSING_QUEUES enabled.